""" Stream-parse candidates.jsonl into compact, normalized records. Handles both plain .jsonl and gzipped .jsonl.gz. Designed to be memory-safe (line-by-line) and fast - no third-party deps. The normalized record exposes everything the trap gate, scorer and reasoning need, with the raw profile kept for grounded reasoning / fact-checking. """ from __future__ import annotations import gzip import io import json from datetime import date from typing import Dict, Iterator, List def _open(path: str): if str(path).endswith(".gz"): return io.TextIOWrapper(gzip.open(path, "rb"), encoding="utf-8") return open(path, "r", encoding="utf-8") def iter_raw(path: str) -> Iterator[dict]: """Yield raw candidate dicts one per line.""" with _open(path) as f: for line in f: line = line.strip() if line: yield json.loads(line) def _pdate(s): try: return date.fromisoformat(s) except (TypeError, ValueError): return None def build_narrative(raw: dict) -> str: """Career-narrative document for retrieval: headline + summary + role descriptions + skill names. This is what BM25 and the embedder see.""" p = raw.get("profile", {}) parts: List[str] = [p.get("headline", ""), p.get("summary", "")] for h in raw.get("career_history", []): parts.append(h.get("title", "")) parts.append(h.get("description", "")) parts.append(" ".join(s.get("name", "") for s in raw.get("skills", []))) return " ".join(x for x in parts if x).strip() def normalize(raw: dict, reference_date: date) -> dict: """Return a normalized record with derived fields used downstream.""" p = raw.get("profile", {}) sig = raw.get("redrob_signals", {}) assess = sig.get("skill_assessment_scores", {}) or {} assess_lower = {k.lower(): v for k, v in assess.items()} skills = [] for s in raw.get("skills", []): name = s.get("name", "") skills.append({ "name": name, "name_lower": name.lower(), "proficiency": s.get("proficiency", ""), "months": s.get("duration_months", 0) or 0, "endorsements": s.get("endorsements", 0) or 0, "assessment": assess_lower.get(name.lower()), }) career = [] companies = [] for h in raw.get("career_history", []): sd, ed = _pdate(h.get("start_date")), _pdate(h.get("end_date")) career.append({ "company": h.get("company", ""), "company_lower": h.get("company", "").lower(), "title": h.get("title", ""), "title_lower": h.get("title", "").lower(), "start": sd, "end": ed, "months": h.get("duration_months", 0) or 0, "is_current": bool(h.get("is_current")), "industry": h.get("industry", ""), "company_size": h.get("company_size", ""), "description": h.get("description", ""), "description_lower": h.get("description", "").lower(), }) companies.append(h.get("company", "").lower()) narrative = build_narrative(raw) rec = { "candidate_id": raw.get("candidate_id", ""), "name": p.get("anonymized_name", ""), "headline": p.get("headline", ""), "summary": p.get("summary", ""), "title": p.get("current_title", ""), "title_lower": p.get("current_title", "").lower(), "company": p.get("current_company", ""), "company_size": p.get("current_company_size", ""), "industry": p.get("current_industry", ""), "location": p.get("location", ""), "location_lower": p.get("location", "").lower(), "country": p.get("country", ""), "yoe": float(p.get("years_of_experience", 0) or 0), "skills": skills, "career": career, "companies": companies, "education": raw.get("education", []), "signals": sig, "narrative": narrative, "narrative_lower": narrative.lower(), # raw kept for grounded reasoning / hallucination checks "_raw": raw, } # --- derived career aggregates --- rec["total_career_months"] = sum(c["months"] for c in career) rec["num_jobs"] = len(career) completed = [c for c in career if not c["is_current"] and c["months"] > 0] rec["avg_tenure_months"] = ( sum(c["months"] for c in completed) / len(completed) if completed else 0.0 ) # recency of activity in days la = _pdate(sig.get("last_active_date")) rec["days_since_active"] = (reference_date - la).days if la else None return rec def load_all(path: str, reference_date: date) -> List[dict]: return [normalize(r, reference_date) for r in iter_raw(path)]