| """Structured JD spec — the OBJECTIVE fields (filters + BM25), parsed from the unified extraction. |
| |
| Two sources of JD truth live in ONE combined file (``data/jd_extraction.json``): |
| * OBJECTIVE (this module's ``JDSpec``) — hard facts the filters/BM25 read: skills, years, locations, |
| target/excluded industries & companies, job titles, education fields. |
| * SUBJECTIVE — ``ideal_summary`` only, read by ``read_subjective`` and used as the embedding-similarity |
| query text (compared against candidate profiles). |
| |
| The objective keys are aligned to the candidate schema so each maps to a real candidate field. Live-only: |
| when the generated file is absent/unreadable, ``extract_spec`` returns an EMPTY default spec (no committed |
| fallback). Never raises. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from pathlib import Path |
| from typing import Any |
|
|
| import orjson |
|
|
| from common.logging import get_logger |
| from common.paths import REPO_ROOT |
| from config import Settings, load_settings |
|
|
|
|
| @dataclass(frozen=True) |
| class JDSpec: |
| """The JD's OBJECTIVE fields (filters + BM25). Each maps to a candidate-schema field. |
| |
| Defaults double as a safe fallback for any field the model omits. |
| """ |
|
|
| must_have_skills: list[str] = field(default_factory=list) |
| nice_to_have: list[str] = field(default_factory=list) |
| years_min: float = 5.0 |
| years_max: float = 9.0 |
| locations_preferred: list[str] = field(default_factory=list) |
| locations_acceptable: list[str] = field(default_factory=list) |
| acceptable_countries: list[str] = field(default_factory=list) |
| target_industries: list[str] = field(default_factory=list) |
| exclude_companies: list[str] = field(default_factory=list) |
| job_titles: list[str] = field(default_factory=list) |
| education_fields: list[str] = field(default_factory=list) |
| certifications: list[str] = field(default_factory=list) |
|
|
|
|
| def resolve(path_str: str) -> Path: |
| """Resolve a config path against the repo root when relative (cwd-independent).""" |
| candidate = Path(path_str) |
| return candidate if candidate.is_absolute() else REPO_ROOT / candidate |
|
|
|
|
| def _str_list(value: Any) -> list[str]: |
| """Coerce a value to a clean list[str] (non-list → []).""" |
| if not isinstance(value, list): |
| return [] |
| return [str(item).strip() for item in value if str(item).strip()] |
|
|
|
|
| def _spec_from_mapping(raw: dict[str, Any]) -> JDSpec: |
| """Build a JDSpec from a mapping (the generated extraction); defaults fill omissions.""" |
| return JDSpec( |
| must_have_skills=_str_list(raw.get("must_have_skills")), |
| nice_to_have=_str_list(raw.get("nice_to_have")), |
| years_min=float(raw.get("years_min", 5) or 5), |
| years_max=float(raw.get("years_max", 9) or 9), |
| locations_preferred=_str_list(raw.get("locations_preferred")), |
| locations_acceptable=_str_list(raw.get("locations_acceptable")), |
| acceptable_countries=_str_list(raw.get("acceptable_countries")), |
| target_industries=_str_list(raw.get("target_industries")), |
| exclude_companies=_str_list(raw.get("exclude_companies")), |
| job_titles=_str_list(raw.get("job_titles")), |
| education_fields=_str_list(raw.get("education_fields")), |
| certifications=_str_list(raw.get("certifications")), |
| ) |
|
|
|
|
| def extract_spec(settings: Settings | None = None) -> JDSpec: |
| """Load the JD spec from the live extraction JSON. Absent/unreadable → an empty default spec (live-only, |
| no committed fallback). Never raises — ``run_jd_parse`` generates the JSON each run before this is read.""" |
| active = settings or load_settings() |
| extraction_path = resolve(active.jd.extraction_path) |
| if extraction_path.exists(): |
| try: |
| parsed = orjson.loads(extraction_path.read_bytes()) |
| if isinstance(parsed, dict) and parsed: |
| return _spec_from_mapping(parsed) |
| except Exception as failure: |
| get_logger("jd").warning("jd.extract.unreadable", path=str(extraction_path), error=str(failure)) |
| get_logger("jd").warning("jd.extract.missing", path=str(extraction_path), note="empty spec (defaults)") |
| return JDSpec() |
|
|
|
|
| def embed_query(spec: JDSpec, ideal_summary: str, max_skills: int = 15, max_titles: int = 8) -> str: |
| """Compose the semantic embed/rerank query from the JD: the ideal_summary (candidate-profile register) |
| followed by the top extracted skills and roles — so potion/granite/cross-encoder use the FULL JD, not just |
| the subjective summary. ideal_summary LEADS (most important; survives the embedders' length truncation, and |
| leaves the cross-encoder room for the candidate doc). Empty summary ⇒ skills/roles only; all empty ⇒ "".""" |
| parts: list[str] = [] |
| summary = (ideal_summary or "").strip() |
| if summary: |
| parts.append(summary) |
| if spec.must_have_skills: |
| parts.append("Core skills: " + ", ".join(spec.must_have_skills[:max_skills]) + ".") |
| if spec.job_titles: |
| parts.append("Relevant roles: " + ", ".join(spec.job_titles[:max_titles]) + ".") |
| return " ".join(parts).strip() |
|
|
|
|
| def read_subjective(path: str | Path) -> dict[str, Any]: |
| """Read the SUBJECTIVE block (just ``ideal_summary``, the embedding text) from the JSON.""" |
| source = resolve(str(path)) |
| if not source.exists(): |
| return {} |
| try: |
| raw: Any = orjson.loads(source.read_bytes()) |
| except Exception: |
| return {} |
| subjective = raw.get("subjective") if isinstance(raw, dict) else None |
| return subjective if isinstance(subjective, dict) else {} |
|
|
