HR Architect Coding Problems

from collections import defaultdict

class UnionFind:
    def __init__(self):
        self.parent = {}
    def find(self, x):
        self.parent.setdefault(x, x)
        while self.parent[x] != x:
            self.parent[x] = self.parent[self.parent[x]]
            x = self.parent[x]
        return x
    def union(self, a, b):
        ra, rb = self.find(a), self.find(b)
        if ra != rb:
            self.parent[ra] = rb

def dedup_workers(*sources: list[dict]) -> list[dict]:
    uf = UnionFind()
    record_by_email: dict[str, list[dict]] = defaultdict(list)
    # Step 1: normalize and index
    for source in sources:
        for w in source:
            for raw_email in (w.get("emails") or [w.get("email")]):
                if not raw_email:
                    continue
                e = raw_email.strip().lower()
                record_by_email[e].append(w)
                uf.union(e, e)  # ensures presence
    # Step 2: connect any two emails that appear on the same record
    for source in sources:
        for w in source:
            emails = [e.strip().lower() for e in (w.get("emails") or [w["email"]]) if e]
            for e in emails[1:]:
                uf.union(emails[0], e)
    # Step 3: group by root
    groups: dict[str, list[dict]] = defaultdict(list)
    for email, records in record_by_email.items():
        groups[uf.find(email)].extend(records)
    # Step 4: collapse each group deterministically (prefer Workday source)
    result = []
    for root, records in groups.items():
        records.sort(key=lambda r: (r.get("source") != "workday", r.get("worker_id") or ""))
        canonical = records[0]
        canonical["aliases"] = sorted({r.get("source") for r in records})
        result.append(canonical)
    return result

from datetime import date

def headcount_on(workers: list[dict], as_of: date, term_inclusive: bool = True) -> int:
    """Count workers active on `as_of`.

    term_inclusive=True means termination_date == as_of still counts as active
    (the standard convention: last day worked is paid).
    """
    n = 0
    for w in workers:
        hd = w["hire_date"]
        td = w.get("termination_date")
        if hd > as_of:
            continue
        if td is None:
            n += 1
        elif term_inclusive and td >= as_of:
            n += 1
        elif not term_inclusive and td > as_of:
            n += 1
    return n

from datetime import date

def attrition_rate(workers: list[dict], year: int) -> dict:
    start = date(year, 1, 1)
    end = date(year, 12, 31)
    start_hc = headcount_on(workers, start)
    end_hc = headcount_on(workers, end)
    terms = sum(
        1 for w in workers
        if w.get("termination_date")
        and start <= w["termination_date"] <= end
    )
    avg_hc = (start_hc + end_hc) / 2
    return {
        "terminations": terms,
        "start_headcount": start_hc,
        "end_headcount": end_hc,
        "avg_headcount": avg_hc,
        "attrition_rate": terms / avg_hc if avg_hc else None,
    }

from datetime import date, timedelta
from calendar import monthrange

def is_business_day(d: date) -> bool:
    return d.weekday() < 5  # 0=Mon, 4=Fri (no holiday calendar here; in prod, use one)

def shift_business_days(d: date, days_back: int) -> date:
    """Return the date that is `days_back` business days before d."""
    cur = d
    n = days_back
    while n > 0:
        cur -= timedelta(days=1)
        if is_business_day(cur):
            n -= 1
    return cur

def payroll_window(today: date, effective: date,
                   country: str = "US",
                   lock_business_days_before_eom: int = 5) -> dict:
    """For a monthly payroll cycle with cutoff `lock_business_days_before_eom`
    business days before the end of the month."""
    target_month = effective.month
    target_year = effective.year
    last_day = date(target_year, target_month, monthrange(target_year, target_month)[1])
    cutoff = shift_business_days(last_day, lock_business_days_before_eom)
    locked = today > cutoff
    cycle = f"{target_year}-{target_month:02d}"
    return {
        "cycle": cycle,
        "cutoff": cutoff,
        "locked": locked,
        "days_to_cutoff": (cutoff - today).days,
    }

import csv, io
from datetime import date

EIB_COLUMNS = [
    "Employee_ID", "Effective_Date", "New_Position_ID",
    "New_Job_Profile", "New_Supervisor_ID", "New_Location",
    "New_FTE", "Reason_Code", "Business_Process_Comment",
]
REQUIRED = {"Employee_ID", "Effective_Date", "Reason_Code"}

def to_eib_csv(proposals: list[dict], run_id: str) -> str:
    buf = io.StringIO()
    writer = csv.DictWriter(buf, fieldnames=EIB_COLUMNS, extrasaction="ignore")
    writer.writeheader()
    for p in proposals:
        row = {
            "Employee_ID": p["employee_id"],
            "Effective_Date": p["effective_date"].isoformat()
                if isinstance(p["effective_date"], date) else p["effective_date"],
            "New_Position_ID": p.get("new_position_id", ""),
            "New_Job_Profile": p.get("new_job_profile", ""),
            "New_Supervisor_ID": p.get("new_supervisor_id", ""),
            "New_Location": p.get("new_location", ""),
            "New_FTE": f"{p['new_fte']:.4f}" if p.get("new_fte") is not None else "",
            "Reason_Code": p["reason_code"],
            "Business_Process_Comment": f"{p.get('rationale','')} [run:{run_id} proposal:{p['proposal_id']}]",
        }
        missing = [k for k in REQUIRED if not row[k]]
        if missing:
            raise ValueError(f"Missing required EIB fields {missing} for {p}")
        writer.writerow(row)
    return buf.getvalue()

import hashlib

def hire_idempotency_key(applicant_id: str,
                         hire_date: str,
                         position_id: str,
                         intent_window_days: int = 7) -> str:
    """A hire proposal is the same logical event when:
       - same applicant
       - same intended position
       - hire_date within `intent_window_days` of original

    Implementation: bucket the hire_date into `intent_window_days` periods so
    near-equal dates collide but distant dates don't.
    """
    from datetime import date
    hd = date.fromisoformat(hire_date)
    bucket = hd.toordinal() // intent_window_days
    raw = f"hire:{applicant_id}:{position_id}:bucket={bucket}"
    return hashlib.sha256(raw.encode()).hexdigest()[:24]

from datetime import date
from calendar import monthrange

def accrued_pto(hire_date: date, as_of: date,
                rate_first_year: float = 1.25,
                rate_after: float = 1.5,
                cap_days: float = 30.0) -> float:
    if as_of < hire_date:
        return 0.0
    total = 0.0
    # First (possibly partial) month
    first_month_days = monthrange(hire_date.year, hire_date.month)[1]
    days_in_first = first_month_days - hire_date.day + 1
    fraction = days_in_first / first_month_days
    rate = rate_first_year  # they're definitely < 1y at hire
    total += rate * fraction

    cur = date(hire_date.year, hire_date.month, 1)
    # Advance to next month
    cur = (date(cur.year + (cur.month // 12), (cur.month % 12) + 1, 1))
    while cur <= as_of:
        # Is this month < 12 months of tenure?
        anniv = date(hire_date.year + 1, hire_date.month, hire_date.day)
        rate = rate_first_year if cur < anniv else rate_after
        # Was this month complete by as_of?
        last_day = date(cur.year, cur.month, monthrange(cur.year, cur.month)[1])
        if as_of >= last_day:
            total += rate
        else:
            partial = (as_of.day) / monthrange(cur.year, cur.month)[1]
            total += rate * partial
        # Next month
        cur = date(cur.year + (cur.month // 12), (cur.month % 12) + 1, 1)
    return min(total, cap_days)

from collections import deque

def descendants(children: dict[str, list[str]], root: str) -> set[str]:
    """All workers under `root` inclusive, BFS."""
    seen = {root}
    q = deque([root])
    while q:
        cur = q.popleft()
        for c in children.get(cur, []):
            if c not in seen:
                seen.add(c)
                q.append(c)
    return seen

def affected_by_reorg(children: dict[str, list[str]],
                      moving_worker: str) -> set[str]:
    """Every worker whose skip-level chain changed when `moving_worker` (and
    everyone reporting under them) moves. That's the subtree rooted at the
    moving worker."""
    return descendants(children, moving_worker)

from collections import defaultdict

def detect_leave_overlaps(leaves: list[dict]) -> list[tuple[dict, dict]]:
    by_worker = defaultdict(list)
    for l in leaves:
        by_worker[l["worker_id"]].append(l)
    overlaps = []
    for worker_id, lst in by_worker.items():
        lst.sort(key=lambda x: (x["start"], x["end"]))
        for i in range(len(lst)):
            for j in range(i + 1, len(lst)):
                # Sorted by start, so once a start passes lst[i]'s end, no later one overlaps it
                if lst[j]["start"] > lst[i]["end"]:
                    break
                overlaps.append((lst[i], lst[j]))
    return overlaps

from datetime import date

def active_days_in_range(hire: date, term: date | None,
                         range_start: date, range_end: date) -> int:
    if range_end < range_start:
        return 0
    start = max(hire, range_start)
    end = min(term, range_end) if term else range_end
    if end < start:
        return 0
    return (end - start).days + 1  # inclusive both ends