b1e9cdd182
- A1: fire the hard filter (Apify is precise); embedding vibe-ranker (embed.py) blends with the floor-free white-box (rank.py) to sift ~90 -> top 18 - A2: Opus curator (curate.py) reads the 18 -> honest <=count + Gemini-voiced report cards; graceful fallback to the white-box + templated cards - LLM via opencode.ai/zen gateway (llm.py): Opus chat + direct-OpenAI embeddings - Run LLM work off the event loop (asyncio.to_thread) so the Redis response publishes - C1: dedicated Postgres (app/db/) persists the per-user feed; get_scout_feed replays it so matches survive navigation/refresh - match contract + report-card fields (schema.py); skills.py; tests/
94 lines
4.2 KiB
Python
94 lines
4.2 KiB
Python
"""Skill normalization + coverage-aware overlap — the core accuracy lever (ENGINE_DESIGN §3.2).
|
|
|
|
v1: synonym canonicalization + lowercase/punctuation fold (no ESCO/KG yet — upgrade later).
|
|
**FLOOR-FREE** (§3.3): `overlap` returns a real score ONLY when both sides have skills; otherwise
|
|
`None` (the factor is *absent* → its weight redistributes — never a neutral constant like 60).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
# Seed synonym map → canonical. Extend over time; swap to ESCO + KG traversal later.
|
|
_SYNONYMS = {
|
|
"k8s": "kubernetes", "kube": "kubernetes",
|
|
"js": "javascript", "ts": "typescript", "react.js": "react", "reactjs": "react",
|
|
"node.js": "node", "nodejs": "node", "next.js": "nextjs", "nextjs": "nextjs",
|
|
"ml": "machine learning", "a.i.": "artificial intelligence", "ai": "artificial intelligence",
|
|
"dl": "deep learning", "nlp": "natural language processing", "genai": "generative ai",
|
|
"llm": "large language models", "llms": "large language models",
|
|
"pm": "product management", "product manager": "product management", "apm": "product management",
|
|
"postgres": "postgresql", "pg": "postgresql", "psql": "postgresql",
|
|
"gcp": "google cloud", "aws": "amazon web services",
|
|
"ds": "data science", "da": "data analytics", "ba": "business analysis",
|
|
"ux": "user experience", "ui": "user interface", "qa": "quality assurance",
|
|
"sde": "software engineering", "swe": "software engineering", "fe": "frontend", "be": "backend",
|
|
"fullstack": "full stack", "full-stack": "full stack", "rest": "rest api", "restful": "rest api",
|
|
"ci/cd": "cicd", "scrum": "agile", "go": "golang", "py": "python", "k8": "kubernetes",
|
|
}
|
|
|
|
_LEVEL_SCORE = {"Strong": 90, "Solid": 65, "Light": 35}
|
|
|
|
|
|
def norm(s: str) -> str:
|
|
s = re.sub(r"\s+", " ", (s or "").strip().lower())
|
|
if s in _SYNONYMS: # raw lookup first — keys keep punctuation (react.js, k8s, ci/cd)
|
|
return _SYNONYMS[s]
|
|
s = re.sub(r"\s+", " ", re.sub(r"[._/]+", " ", s)).strip() # then fold punctuation
|
|
return _SYNONYMS.get(s, s)
|
|
|
|
|
|
def canon(skills) -> set[str]:
|
|
out: set[str] = set()
|
|
for s in skills or []:
|
|
if not s:
|
|
continue
|
|
n = norm(str(s))
|
|
if n:
|
|
out.add(n)
|
|
return out
|
|
|
|
|
|
def _tok(s: str) -> set[str]:
|
|
return {t for t in s.split() if len(t) > 2}
|
|
|
|
|
|
def _level(job_skill: str, user_canon: set[str], user_tokens: set[str]) -> str:
|
|
if job_skill in user_canon:
|
|
return "Strong"
|
|
jt = _tok(job_skill)
|
|
if jt and (jt & user_tokens):
|
|
return "Solid"
|
|
return "Light"
|
|
|
|
|
|
def overlap(user_skills, job_skills) -> dict | None:
|
|
"""Coverage-aware skill F1 (precision & recall each ≤ 1). Returns None when EITHER side is
|
|
empty (absent factor — never a constant floor)."""
|
|
user = canon(user_skills)
|
|
job = canon(job_skills)
|
|
if not user or not job:
|
|
return None
|
|
user_tokens: set[str] = set().union(*(_tok(u) for u in user)) if user else set()
|
|
job_tokens: set[str] = set().union(*(_tok(j) for j in job)) if job else set()
|
|
matched_job = [j for j in job if _level(j, user, user_tokens) != "Light"] # job needs the user meets
|
|
matched_user = [u for u in user if u in job or (_tok(u) & job_tokens)] # user skills relevant here
|
|
recall = len(matched_job) / len(job) # ≤ 1
|
|
precision = len(matched_user) / len(user) # ≤ 1
|
|
f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0
|
|
return {
|
|
"score": f1, "recall": recall,
|
|
"matched": sorted(matched_job),
|
|
"missing": sorted(j for j in job if j not in matched_job),
|
|
}
|
|
|
|
|
|
def dims(user_skills, job_skills, k: int = 4) -> list[dict]:
|
|
"""Per-skill breakdown rows (job's top skills marked Strong/Solid/Light vs the user)."""
|
|
user = canon(user_skills)
|
|
user_tokens: set[str] = set().union(*(_tok(u) for u in user)) if user else set()
|
|
rows = []
|
|
for j in list(canon(job_skills))[:k]:
|
|
lvl = _level(j, user, user_tokens) if user else "Light"
|
|
rows.append({"name": j.title(), "score": _LEVEL_SCORE[lvl], "level": lvl})
|
|
return rows
|