Spaces:
Sleeping
Sleeping
File size: 4,781 Bytes
e0a3391 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """
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)]
|