""" 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)