""" Validation batch runner (spec #10) — shared by the 20-job CLI script and the UI "Run N-job validation" button. Generates resumes for a set of jobs using the production provider FALLBACK CHAIN (config.LLM_GENERATION.provider_order), exports files, re-parses them, scores internal + independent, and returns a batch table plus optional validation packages. Model-independent: the same code runs against Kimi / NVIDIA / Claude / Stub depending on configured keys. """ from __future__ import annotations import os import re from datetime import datetime from typing import List, Optional from .resume_model import Resume def _status_risk_level(report: dict) -> str: n_high = len(report.get("high_risk_terms_for_confirmation", []) or []) n_med = len(report.get("review_terms_for_user_review", []) or []) if n_high: return "HIGH" if n_med: return "MEDIUM" return "LOW" def run_validation_batch(jobs: List[dict], base_resume: Resume = None, llm=None, output_dir: str = None, build_packages: bool = True, progress_cb=None, chain: list = None) -> dict: """Run the provider chain on each job; return {rows, packages, summary}. `chain` overrides the configured provider chain (used by offline tests to force a StubProvider-only chain).""" from .providers import build_provider_chain from .resume_customizer import ResumeCustomizer, _read_docx_text from .validation_package import build_validation_package if base_resume is None: from .provider_eval import load_base_resume base_resume = load_base_resume() if base_resume is None: return {"error": "No base resume (data/resume/_parsed.json or resume.pdf) available."} if output_dir is None: output_dir = os.path.join("data", "output", "validation_runs", datetime.now().strftime("%Y-%m-%d_%H-%M-%S")) os.makedirs(output_dir, exist_ok=True) if chain is None: chain = build_provider_chain(llm) chain_names = " -> ".join(getattr(p, "name", "?") for p in chain) cust = ResumeCustomizer.__new__(ResumeCustomizer) cust.llm = llm cust.resume_text = base_resume.to_flat_text() cust.fast_model_cfg = None cust.output_dir = output_dir rows: List[dict] = [] packages: List[str] = [] total = len(jobs) for i, job in enumerate(jobs): safe = re.sub(r'[\\/*?:"<>|]', "", f"{job.get('company','Co')}_{job.get('title','Role')}")[:100] filepath = os.path.join(output_dir, safe + ".docx") try: cust._run_provider_chain(job, filepath, chain, base_resume_override=base_resume) job["resume_path"] = filepath except Exception as e: job["_v2_report"] = {"status": f"ERROR:{str(e)[:40]}"} job["status"] = "ERROR" report = job.get("_v2_report", {}) or {} est = report.get("estimated_scores", {}) or {} # Before/after ATS (original vs generated) for the table. try: from .ats_scorer import score_resume, conservative_display_score before = conservative_display_score( score_resume(cust.resume_text, job.get("description", ""))["ats_score"]) except Exception: before = 0 row = { "job_title": job.get("title", ""), "company": job.get("company", ""), "platform": job.get("platform", job.get("source", "fixture")), "provider_used": report.get("provider_used", job.get("provider_used", "")), "status": report.get("status", job.get("status", "")), "internal_score": est.get("jd_match", 0), "independent_score": report.get("independent_jd_match", job.get("independent_jd_match", 0)), "ats_readability": est.get("ats_readability", 0), "ats_before": before, "risk_level": _status_risk_level(report), "review_terms": len(report.get("review_terms_for_user_review", []) or []), "repair_attempts": len(report.get("repair_attempts", []) or []), "download_allowed": bool(report.get("download_allowed")), "file_link": job.get("resume_path", ""), "provider_attempts": report.get("provider_attempts", []), } rows.append(row) if build_packages and job.get("resume_path") and os.path.exists(job["resume_path"]): try: packages.append(build_validation_package(job, cust.resume_text)) except Exception: pass if progress_cb: try: progress_cb(i + 1, total, row) except Exception: pass ready = sum(1 for r in rows if r["download_allowed"]) summary = { "chain": chain_names, "total": total, "ready": ready, "ready_rate": round(100 * ready / total) if total else 0, "needs_user_input": sum(1 for r in rows if r["status"] == "NEEDS_USER_INPUT"), "blocked": sum(1 for r in rows if not r["download_allowed"]), "output_dir": output_dir, } return {"rows": rows, "packages": packages, "summary": summary} def select_jobs(n: int = 20, live: bool = False, llm=None) -> List[dict]: """Select N jobs for validation. live=True scrapes/assesses real jobs via the pipeline; otherwise (default, offline-safe) builds from JD fixtures.""" if live: jobs = _select_live_jobs(n, llm) if jobs: return jobs[:n] # Offline: JD fixtures, cycled up to n from .provider_eval import load_jd_fixtures fixtures = load_jd_fixtures() if not fixtures: return [] out: List[dict] = [] while len(out) < n and fixtures: for jd in fixtures: out.append(dict(jd)) if len(out) >= n: break return out[:n] def _select_live_jobs(n: int, llm=None) -> List[dict]: """Best-effort live job selection using the existing scraper stack.""" try: from config import JOB_SEARCH, PLATFORMS jobs: List[dict] = [] if PLATFORMS.get("linkedin"): from .scrapers.linkedin import LinkedInScraper sc = LinkedInScraper() for role in JOB_SEARCH["roles"][:2]: for loc in JOB_SEARCH["locations"][:2]: try: for j in sc.search(role, loc, max_results=10): if sc.is_pm_role(j.title): jobs.append({"title": j.title, "company": j.company, "location": j.location, "platform": "LinkedIn", "url": j.url, "description": j.description or "", "ats_keywords": "", "_raw_assessment": {}}) if len(jobs) >= n: return jobs except Exception: continue return jobs except Exception: return []