Spaces:
Sleeping
Sleeping
| """ | |
| AI Talent Screening System — FastAPI Backend | |
| server.py — Serves the UI and exposes API endpoints. | |
| Run with: | |
| uvicorn server:app --reload --port 8000 | |
| Then open: http://localhost:8000 | |
| """ | |
| import csv | |
| import json | |
| import os | |
| import re | |
| import subprocess | |
| import sys | |
| from datetime import datetime | |
| from pathlib import Path | |
| from fastapi import FastAPI, File, HTTPException, Query, UploadFile | |
| #from fastapi.requests import Request | |
| from fastapi import Request | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.templating import Jinja2Templates | |
| from typing import List | |
| # ─── PATHS ──────────────────────────────────────────────────────────────────── | |
| BASE = Path(__file__).parent | |
| # Honour the same env var as config.py so HF Spaces / Docker deploys keep | |
| # resumes, JDs, the DB and CSV outputs on a persistent volume. | |
| _DATA_ROOT = Path(os.environ.get("RESUME_ANALYZER_DATA_ROOT", BASE)) | |
| RESUMES_DIR = _DATA_ROOT / "data" / "resumes" | |
| JDS_DIR = _DATA_ROOT / "data" / "jds" | |
| RESUME_CSV = _DATA_ROOT / "outputs" / "match_results" / "resume_data.csv" | |
| SCREENING_CSV = _DATA_ROOT / "outputs" / "match_results" / "screening_history.csv" | |
| MAIN_PY = BASE / "main.py" | |
| # Ensure directories exist | |
| for d in [RESUMES_DIR, JDS_DIR, RESUME_CSV.parent]: | |
| d.mkdir(parents=True, exist_ok=True) | |
| # ─── BOOT DIAGNOSTICS — printed to stderr so they show up in HF Space logs ─── | |
| import sys as _sys | |
| _static_dir = BASE / "static" | |
| _tpl_dir = BASE / "templates" | |
| print(f"[boot] cwd : {os.getcwd()}", file=_sys.stderr) | |
| print(f"[boot] BASE : {BASE}", file=_sys.stderr) | |
| print(f"[boot] static dir : {_static_dir} " | |
| f"exists={_static_dir.exists()} " | |
| f"contents={sorted(p.name for p in _static_dir.iterdir()) if _static_dir.exists() else 'N/A'}", | |
| file=_sys.stderr) | |
| print(f"[boot] templates : {_tpl_dir} " | |
| f"exists={_tpl_dir.exists()} " | |
| f"contents={sorted(p.name for p in _tpl_dir.iterdir()) if _tpl_dir.exists() else 'N/A'}", | |
| file=_sys.stderr) | |
| print(f"[boot] /app top : {sorted(os.listdir('/app')) if os.path.isdir('/app') else 'N/A'}", | |
| file=_sys.stderr) | |
| # ─── APP SETUP ──────────────────────────────────────────────────────────────── | |
| app = FastAPI(title="AI Talent Screener") | |
| # Mount static files. If the directory is missing for any reason, log loudly | |
| # rather than failing import — at least the API still works for diagnosis. | |
| if _static_dir.exists(): | |
| app.mount("/static", StaticFiles(directory=_static_dir), name="static") | |
| print(f"[boot] mounted /static -> {_static_dir}", file=_sys.stderr) | |
| else: | |
| print(f"[boot] WARNING: {_static_dir} does not exist; /static will 404", | |
| file=_sys.stderr) | |
| templates = Jinja2Templates(directory=_tpl_dir) | |
| # Tiny ad-hoc endpoint that mirrors the boot diagnostics over HTTP so we can | |
| # inspect a deployed instance without shell access. Hit /api/_boot to confirm. | |
| async def _boot_info(): | |
| static = BASE / "static" | |
| tpl = BASE / "templates" | |
| return { | |
| "cwd": os.getcwd(), | |
| "BASE": str(BASE), | |
| "static_dir": str(static), | |
| "static_exists": static.exists(), | |
| "static_contents": (sorted(p.name for p in static.iterdir()) | |
| if static.exists() else None), | |
| "templates_dir": str(tpl), | |
| "templates_exists": tpl.exists(), | |
| "templates_contents": (sorted(p.name for p in tpl.iterdir()) | |
| if tpl.exists() else None), | |
| "app_root_listing": (sorted(os.listdir("/app")) | |
| if os.path.isdir("/app") else None), | |
| } | |
| # ─── HELPERS ────────────────────────────────────────────────────────────────── | |
| def fmt_bytes(n: int) -> str: | |
| for u in ["B", "KB", "MB"]: | |
| if n < 1024: | |
| return f"{n:.0f} {u}" | |
| n /= 1024 | |
| return f"{n:.1f} GB" | |
| def file_info(path: Path) -> dict: | |
| stat = path.stat() | |
| return { | |
| "name": path.name, | |
| "size": fmt_bytes(stat.st_size), | |
| "modified": datetime.fromtimestamp(stat.st_mtime).strftime("%d %b %Y %H:%M"), | |
| } | |
| def read_csv_safe(path: Path): | |
| """Read a CSV safely, skipping bad lines.""" | |
| if not path.exists(): | |
| return [] | |
| rows = [] | |
| try: | |
| with open(path, encoding="utf-8", errors="replace") as f: | |
| reader = csv.DictReader(f) | |
| for row in reader: | |
| rows.append(dict(row)) | |
| except Exception: | |
| pass | |
| return rows | |
| def parse_list_field(val: str) -> list[str]: | |
| """Turn comma string or JSON list into a clean Python list.""" | |
| if not val or val.strip() in ["", "nan", "[]", "0"]: | |
| return [] | |
| s = val.strip() | |
| if s.startswith("["): | |
| try: | |
| items = json.loads(s) | |
| return [str(i).strip() for i in items if str(i).strip()] | |
| except Exception: | |
| pass | |
| return [x.strip() for x in s.split(",") if x.strip()] | |
| # ─── ROUTES ─────────────────────────────────────────────────────────────────── | |
| async def index(request: Request): | |
| """Serve the main single-page application.""" | |
| return templates.TemplateResponse(request, "index.html") | |
| # ── File Management ────────────────────────────────────────────────────────── | |
| async def list_resumes(): | |
| files = [file_info(f) for f in sorted(RESUMES_DIR.iterdir()) | |
| if f.is_file() and not f.name.startswith(".")] | |
| return {"files": files, "count": len(files)} | |
| async def list_jds(): | |
| files = [file_info(f) for f in sorted(JDS_DIR.iterdir()) | |
| if f.is_file() and not f.name.startswith(".")] | |
| return {"files": files, "count": len(files)} | |
| async def upload_resumes(files: List[UploadFile] = File(...)): | |
| saved = [] | |
| for file in files: | |
| dest = RESUMES_DIR / file.filename | |
| content = await file.read() | |
| dest.write_bytes(content) | |
| saved.append(file.filename) | |
| return {"saved": saved, "count": len(saved)} | |
| async def upload_jds(files: list[UploadFile] = File(...)): | |
| saved = [] | |
| for file in files: | |
| dest = JDS_DIR / file.filename | |
| content = await file.read() | |
| dest.write_bytes(content) | |
| saved.append(file.filename) | |
| return {"saved": saved, "count": len(saved)} | |
| async def delete_resume(filename: str): | |
| path = RESUMES_DIR / filename | |
| if not path.exists(): | |
| raise HTTPException(status_code=404, detail="File not found") | |
| path.unlink() | |
| return {"deleted": filename} | |
| async def delete_jd(filename: str): | |
| path = JDS_DIR / filename | |
| if not path.exists(): | |
| raise HTTPException(status_code=404, detail="File not found") | |
| path.unlink() | |
| return {"deleted": filename} | |
| # ── Output CSVs ────────────────────────────────────────────────────────────── | |
| async def get_resume_data(): | |
| rows = read_csv_safe(RESUME_CSV) | |
| return { | |
| "rows": rows, | |
| "count": len(rows), | |
| "columns": list(rows[0].keys()) if rows else [], | |
| "exists": RESUME_CSV.exists(), | |
| } | |
| async def get_screening_history(): | |
| rows = read_csv_safe(SCREENING_CSV) | |
| return { | |
| "rows": rows, | |
| "count": len(rows), | |
| "columns": list(rows[0].keys()) if rows else [], | |
| "exists": SCREENING_CSV.exists(), | |
| } | |
| # ── Pipeline ───────────────────────────────────────────────────────────────── | |
| # The pipeline can take 2-15 minutes per run depending on resume count, LLM | |
| # rate limits, and whether JobBERT needs warming. Holding the HTTP request | |
| # open that long doesn't work behind HF Spaces' ~90s proxy timeout — the | |
| # browser sees a disconnect while the subprocess keeps running silently. | |
| # | |
| # So we run main.py in a background thread, key it by a job UUID, and let | |
| # the UI poll /api/run/status/{job_id} until the job reports completed. | |
| import threading as _threading | |
| import uuid as _uuid_pipeline | |
| import time as _time | |
| _PIPELINE_JOBS: dict = {} # job_id -> {status, result, started_at, finished_at, cmd} | |
| _PIPELINE_LOCK = _threading.Lock() | |
| def _strip_ansi(s: str) -> str: | |
| return re.sub(r'\x1b\[[0-9;]*m', '', s or '') | |
| def _run_pipeline_worker(job_id: str, cmd: list[str]) -> None: | |
| """ | |
| Background-thread worker that runs main.py with live stdout/stderr capture. | |
| Uses Popen + threaded readers so the /api/run/status endpoint can show | |
| progress while the pipeline is still running, not only after it finishes. | |
| """ | |
| with _PIPELINE_LOCK: | |
| _PIPELINE_JOBS[job_id]["status"] = "running" | |
| _PIPELINE_JOBS[job_id]["stdout"] = "" | |
| _PIPELINE_JOBS[job_id]["stderr"] = "" | |
| proc = subprocess.Popen( | |
| cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, | |
| text=True, bufsize=1, # line-buffered | |
| cwd=str(BASE), | |
| ) | |
| def _pump(stream, key): | |
| """Read one stream line-by-line, append to the job's buffer.""" | |
| try: | |
| for line in iter(stream.readline, ""): | |
| stripped = _strip_ansi(line) | |
| with _PIPELINE_LOCK: | |
| j = _PIPELINE_JOBS.get(job_id) | |
| if j is None: | |
| return | |
| # Keep the rolling tail bounded so memory doesn't explode | |
| # on chatty runs (~10 KB of stdout, ~5 KB of stderr). | |
| j[key] = (j.get(key, "") + stripped)[-10_000:] | |
| finally: | |
| try: stream.close() | |
| except Exception: pass | |
| t_out = _threading.Thread(target=_pump, args=(proc.stdout, "stdout"), daemon=True) | |
| t_err = _threading.Thread(target=_pump, args=(proc.stderr, "stderr"), daemon=True) | |
| t_out.start(); t_err.start() | |
| try: | |
| rc = proc.wait(timeout=1500) # 25 min hard ceiling | |
| t_out.join(timeout=5) | |
| t_err.join(timeout=5) | |
| with _PIPELINE_LOCK: | |
| _PIPELINE_JOBS[job_id].update({ | |
| "status": "completed" if rc == 0 else "failed", | |
| "returncode": rc, | |
| "finished_at": _time.time(), | |
| }) | |
| except subprocess.TimeoutExpired: | |
| proc.kill() | |
| with _PIPELINE_LOCK: | |
| _PIPELINE_JOBS[job_id].update({ | |
| "status": "timeout", | |
| "stderr": (_PIPELINE_JOBS[job_id].get("stderr", "") + | |
| "\nPipeline exceeded 25-minute internal timeout."), | |
| "finished_at": _time.time(), | |
| }) | |
| except Exception as exc: | |
| with _PIPELINE_LOCK: | |
| _PIPELINE_JOBS[job_id].update({ | |
| "status": "error", | |
| "stderr": (_PIPELINE_JOBS[job_id].get("stderr", "") + | |
| f"\nWorker crashed: {type(exc).__name__}: {exc}"), | |
| "finished_at": _time.time(), | |
| }) | |
| async def run_pipeline( | |
| top_n: int = Query(default=10, ge=1, le=50), | |
| explain_n: int = Query(default=3, ge=1, le=10), | |
| jd: str = Query(default=""), | |
| weights: str = Query(default=""), | |
| ): | |
| """ | |
| Kick off the pipeline as a background job and return a job_id immediately | |
| (within ~10ms). The UI polls /api/run/status/{job_id} for completion. | |
| """ | |
| if not MAIN_PY.exists(): | |
| raise HTTPException(status_code=404, detail="main.py not found") | |
| cmd = [sys.executable, str(MAIN_PY), "--top", str(top_n), "--explain", str(explain_n)] | |
| if jd: | |
| stem = jd.rsplit(".", 1)[0] | |
| stem = stem.replace("_", " ").replace("-", " ") | |
| stem = re.sub(r"\s+[Jj][Dd]$", "", stem).strip() | |
| stem = re.sub(r"\s+[Jj]ob\s+[Dd]escription$", "", stem, | |
| flags=re.IGNORECASE).strip() | |
| stem = re.sub(r"\s+", " ", stem).strip() | |
| cmd += ["--jd", stem] | |
| if weights: | |
| cmd += ["--weights", weights] | |
| # Reject concurrent runs — main.py writes to a shared SQLite + CSVs; | |
| # parallel runs would corrupt them. | |
| with _PIPELINE_LOCK: | |
| running = [jid for jid, j in _PIPELINE_JOBS.items() | |
| if j.get("status") in ("queued", "running")] | |
| if running: | |
| raise HTTPException(status_code=409, | |
| detail=f"Another pipeline is still running: {running[0]}") | |
| job_id = str(_uuid_pipeline.uuid4()) | |
| _PIPELINE_JOBS[job_id] = { | |
| "status": "queued", | |
| "cmd": cmd, | |
| "started_at": _time.time(), | |
| } | |
| _threading.Thread(target=_run_pipeline_worker, | |
| args=(job_id, cmd), daemon=True).start() | |
| return {"job_id": job_id, "status": "queued"} | |
| async def run_status(job_id: str): | |
| """ | |
| Return the current status of a pipeline job. UI polls this every few | |
| seconds. Returns 404 only for truly unknown IDs — otherwise always | |
| returns a dict with at least `status`. | |
| """ | |
| with _PIPELINE_LOCK: | |
| job = _PIPELINE_JOBS.get(job_id) | |
| if not job: | |
| raise HTTPException(status_code=404, detail="Unknown job_id") | |
| elapsed = _time.time() - job["started_at"] | |
| return { | |
| "job_id": job_id, | |
| "status": job["status"], | |
| "elapsed_s": round(elapsed, 1), | |
| "success": job["status"] == "completed", | |
| "returncode": job.get("returncode"), | |
| "stdout": job.get("stdout", ""), | |
| "stderr": job.get("stderr", ""), | |
| } | |
| async def run_jobs(): | |
| """Debug: list every job and its current status. Useful when a UI lost its job_id.""" | |
| with _PIPELINE_LOCK: | |
| return { | |
| "jobs": [ | |
| {"job_id": jid, "status": j["status"], | |
| "elapsed_s": round(_time.time() - j["started_at"], 1)} | |
| for jid, j in sorted(_PIPELINE_JOBS.items(), | |
| key=lambda kv: kv[1].get("started_at", 0), | |
| reverse=True) | |
| ][:20] | |
| } | |
| # ── Results & Candidates ───────────────────────────────────────────────────── | |
| async def get_results( | |
| run: str = Query(default=""), | |
| jd: str = Query(default=""), | |
| tier: str = Query(default=""), | |
| ): | |
| rows = read_csv_safe(SCREENING_CSV) | |
| if run: | |
| rows = [r for r in rows if r.get("run_timestamp", "") == run] | |
| if jd: | |
| rows = [r for r in rows if r.get("jd_title", "") == jd] | |
| if tier: | |
| rows = [r for r in rows if r.get("confidence_tier", "").lower() == tier.lower()] | |
| # Build filter options from full dataset | |
| all_rows = read_csv_safe(SCREENING_CSV) | |
| runs = sorted(set(r.get("run_timestamp", "") for r in all_rows if r.get("run_timestamp")), reverse=True) | |
| jds = sorted(set(r.get("jd_title", "") for r in all_rows if r.get("jd_title"))) | |
| tiers = sorted(set(r.get("confidence_tier", "") for r in all_rows if r.get("confidence_tier"))) | |
| return { | |
| "rows": rows, | |
| "count": len(rows), | |
| "filters": {"runs": runs, "jds": jds, "tiers": tiers}, | |
| } | |
| async def get_candidate(name: str): | |
| resume_rows = read_csv_safe(RESUME_CSV) | |
| hist_rows = read_csv_safe(SCREENING_CSV) | |
| candidate = next( | |
| (r for r in resume_rows if r.get("candidate_name", "").lower() == name.lower()), | |
| None | |
| ) | |
| if not candidate: | |
| raise HTTPException(status_code=404, detail=f"Candidate '{name}' not found") | |
| history = [r for r in hist_rows | |
| if r.get("candidate_name", "").lower() == name.lower()] | |
| # Parse list fields | |
| candidate["_skills_list"] = parse_list_field(candidate.get("technical_skills", "")) | |
| candidate["_soft_skills_list"] = parse_list_field(candidate.get("soft_skills", "")) | |
| candidate["_pros_list"] = parse_list_field(candidate.get("pros", "")) | |
| candidate["_cons_list"] = parse_list_field(candidate.get("cons", "")) | |
| candidate["_projects_list"] = parse_list_field(candidate.get("projects", "")) | |
| candidate["_achievements_list"] = parse_list_field(candidate.get("achievements", "")) | |
| for h in history: | |
| h["_matched_list"] = parse_list_field(h.get("matched_skills", "")) | |
| h["_missing_list"] = parse_list_field(h.get("missing_skills", "")) | |
| return {"candidate": candidate, "history": history} | |
| async def list_candidates(search: str = Query(default="")): | |
| rows = read_csv_safe(RESUME_CSV) | |
| names = [r.get("candidate_name", "") for r in rows if r.get("candidate_name")] | |
| if search: | |
| names = [n for n in names if search.lower() in n.lower()] | |
| return {"candidates": sorted(set(names))} | |
| async def get_ranked_cards( | |
| run: str = Query(default=""), | |
| jd: str = Query(default=""), | |
| tier: str = Query(default=""), | |
| ): | |
| """Return screening rows enriched with parsed interview questions and resume pros/cons.""" | |
| hist = read_csv_safe(SCREENING_CSV) | |
| resumes = {r.get("candidate_name", "").lower(): r for r in read_csv_safe(RESUME_CSV)} | |
| if run: hist = [r for r in hist if r.get("run_timestamp", "") == run] | |
| if jd: hist = [r for r in hist if r.get("jd_title", "") == jd] | |
| if tier: hist = [r for r in hist if r.get("confidence_tier","").lower() == tier.lower()] | |
| cards = [] | |
| for r in hist: | |
| name_key = r.get("candidate_name", "").lower() | |
| resume = resumes.get(name_key, {}) | |
| iq_raw = r.get("interview_questions", "[]") or "[]" | |
| try: | |
| iq_list = json.loads(iq_raw) if iq_raw.strip().startswith("[") else [] | |
| except Exception: | |
| iq_list = [] | |
| cards.append({ | |
| **r, | |
| "_matched_list": parse_list_field(r.get("matched_skills", "")), | |
| "_missing_list": parse_list_field(r.get("missing_skills", "")), | |
| "_pros_list": parse_list_field(resume.get("pros", "")), | |
| "_cons_list": parse_list_field(resume.get("cons", "")), | |
| "_interview_qs": iq_list, | |
| }) | |
| return {"cards": cards, "count": len(cards)} | |
| # ── Workspace Stats ─────────────────────────────────────────────────────────── | |
| async def get_stats(): | |
| resumes = len([f for f in RESUMES_DIR.iterdir() if f.is_file() and not f.name.startswith(".")]) | |
| jds = len([f for f in JDS_DIR.iterdir() if f.is_file() and not f.name.startswith(".")]) | |
| hist = read_csv_safe(SCREENING_CSV) | |
| runs = len(set(r.get("run_timestamp", "") for r in hist if r.get("run_timestamp"))) | |
| screened = len(hist) | |
| model_ok = (BASE / "models" / "skill_model.pkl").exists() | |
| main_ok = MAIN_PY.exists() | |
| return { | |
| "resumes": resumes, | |
| "jds": jds, | |
| "runs": runs, | |
| "screened": screened, | |
| "model_ok": model_ok, | |
| "main_ok": main_ok, | |
| } | |
| # ── Results Management ──────────────────────────────────────────────────────── | |
| async def clear_results(): | |
| """Delete screening_history.csv and resume_data.csv so stale data is gone.""" | |
| deleted = [] | |
| for path in [SCREENING_CSV, RESUME_CSV]: | |
| if path.exists(): | |
| path.unlink() | |
| deleted.append(path.name) | |
| return {"deleted": deleted, "message": f"Cleared {len(deleted)} file(s). Re-run pipeline to generate fresh results."} | |
| async def stale_check(): | |
| """ | |
| Detect whether the current results CSV is out of sync with files on disk. | |
| Uses resume_data.csv (ALL processed resumes) — not screening_history.csv | |
| (which only has the top-N ranked candidates) — so resumes ranked outside | |
| top-N are not falsely flagged as "not yet screened". | |
| Returns: { stale: bool, reason: str } | |
| """ | |
| # Need at least one run to exist | |
| hist = read_csv_safe(SCREENING_CSV) | |
| if not hist: | |
| return {"stale": False, "reason": "No results yet — run pipeline first."} | |
| # --- Use resume_data.csv to know which files have been processed --- | |
| resume_rows = read_csv_safe(RESUME_CSV) | |
| # resume_data.csv stores file_name; fall back to screening_history if empty | |
| if resume_rows: | |
| processed_files = set(r.get("file_name", "") for r in resume_rows if r.get("file_name")) | |
| else: | |
| processed_files = set(r.get("file_name", "") for r in hist if r.get("file_name")) | |
| disk_files = set( | |
| f.name for f in RESUMES_DIR.iterdir() | |
| if f.is_file() and not f.name.startswith(".") | |
| ) | |
| disk_jd_files = set( | |
| f.name for f in JDS_DIR.iterdir() | |
| if f.is_file() and not f.name.startswith(".") | |
| ) | |
| csv_jds = set(r.get("jd_title", "") for r in hist if r.get("jd_title")) | |
| # Resumes on disk that were never processed | |
| new_resumes = disk_files - processed_files | |
| # Resumes that were processed but the file has since been deleted | |
| removed_resumes = processed_files - disk_files | |
| # ── JD mismatch check ──────────────────────────────────────────────────── | |
| # Normalise both sides the same way so "Jr. Accountant JD.docx" matches | |
| # the stored title "Jr. Accountant": | |
| # 1. Strip file extension 2. Lowercase 3. Collapse separators → space | |
| # 4. Drop trailing "jd" / "job description" suffix 5. Strip whitespace | |
| import re as _re | |
| def _norm_jd(s: str) -> str: | |
| s = s.lower().strip() | |
| s = _re.sub(r"[_\-\.]+", " ", s) | |
| s = _re.sub(r"\s+jd$", "", s).strip() | |
| s = _re.sub(r"\s+job\s+description$", "", s).strip() | |
| s = _re.sub(r"\s+", " ", s).strip() | |
| return s | |
| disk_jd_stems_norm = {_norm_jd(f.rsplit(".", 1)[0]) for f in disk_jd_files} | |
| csv_jds_norm = {_norm_jd(j) for j in csv_jds} | |
| jd_mismatch = bool(disk_jd_files) and not bool(csv_jds_norm & disk_jd_stems_norm) | |
| stale = bool(new_resumes or removed_resumes or jd_mismatch) | |
| reasons = [] | |
| if new_resumes: | |
| reasons.append(f"{len(new_resumes)} new resume(s) not yet screened: {', '.join(sorted(new_resumes)[:3])}") | |
| if removed_resumes: | |
| reasons.append(f"{len(removed_resumes)} resume(s) deleted but still in results") | |
| if jd_mismatch and not new_resumes and not removed_resumes: | |
| reasons.append("Job descriptions may have changed since last run") | |
| return { | |
| "stale": stale, | |
| "reason": " · ".join(reasons) if reasons else "Results are up to date.", | |
| "new_resumes": sorted(new_resumes), | |
| "removed_resumes": sorted(removed_resumes), | |
| } | |
| # ── Mailing List ────────────────────────────────────────────────────────────── | |
| import uuid as _uuid | |
| import sys as _sys | |
| _sys.path.insert(0, str(BASE)) | |
| import config as _cfg | |
| MAILING_LIST_PATH = Path(getattr(_cfg, "MAILING_LIST_PATH", | |
| BASE / "outputs" / "match_results" / "mailing_list.json")) | |
| MAILING_LIST_PATH.parent.mkdir(parents=True, exist_ok=True) | |
| # Shortlist DB (separate from the screening pipeline tables). | |
| from engine import shortlist_db as _sldb # noqa: E402 | |
| _sldb.init_shortlist_db() | |
| # One-time import of legacy mailing_list.json entries (no-op after first run). | |
| _sldb.migrate_from_mailing_json(MAILING_LIST_PATH) | |
| def _read_mailing_list() -> list: | |
| if not MAILING_LIST_PATH.exists(): | |
| return [] | |
| try: | |
| with open(MAILING_LIST_PATH, encoding="utf-8") as f: | |
| data = json.load(f) | |
| return data if isinstance(data, list) else [] | |
| except (json.JSONDecodeError, OSError): | |
| return [] | |
| def _write_mailing_list(entries: list) -> None: | |
| with open(MAILING_LIST_PATH, "w", encoding="utf-8") as f: | |
| json.dump(entries, f, indent=2, ensure_ascii=False) | |
| def _dedupe_key(name: str, jd: str) -> tuple[str, str]: | |
| return (name.strip().lower(), jd.strip().lower()) | |
| async def list_mailing(): | |
| """Return all shortlisted candidates (newest first).""" | |
| entries = _read_mailing_list() | |
| entries.sort(key=lambda e: e.get("shortlisted_at", ""), reverse=True) | |
| return { | |
| "entries": entries, | |
| "count": len(entries), | |
| "default_subject": getattr(_cfg, "DEFAULT_EMAIL_SUBJECT", | |
| "Shortlisted for next round for the post of {job_role}"), | |
| "default_body": getattr(_cfg, "DEFAULT_EMAIL_BODY", ""), | |
| } | |
| async def add_to_mailing(request: Request): | |
| """ | |
| Append a candidate to the mailing list. | |
| Body: { candidate_name: str, jd_title: str (optional — defaults to "") } | |
| Server enriches the entry with email/phone (from resume_data.csv) and | |
| final_score/tier (from screening_history.csv) so the UI never re-queries. | |
| Dedupes on (candidate_name, jd_title); returns {added: bool, entry: …}. | |
| """ | |
| body = await request.json() | |
| name = (body.get("candidate_name") or "").strip() | |
| jd = (body.get("jd_title") or "").strip() | |
| if not name: | |
| raise HTTPException(status_code=400, detail="candidate_name is required") | |
| # Enrich from resume_data.csv (identity) | |
| resume = next( | |
| (r for r in read_csv_safe(RESUME_CSV) | |
| if (r.get("candidate_name") or "").lower() == name.lower()), | |
| {}, | |
| ) | |
| # Enrich from screening_history.csv (score + tier for that JD) | |
| hist = [r for r in read_csv_safe(SCREENING_CSV) | |
| if (r.get("candidate_name") or "").lower() == name.lower()] | |
| if jd: | |
| hist = [r for r in hist if r.get("jd_title") == jd] | |
| # If no JD passed, pick the candidate's top-scoring JD automatically | |
| def _sc(r): | |
| try: return float(r.get("final_score") or 0) | |
| except (TypeError, ValueError): return 0.0 | |
| hist.sort(key=_sc, reverse=True) | |
| best = hist[0] if hist else {} | |
| if not jd: | |
| jd = best.get("jd_title", "") or "" | |
| entries = _read_mailing_list() | |
| key = _dedupe_key(name, jd) | |
| existing = next((e for e in entries | |
| if _dedupe_key(e.get("candidate_name", ""), | |
| e.get("jd_title", "")) == key), None) | |
| if existing: | |
| return {"added": False, "entry": existing, | |
| "message": f"Already shortlisted for {jd or 'this role'}."} | |
| entry = { | |
| "id" : str(_uuid.uuid4()), | |
| "candidate_name" : name, | |
| "jd_title" : jd, | |
| "email" : (resume.get("email") or "").strip(), | |
| "phone" : (resume.get("phone") or "").strip(), | |
| "final_score" : float(best.get("final_score") or 0), | |
| "confidence_tier" : best.get("confidence_tier", ""), | |
| "shortlisted_at" : datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| } | |
| entries.append(entry) | |
| _write_mailing_list(entries) | |
| # Also write to the shortlist DB so the "Shortlist & Decisions" page | |
| # picks up entries shortlisted via the legacy candidate-card button. | |
| try: | |
| _sldb.add_to_shortlist( | |
| { | |
| "candidate_name": name, | |
| "jd_title": jd, | |
| "resume_id": resume.get("resume_id"), | |
| "email": resume.get("email", ""), | |
| "phone": resume.get("phone", ""), | |
| "final_score": float(best.get("final_score") or 0), | |
| "confidence_tier": best.get("confidence_tier", ""), | |
| "hirer_remarks": "", | |
| }, | |
| actor="mailing_list", | |
| ) | |
| except (ValueError, Exception): | |
| pass # never fail the mailing-list write because of shortlist sync | |
| return {"added": True, "entry": entry, | |
| "message": f"Added {name} to the mailing list."} | |
| async def remove_from_mailing(entry_id: str): | |
| entries = _read_mailing_list() | |
| target = next((e for e in entries if e.get("id") == entry_id), None) | |
| new_entries = [e for e in entries if e.get("id") != entry_id] | |
| if len(new_entries) == len(entries): | |
| raise HTTPException(status_code=404, detail="Entry not found") | |
| _write_mailing_list(new_entries) | |
| # Mirror the removal in the shortlist DB. Match on (name, jd_title) | |
| # since the JSON entry ID is not the shortlist_id. | |
| if target: | |
| for row in _sldb.list_shortlist(jd_title=target.get("jd_title") or None): | |
| if (row.get("candidate_name") or "").lower() == \ | |
| (target.get("candidate_name") or "").lower(): | |
| _sldb.delete_shortlist(row["shortlist_id"]) | |
| break | |
| return {"deleted": entry_id, "count": len(new_entries)} | |
| # ============================================================================= | |
| # SHORTLIST ENDPOINTS — recruiter decisions + hirer remarks | |
| # ============================================================================= | |
| async def shortlist_list(jd: str = Query("", description="Filter by JD title"), | |
| status: str = Query("", description="Filter by status")): | |
| """List all shortlisted candidates with status / stage / remarks.""" | |
| entries = _sldb.list_shortlist(jd_title=jd or None, status=status or None) | |
| return { | |
| "entries": entries, | |
| "count": len(entries), | |
| "valid_status": sorted(_sldb.VALID_STATUS), | |
| "valid_stages": sorted(_sldb.VALID_STAGES), | |
| } | |
| async def shortlist_add(request: Request): | |
| """ | |
| Add a candidate to the shortlist. | |
| Body: { candidate_name, jd_title?, hirer_remarks? } | |
| Auto-enriches email/phone from resume_data.csv and score/tier from | |
| screening_history.csv (best score across JDs if jd_title is empty). | |
| """ | |
| body = await request.json() | |
| name = (body.get("candidate_name") or "").strip() | |
| jd = (body.get("jd_title") or "").strip() | |
| if not name: | |
| raise HTTPException(status_code=400, detail="candidate_name is required") | |
| resume = next( | |
| (r for r in read_csv_safe(RESUME_CSV) | |
| if (r.get("candidate_name") or "").lower() == name.lower()), | |
| {}, | |
| ) | |
| hist = [r for r in read_csv_safe(SCREENING_CSV) | |
| if (r.get("candidate_name") or "").lower() == name.lower()] | |
| if jd: | |
| hist = [r for r in hist if r.get("jd_title") == jd] | |
| def _sc(r): | |
| try: return float(r.get("final_score") or 0) | |
| except (TypeError, ValueError): return 0.0 | |
| hist.sort(key=_sc, reverse=True) | |
| best = hist[0] if hist else {} | |
| if not jd: | |
| jd = best.get("jd_title", "") or "" | |
| try: | |
| result = _sldb.add_to_shortlist( | |
| { | |
| "candidate_name": name, | |
| "jd_title": jd, | |
| "resume_id": resume.get("resume_id"), | |
| "email": resume.get("email", ""), | |
| "phone": resume.get("phone", ""), | |
| "final_score": _sc(best), | |
| "confidence_tier": best.get("confidence_tier", ""), | |
| "hirer_remarks": body.get("hirer_remarks", ""), | |
| }, | |
| actor=body.get("actor", "user"), | |
| ) | |
| except ValueError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) | |
| return result | |
| async def shortlist_update(shortlist_id: str, request: Request): | |
| """ | |
| Update status, stage, remarks, or any decision field. | |
| Body keys: status, current_stage, rejection_stage, rejection_reason, | |
| selection_reason, hirer_remarks, interviewer, decided_by, actor | |
| """ | |
| body = await request.json() | |
| actor = body.pop("actor", "user") if isinstance(body, dict) else "user" | |
| try: | |
| row = _sldb.update_shortlist(shortlist_id, body, actor=actor) | |
| except KeyError: | |
| raise HTTPException(status_code=404, detail="Shortlist entry not found") | |
| except ValueError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) | |
| return {"updated": True, "entry": row} | |
| async def shortlist_delete(shortlist_id: str): | |
| ok = _sldb.delete_shortlist(shortlist_id) | |
| if not ok: | |
| raise HTTPException(status_code=404, detail="Shortlist entry not found") | |
| return {"deleted": shortlist_id} | |
| async def shortlist_events(shortlist_id: str): | |
| """Audit trail for one shortlist row (status changes, remarks, stage moves).""" | |
| if not _sldb.get_shortlist(shortlist_id): | |
| raise HTTPException(status_code=404, detail="Shortlist entry not found") | |
| return {"events": _sldb.get_events(shortlist_id)} | |
| # ============================================================================= | |
| # API KEY MANAGEMENT — per-user, stored at OS-standard config location | |
| # ============================================================================= | |
| # Which keys we expose to the UI and the order we prefer at runtime | |
| _API_KEY_NAMES = ("ANTHROPIC_API_KEY", "GROQ_API_KEY", "OPENAI_API_KEY") | |
| def _mask(value: str) -> str: | |
| """Return a masked form like 'sk-…aB12' for display.""" | |
| if not value: | |
| return "" | |
| if len(value) <= 8: | |
| return "•" * len(value) | |
| return value[:3] + "…" + value[-4:] | |
| async def api_keys_get(): | |
| """ | |
| Return which keys are set + a masked preview. Never returns the raw key. | |
| Tells the UI where the .env file lives so a power user can edit it. | |
| """ | |
| from engine import extractor as _ext | |
| # Re-read user .env so the response reflects the on-disk state | |
| _ext._load_user_api_keys() | |
| keys = {} | |
| for name in _API_KEY_NAMES: | |
| v = os.environ.get(name, "") | |
| keys[name] = {"set": bool(v), "preview": _mask(v) if v else ""} | |
| return { | |
| "keys": keys, | |
| "active_provider": _ext._llm_extractor_model or "", | |
| "llm_ready": bool(_ext._LLM_EXTRACTOR_OK), | |
| "config_dir": _ext.user_config_dir(), | |
| "env_path": _ext.user_env_path(), | |
| } | |
| async def api_keys_set(request: Request): | |
| """ | |
| Save one or more API keys to the per-user .env file and reload the | |
| extractor's LLM client so the new key takes effect immediately — | |
| no server restart needed. | |
| Body: { "ANTHROPIC_API_KEY": "sk-...", "GROQ_API_KEY": "gsk_...", ... } | |
| Empty string for a key clears it. | |
| """ | |
| body = await request.json() | |
| if not isinstance(body, dict): | |
| raise HTTPException(status_code=400, detail="expected a JSON object") | |
| from engine import extractor as _ext | |
| # 1. Read existing .env file into a dict | |
| env_path = _ext.user_env_path() | |
| cfg_dir = _ext.user_config_dir() | |
| os.makedirs(cfg_dir, exist_ok=True) | |
| existing: dict[str, str] = {} | |
| if os.path.exists(env_path): | |
| with open(env_path, encoding="utf-8") as f: | |
| for line in f: | |
| line = line.strip() | |
| if not line or line.startswith("#") or "=" not in line: | |
| continue | |
| k, _, v = line.partition("=") | |
| existing[k.strip()] = v.strip().strip('"').strip("'") | |
| # 2. Apply patch from body — only known key names | |
| changed = [] | |
| for name in _API_KEY_NAMES: | |
| if name in body: | |
| new_val = (body.get(name) or "").strip() | |
| if new_val: | |
| existing[name] = new_val | |
| os.environ[name] = new_val | |
| changed.append(name) | |
| else: | |
| # Empty string → unset | |
| existing.pop(name, None) | |
| os.environ.pop(name, None) | |
| changed.append(f"-{name}") | |
| # 3. Write the file back atomically (write to tmp, rename) | |
| tmp = env_path + ".tmp" | |
| with open(tmp, "w", encoding="utf-8") as f: | |
| f.write("# Resume Analyzer — per-user API keys\n") | |
| f.write("# Generated by Settings UI. Edit only if you know what you're doing.\n\n") | |
| for k, v in existing.items(): | |
| f.write(f"{k}={v}\n") | |
| os.replace(tmp, env_path) | |
| # Tighten file permissions on POSIX (Windows ACL via LOCALAPPDATA is enough) | |
| try: | |
| os.chmod(env_path, 0o600) | |
| except OSError: | |
| pass | |
| # 4. Re-init the LLM client so the new key takes effect right now | |
| _ext._LLM_EXTRACTOR_OK = False | |
| _ext._llm_extractor_client = None | |
| _ext._llm_extractor_model = "" | |
| _ext._init_llm_extractor() | |
| return { | |
| "saved": True, | |
| "changed": changed, | |
| "env_path": env_path, | |
| "llm_ready": bool(_ext._LLM_EXTRACTOR_OK), | |
| "active_provider": _ext._llm_extractor_model or "", | |
| } | |
| async def candidate_jds(name: str): | |
| """Return all JDs a candidate has been scored against, ordered by score.""" | |
| hist = [r for r in read_csv_safe(SCREENING_CSV) | |
| if (r.get("candidate_name") or "").lower() == name.lower()] | |
| def _sc(r): | |
| try: return float(r.get("final_score") or 0) | |
| except (TypeError, ValueError): return 0.0 | |
| hist.sort(key=_sc, reverse=True) | |
| seen, jds = set(), [] | |
| for r in hist: | |
| j = r.get("jd_title", "") | |
| if j and j not in seen: | |
| seen.add(j) | |
| jds.append({ | |
| "jd_title": j, | |
| "final_score": _sc(r), | |
| "confidence_tier": r.get("confidence_tier", ""), | |
| }) | |
| return {"name": name, "jds": jds} | |
| # ── Parameter Availability ─────────────────────────────────────────────────── | |
| # Maps weight-slider key → field-presence test against a resume row. | |
| # A slider is "available" when ≥1 resume on disk has data for that field. | |
| def _resume_field_filled(row: dict, key: str) -> bool: | |
| """True iff this resume row carries meaningful data for the given key.""" | |
| if key == "skill_match": | |
| return bool((row.get("technical_skills") or "").strip()) | |
| if key == "experience": | |
| try: | |
| return float(row.get("total_experience_years") or 0) > 0 | |
| except (TypeError, ValueError): | |
| return False | |
| if key == "education": | |
| return bool((row.get("education_degree") or "").strip()) | |
| if key == "keyword_alignment": | |
| # Available whenever at least one JD + one resume exist | |
| return True | |
| if key == "recency": | |
| return bool((row.get("recency_flag") or "").strip()) | |
| if key == "project_relevance": | |
| return bool(parse_list_field(row.get("projects", ""))) | |
| if key == "achievement_impact": | |
| return bool(parse_list_field(row.get("achievements", ""))) | |
| if key == "resume_quality": | |
| return bool((row.get("resume_summary") or "").strip()) or bool(row.get("email")) | |
| if key == "career_consistency": | |
| try: | |
| return float(row.get("avg_tenure_months") or 0) > 0 | |
| except (TypeError, ValueError): | |
| return False | |
| return False | |
| async def parameter_availability(): | |
| """ | |
| For each of the 9 weight parameters, report whether at least one resume on | |
| disk has data for it. The frontend uses this to grey out unusable sliders. | |
| Falls back to "all available" when no resumes have been processed yet so | |
| users can still pick weights before their first run. | |
| """ | |
| keys = [ | |
| "skill_match", "experience", "education", "keyword_alignment", "recency", | |
| "project_relevance", "achievement_impact", "resume_quality", "career_consistency", | |
| ] | |
| rows = read_csv_safe(RESUME_CSV) | |
| has_jd = any( | |
| f.is_file() and not f.name.startswith(".") | |
| for f in JDS_DIR.iterdir() | |
| ) | |
| # Before any pipeline run, leave every slider enabled so weights can be set | |
| # for the first run. | |
| if not rows: | |
| return { | |
| "available": {k: True for k in keys}, | |
| "candidate_count": 0, | |
| "has_jd": has_jd, | |
| } | |
| avail = {} | |
| for k in keys: | |
| if k == "keyword_alignment": | |
| avail[k] = has_jd | |
| continue | |
| avail[k] = any(_resume_field_filled(r, k) for r in rows) | |
| return { | |
| "available": avail, | |
| "candidate_count": len(rows), | |
| "has_jd": has_jd, | |
| } | |
| # ── Candidate Comparison ────────────────────────────────────────────────────── | |
| async def comparison_candidates(jd: str = Query(default="")): | |
| """List candidates that have been screened against a JD (or any JD).""" | |
| rows = read_csv_safe(SCREENING_CSV) | |
| if jd: | |
| rows = [r for r in rows if r.get("jd_title") == jd] | |
| # Preserve rank order: dedupe while keeping the best-scoring instance | |
| seen = {} | |
| for r in rows: | |
| n = r.get("candidate_name") | |
| if not n: | |
| continue | |
| try: | |
| sc = float(r.get("final_score") or 0) | |
| except (TypeError, ValueError): | |
| sc = 0.0 | |
| if n not in seen or sc > seen[n]: | |
| seen[n] = sc | |
| return { | |
| "candidates": sorted(seen.keys(), key=lambda n: -seen[n]), | |
| "jds": sorted(set(r.get("jd_title", "") for r in read_csv_safe(SCREENING_CSV) if r.get("jd_title"))), | |
| } | |
| async def comparison_row(name: str = Query(...), jd: str = Query(default="")): | |
| """ | |
| Return the screening row + extracted resume fields for one candidate (best | |
| JD by default, or the specified jd_title). | |
| """ | |
| hist = read_csv_safe(SCREENING_CSV) | |
| hist = [r for r in hist if r.get("candidate_name", "").lower() == name.lower()] | |
| if jd: | |
| hist = [r for r in hist if r.get("jd_title") == jd] | |
| if not hist: | |
| raise HTTPException(status_code=404, detail=f"No screening row for {name}") | |
| # Pick highest scoring row when several available | |
| def _sc(r): | |
| try: | |
| return float(r.get("final_score") or 0) | |
| except (TypeError, ValueError): | |
| return 0.0 | |
| hist.sort(key=_sc, reverse=True) | |
| row = hist[0] | |
| resume_row = next( | |
| (r for r in read_csv_safe(RESUME_CSV) | |
| if r.get("candidate_name", "").lower() == name.lower()), | |
| {}, | |
| ) | |
| row["_matched_list"] = parse_list_field(row.get("matched_skills", "")) | |
| row["_missing_list"] = parse_list_field(row.get("missing_skills", "")) | |
| row["_resume"] = resume_row | |
| return {"row": row} | |
| # ── Config Editor ───────────────────────────────────────────────────────────── | |
| async def get_config(): | |
| cfg = BASE / "config.py" | |
| if not cfg.exists(): | |
| raise HTTPException(status_code=404, detail="config.py not found") | |
| return {"content": cfg.read_text()} | |
| async def save_config(request: Request): | |
| body = await request.json() | |
| content = body.get("content", "") | |
| cfg = BASE / "config.py" | |
| cfg.write_text(content) | |
| return {"saved": True} | |
| # ─── DEV ENTRY POINT ────────────────────────────────────────────────────────── | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True) |