| """Layer 1 — FactEngine: the grounded, JD-relative fact bundle for one finalist. |
| |
| Every downstream clause traces back to a key here; nothing in prose is invented. Extends the old fact set with |
| the Appendix-C1 gaps (missing must-haves, nice-to-have hits, 3-tier location, country, exclude-company, certs, |
| notice/recency bands), the behavioral ``signals`` sub-dict, and a dominant-signal pick widened to all score legs. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| from jd import JDSpec |
|
|
| from . import bands |
| from .banks import SIGNAL_PHRASES, norm |
| from .signals import signal_bands |
|
|
| |
| SIGNAL_COLUMNS: tuple[str, ...] = ( |
| "reranker_score", "granite_score", "composite_score", "candidate_quality", |
| "potion_score", "soft_score", "bm25_score", |
| ) |
| BM25_FIELDS: tuple[str, ...] = ("skills", "titles", "education", "locations", "industries") |
|
|
|
|
| def _overlap(jd_terms: list[str], cand_terms: list[str]) -> list[str]: |
| """JD terms the candidate has (case-insensitive substring either way), in JD order, deduped.""" |
| cand = [norm(t) for t in cand_terms if str(t).strip()] |
| hits: list[str] = [] |
| seen: set[str] = set() |
| for jd in jd_terms: |
| key = norm(jd) |
| if key and key not in seen and any(key in c or c in key for c in cand): |
| hits.append(jd) |
| seen.add(key) |
| return hits |
|
|
|
|
| def matched_skills(jd_skills: list[str], cand_skills: list[str]) -> list[str]: |
| return _overlap(jd_skills, cand_skills) |
|
|
|
|
| def matched_industries(jd_industries: list[str], current_industry: str, career_industries: list[str]) -> list[str]: |
| return _overlap(jd_industries, [current_industry, *career_industries]) |
|
|
|
|
| def matched_education(jd_fields: list[str], education: list[dict[str, Any]]) -> list[str]: |
| fields = [str(degree.get("field_of_study") or "") for degree in (education or [])] |
| return _overlap(jd_fields, fields) |
|
|
|
|
| def title_fit(jd_titles: list[str], current_title: str, has_skill_match: bool) -> tuple[str | None, str | None]: |
| """Return (matched_jd_title, band). band ∈ {align, adjacent, mismatch, None}.""" |
| title = norm(current_title) |
| if not title: |
| return None, None |
| for jd in jd_titles: |
| key = norm(jd) |
| if key and (key in title or title in key): |
| return jd, "align" |
| |
| title_tokens = set(title.split()) |
| for jd in jd_titles: |
| if title_tokens & set(norm(jd).split()): |
| return jd, "adjacent" |
| if jd_titles and has_skill_match: |
| return jd_titles[0], "mismatch" |
| return None, None |
|
|
|
|
| def _zmaps(rows: list[dict[str, Any]]) -> dict[str, dict[int, float]]: |
| """Per-column z-scores across the finalists (for picking each candidate's standout signal).""" |
| maps: dict[str, dict[int, float]] = {} |
| for column in SIGNAL_COLUMNS: |
| if not any(column in r for r in rows): |
| continue |
| values = [float(r.get(column) or 0.0) for r in rows] |
| mean = sum(values) / len(values) |
| var = sum((v - mean) ** 2 for v in values) / len(values) |
| std = var ** 0.5 or 1.0 |
| maps[column] = {i: (v - mean) / std for i, v in enumerate(values)} |
| return maps |
|
|
|
|
| def _dominant_signal(zmaps: dict[str, dict[int, float]], index: int) -> str: |
| best, best_z = "reranker_score", float("-inf") |
| for column, zmap in zmaps.items(): |
| z = zmap.get(index, 0.0) |
| if z > best_z: |
| best, best_z = column, z |
| return best |
|
|
|
|
| def _recently_active(row: dict[str, Any], recent_days: int) -> bool: |
| days = row.get("days_since_active") |
| return days is not None and float(days) <= recent_days |
|
|
|
|
| def _evidence_snippet(row: dict[str, Any], summary_limit: int = 500, experience_limit: int = 900) -> str: |
| """The candidate's OWN WORDS (profile summary + recent-role descriptions) — the raw material for an |
| outcome-oriented recommendation. The LLM extracts a concrete achievement from this, the NLG ignores it (it |
| can only template grounded facts). Summary and experience are truncated INDEPENDENTLY (not one combined |
| budget) — a verbose summary would otherwise eat the whole limit and starve out the role bullets, which is |
| where the concrete numbers (users served, % improvement, scale) actually live.""" |
| summary = str(row.get("summary") or "").strip()[:summary_limit] |
| experience = str(row.get("experience_text") or "").strip().replace("\n", "; ")[:experience_limit] |
| return f"{summary} {experience}".strip() |
|
|
|
|
| def build_facts( |
| row: dict[str, Any], spec: JDSpec, zmaps: dict[str, dict[int, float]], index: int, recent_days: int |
| ) -> dict[str, Any]: |
| """The grounded fact bundle for one candidate (JD-relative + behavioral signals).""" |
| years = round(float(row.get("years_of_experience") or 0.0), 1) |
| skills = matched_skills(spec.must_have_skills, row.get("flat_skills") or []) |
| missing = [s for s in spec.must_have_skills if s not in skills] |
| nice = matched_skills(spec.nice_to_have, row.get("flat_skills") or []) |
| certs = _overlap(spec.certifications, row.get("flat_certifications") or []) |
| city = str(row.get("location") or "").split(",")[0].strip() |
| loc_tier = bands.location_tier(city, spec.locations_preferred, spec.locations_acceptable) |
| title_match, title_band = title_fit(spec.job_titles, row.get("current_title") or "", bool(skills)) |
| rec_band = bands.recency_band(row.get("days_since_active"), recent_days) |
| signal = _dominant_signal(zmaps, index) |
| country = norm(row.get("country")) |
| country_fit = None |
| if spec.acceptable_countries: |
| country_fit = "ok" if any(norm(c) == country for c in spec.acceptable_countries) else "outside" |
| exclude_hit = any( |
| norm(e) and norm(e) in norm(row.get("current_company")) for e in spec.exclude_companies |
| ) |
| return { |
| "candidate_id": row.get("candidate_id", ""), |
| "title": str(row.get("current_title") or "Candidate").strip(), |
| "seniority": bands.seniority_label(row.get("experience_band"), row.get("current_title") or ""), |
| "years": years, |
| "in_band": spec.years_min <= years <= spec.years_max, |
| "exp_fit": bands.experience_fit(years, spec.years_min, spec.years_max), |
| "matched_skills": skills, |
| "missing_must": missing, |
| "matched_nice": nice, |
| "skill_band": bands.skill_coverage(len(skills), len(spec.must_have_skills)), |
| "matched_industries": matched_industries( |
| spec.target_industries, str(row.get("current_industry") or ""), row.get("flat_career_industries") or [] |
| ), |
| "matched_education": matched_education(spec.education_fields, row.get("education") or []), |
| "matched_certs": certs, |
| "title_match": title_match, |
| "title_band": title_band, |
| "city": city, |
| "location_tier": loc_tier, |
| "country_fit": country_fit, |
| "exclude_hit": exclude_hit, |
| "recency_band": rec_band, |
| "recently_active": _recently_active(row, recent_days), |
| "notice_band": bands.notice_band(row.get("notice_period_days"), row.get("is_available_soon")), |
| "bm25_hits": {f: bool(row.get(f"bm25_hit_{f}")) for f in BM25_FIELDS}, |
| "signals": signal_bands(row, rec_band, loc_tier == "outside"), |
| "dominant_signal": signal, |
| "signal_phrase": SIGNAL_PHRASES.get(signal, "strong overall match"), |
| "evidence": _evidence_snippet(row), |
| "scores": {col: round(float(row.get(col) or 0.0), 4) |
| for col in (*SIGNAL_COLUMNS, "final_score") if col in row or col == "final_score"}, |
| } |
|
|
