lib/schema.py

"""
lib/schema.py

Defensive accessors over the raw candidate dict (candidate_schema.json).
Every field is technically required, but we never trust that in practice --
a missing/odd field should degrade a candidate's signal quietly, not crash.
Every getter has a safe default.
"""
from __future__ import annotations
import datetime as dt
from typing import Any


def _parse_date(s: Any) -> dt.date | None:
    if not s or not isinstance(s, str):
        return None
    try:
        return dt.date.fromisoformat(s)
    except ValueError:
        return None


def profile(c: dict) -> dict:
    return c.get("profile") or {}


def signals(c: dict) -> dict:
    return c.get("redrob_signals") or {}


def career_history(c: dict) -> list[dict]:
    """Returns career_history sorted most-recent-first by start_date.

    M2 fix: result is cached on the dict itself. career_history() is called
    6+ times per candidate across features/honeypot/scoring; re-sorting and
    re-parsing dates each time adds ~3.5M redundant date parses at 100K
    scale. Mutation is safe here because we own the loop in precompute.py.
    """
    if "_career_sorted" not in c:
        ch = c.get("career_history") or []
        c["_career_sorted"] = sorted(
            ch,
            key=lambda r: _parse_date(r.get("start_date")) or dt.date.min,
            reverse=True,
        )
    return c["_career_sorted"]


def skills(c: dict) -> list[dict]:
    return c.get("skills") or []


def education(c: dict) -> list[dict]:
    return c.get("education") or []


def years_of_experience(c: dict) -> float:
    try:
        return float(profile(c).get("years_of_experience", 0) or 0)
    except (TypeError, ValueError):
        return 0.0


def current_title(c: dict) -> str:
    return (profile(c).get("current_title") or "").strip()


def current_company(c: dict) -> str:
    return (profile(c).get("current_company") or "").strip()


def unified_text_blob(c: dict) -> str:
    """
    Concatenated free text where skills show up *in context*
    (title, headline, summary, every role's title + description).
    This is the surface the JD-fit and production-evidence scanners run
    against. Deliberately excludes the raw `skills[]` list -- that list is
    scored separately and downweighted, because it is the easiest field to
    stuff with irrelevant buzzwords (confirmed in the real 100K pool: 365
    "Marketing Manager" profiles list rag/pinecone/embeddings as skills).
    """
    p = profile(c)
    parts = [
        p.get("headline") or "",
        p.get("current_title") or "",
        p.get("summary") or "",
    ]
    for role in career_history(c):
        parts.append(role.get("title") or "")
        parts.append(role.get("description") or "")
    return " \n ".join(parts).lower()


def listed_skill_names(c: dict) -> list[str]:
    return [s.get("name", "").lower() for s in skills(c) if s.get("name")]


def parse_date(s: Any) -> dt.date | None:
    return _parse_date(s)