"""Coverage-aware, FLOOR-FREE white-box ranker (Phase 1 — ENGINE_DESIGN §3, floor-free per §3.3). The cheap cold-start ranker — one of the three rankers fused in Phase 3, NOT "the engine." Hard rule (the one the old engine broke): **every factor returns a real 0–1 score OR `None`**; an absent factor leaves the weighted average (its weight redistributes) — never a constant floor. """ from __future__ import annotations import re from dataclasses import dataclass, field from app.engine import skills as S from app.engine.board_adapters import coerce as C from app.engine.schema import Growth, MatchDim, MatchResult # Expert-prior weights (§3.4 — to calibrate, not optima). semantic = keyword placeholder → embedding in §3. WEIGHTS = {"skill": 0.35, "experience": 0.25, "location": 0.15, "salary": 0.10, "industry": 0.05, "role": 0.05, "semantic": 0.10} FACTOR_LABELS = {"skill": "Skill match", "experience": "Experience", "location": "Location fit", "salary": "Salary fit", "industry": "Industry", "role": "Role fit", "semantic": "Role relevance"} # job seniority_level → (min, max) years band BAND_YEARS = {"junior": (0, 3), "mid": (3, 8), "senior": (8, 13), "lead": (12, 99)} SELECT = {"safe": 0.85, "balanced": 0.70, "reach": 0.55} FLOOR, CAP = 10, 25 _STOP = {"the", "and", "for", "with", "you", "our", "your", "are", "this", "that", "will", "job", "role"} @dataclass class UserProfile: title: str | None = None skills: set[str] = field(default_factory=set) years: int | None = None cities: list[str] = field(default_factory=list) countries: list[str] = field(default_factory=list) onsite_only: bool = False has_loc_pref: bool = False industries: list[str] = field(default_factory=list) roles: list[str] = field(default_factory=list) comp_lpa: float | None = None # target minimum, in lakhs deal_breakers: list[str] = field(default_factory=list) stretch: str = "balanced" profile_tokens: set[str] = field(default_factory=set) def _comp_min_lpa(prefs: dict) -> float | None: nums = re.findall(r"\d+", str(prefs.get("targetComp") or "")) return float(nums[0]) if nums else None def profile_from(prefs: dict, user_context: dict | None = None) -> UserProfile: locs = C.parsed_locations(prefs) cities = [lp["city"].lower() for lp in locs if lp.get("city")] countries = [lp["country"].lower() for lp in locs if lp.get("country")] wm = [w.lower() for w in (prefs.get("workMode") or [])] title = C.title_of(prefs) or None skills = S.canon((user_context or {}).get("skills")) # empty in Phase 1 (prefs-only) toks = {t for t in re.findall(r"[a-z0-9+#]+", f"{title or ''} {' '.join(skills)}".lower()) if len(t) > 2} return UserProfile( title=title, skills=skills, years=C.years_of(prefs), cities=cities, countries=countries, onsite_only=("on-site" in wm or "onsite" in wm), has_loc_pref=bool(cities or countries), industries=[i for i in (prefs.get("industry") or []) if i and i != "Any"], roles=[r for r in (prefs.get("role") or []) if r], comp_lpa=_comp_min_lpa(prefs), deal_breakers=prefs.get("dealBreakers") or [], stretch=(prefs.get("stretch") or "balanced"), profile_tokens=toks, ) # ── job accessors ── def _jskills(job): return (job.get("details") or {}).get("skills") or job.get("required_skills") or [] def _jindustry(job): return (job.get("details") or {}).get("industry") or job.get("industry") def _jrole(job): return (job.get("details") or {}).get("role_category") # ── factors: each returns float in [0,1] OR None (absent → weight redistributes) ── def f_skill(p, job): o = S.overlap(p.skills, _jskills(job)) return o["score"] if o else None def f_experience(p, job): band = BAND_YEARS.get((job.get("seniority_level") or "").lower()) if band is None or p.years is None: return None lo, hi = band if p.years < lo: return max(0.2, 1.0 - (lo - p.years) / max(lo, 1) * 1.2) if p.years > hi: return max(0.4, 1.0 - (p.years - hi) / max(p.years, 1) * 0.9) return 1.0 def f_location(p, job): if (job.get("location_mode") or "").lower() == "remote": return 0.0 if p.onsite_only else 1.0 if not p.has_loc_pref: return None # "Anywhere"/open → absent, NOT a 60 floor jl = f"{job.get('location_city') or ''} {job.get('location_country') or ''}".lower().strip() if not jl: return None if any(c and c in jl for c in p.cities): return 1.0 # exact city / metro if any(c and c in jl for c in p.countries): return 0.6 # same country, different city return 0.0 # different country → hard-filtered out def f_salary(p, job): jl = job.get("salary_lpa") if jl is None or p.comp_lpa is None: return None if jl >= p.comp_lpa: return 1.0 gap = (p.comp_lpa - jl) / max(p.comp_lpa, 1) return max(0.3, 1.0 - gap) def _cat_match(wants, value): if not wants or not value: return None v = str(value).lower() return 1.0 if any(w.lower() in v or v in w.lower() for w in wants) else 0.3 def f_industry(p, job): return _cat_match(p.industries, _jindustry(job)) def f_role(p, job): return _cat_match(p.roles, _jrole(job)) def f_semantic(p, job): if not p.profile_tokens: return None blob = f"{job.get('title') or ''} {(job.get('details') or {}).get('description') or ''}".lower() jt = {t for t in re.findall(r"[a-z0-9+#]+", blob) if len(t) > 2 and t not in _STOP} if not jt: return None return len(p.profile_tokens & jt) / len(p.profile_tokens) FACTORS = {"skill": f_skill, "experience": f_experience, "location": f_location, "salary": f_salary, "industry": f_industry, "role": f_role, "semantic": f_semantic} def utility(p: UserProfile, job: dict): factors = {k: fn(p, job) for k, fn in FACTORS.items()} num = sum(WEIGHTS[k] * v for k, v in factors.items() if v is not None) den = sum(WEIGHTS[k] for k, v in factors.items() if v is not None) U = (num / den) if den else None # NO present signals → None (filtered out) return U, factors def passes_filter(p: UserProfile, job: dict) -> bool: loc = f_location(p, job) if loc == 0.0: # remote-for-onsite-only OR explicit no-fit return False band = BAND_YEARS.get((job.get("seniority_level") or "").lower()) if band and p.years is not None and p.stretch != "reach": lo, hi = band if p.years + (3 if p.stretch == "balanced" else 0) < lo: # job needs much more seniority return False if p.years - 8 > hi: # wildly over-qualified return False return True def _level(score: int) -> str: return "Strong" if score >= 72 else "Solid" if score >= 52 else "Light" # per-dimension one-line notes (the Gemini report-card voice). Templated → Opus replaces later. _DIM_NOTES = { "skill": {"Strong": "Covers most of the role's must-have skills", "Solid": "Covers several must-haves — a few gaps", "Light": "Light on the listed skills"}, "experience": {"Strong": "Your level sits right in their band", "Solid": "Just outside their experience band", "Light": "Off their experience band"}, "location": {"Strong": "Right in your target location", "Solid": "Same country, different city", "Light": "Off your preferred locations"}, "salary": {"Strong": "Meets your target pay", "Solid": "Near your target pay", "Light": "Below your target pay"}, "industry": {"Strong": "In your target industry", "Solid": "An adjacent industry", "Light": "A different industry"}, "role": {"Strong": "Matches your target role area", "Solid": "An adjacent role area", "Light": "A different role area"}, "semantic": {"Strong": "Reads very close to your search", "Solid": "Reads close to your search", "Light": "Loosely related to your search"}, } def _breakdown(factors) -> list[MatchDim]: dims = [] for k, v in factors.items(): if v is None: continue sc = round(v * 100) lvl = _level(sc) dims.append(MatchDim(name=FACTOR_LABELS[k], score=sc, level=lvl, note=_DIM_NOTES.get(k, {}).get(lvl))) dims.sort(key=lambda d: -d.score) return dims[:4] def _archetype(fit: str, score: int) -> str: if fit == "fit": return "The Strong Match" if score >= 85 else "The Solid Fit" return "The Stretch Worth Taking" if score >= 68 else "The Long Shot" def _one_line(dims: list[MatchDim]) -> str: strong = [d.name.lower() for d in dims if d.level == "Strong"][:2] light = [d.name.lower() for d in dims if d.level == "Light"][:1] if strong and light: return f"Your {' and '.join(strong)} land cleanly — {light[0]} is the gap to bridge." if strong: return f"Your {' and '.join(strong)} make this a clean match." if light: return f"A reach — {light[0]} is the stretch, but the rest lines up." return "A balanced match worth a closer look." def _coach(dims: list[MatchDim]) -> str: strong = next((d.name.lower() for d in dims if d.level == "Strong"), None) weak = min((d for d in dims if d.level == "Light"), key=lambda d: d.score, default=None) if weak and strong: return f"Lead with your {strong}; shore up {weak.name.lower()} before you apply and this reads as a clean fit." if weak: return f"Worth a look — close the {weak.name.lower()} gap to strengthen your case." return "A strong fit — apply with confidence." def _fit(p: UserProfile, U: float, factors) -> str: sk = factors.get("skill") if p.stretch == "reach": return "stretch" if U < 0.8 else "fit" strong = (sk is None or sk >= 0.5) and U >= 0.7 return "fit" if strong else "stretch" def _reason(job, dims: list[MatchDim], fit: str) -> str: board = (job.get("note") or "").replace("Live from ", "") or "the boards" top = [d.name.lower() for d in dims if d.level in ("Strong", "Solid")][:2] lead = " · ".join(top) if top else ("a reach worth a look" if fit == "stretch" else "a solid fit") return f"{lead.capitalize()} — fresh from {board}" def _growth(dims: list[MatchDim], score: int) -> Growth | None: light = [d for d in dims if d.level == "Light"] if not light: return None w = min(light, key=lambda d: d.score) return Growth(text=f"Close the {w.name.lower()} gap to lift this match", **{"from": score, "to": min(99, score + 8)}) def _match(p: UserProfile, job: dict, U: float, factors) -> MatchResult: score = round(U * 100) dims = _breakdown(factors) fit = _fit(p, U, factors) return MatchResult( score=score, fit=fit, archetype=_archetype(fit, score), one_line=_one_line(dims), reason=_reason(job, dims, fit), breakdown=dims, growth=_growth(dims, score), coach_note=_coach(dims), proofReady=(factors.get("skill") is not None and factors["skill"] >= 0.72), factors={k: (round(v, 3) if v is not None else None) for k, v in factors.items()}, ) def rank(prefs: dict, user_context: dict | None, jobs: list[dict]) -> dict: """Score ALL jobs (no filter, no cut). The bouncer is fired — Apify already guarantees city/title precision, so location/seniority are *scoring* signals, not gates. This is the cheap SIFTER pass: it ranks everything; `sift.sift()` then blends it with the embedding vibe to pick the top ~18 the curator (Opus) reads. The final cut is Opus's call (or `select()` in the fallback path).""" p = profile_from(prefs, user_context) scored = [] for job in jobs: U, factors = utility(p, job) if U is None: # no scorable factor at all → can't rank it continue scored.append((U, factors, job)) scored.sort(key=lambda x: -x[0]) out = [] for U, factors, job in scored: m = _match(p, job, U, factors) job["match"] = m.as_dict() job["matchScore"] = m.score out.append(job) return {"opportunities": out, "ranked": len(out), "scanned": len(jobs)} def select(jobs: list[dict], stretch: str = "balanced", floor: int = FLOOR, cap: int = CAP) -> list[dict]: """Final shortlist cut — applied AFTER fusion (Phase 3), NEVER on the cheap white-box alone. Cuts by the (fused) `matchScore` against the stretch threshold, with a floor (never empty) and cap (~25). `jobs` must already be ranked best-first.""" bar = round(SELECT.get(stretch, 0.70) * 100) passing = [j for j in jobs if j.get("matchScore", 0) >= bar] if len(passing) < floor: passing = jobs[:floor] return passing[:cap]