""" Jobalytics paste -> regenerate flow (spec #8). The user pastes the "missing keywords" a real external checker (Jobalytics) reported. We: 1. Classify each keyword against the original JD, the parsed final resume, the candidate vault, and risk severity -> one of: already_present_parser_issue add_to_skills add_to_experience_bullet add_to_summary medium_review_auto_include high_risk_needs_confirmation blocked_do_not_include 2. Regenerate the resume in external_checker_mode = "jobalytics_repair" via the provider (model-independent), then DETERMINISTICALLY enforce placement so it works even with a weak/stub model. Important terms are distributed across Summary, Skills, and Experience — never dumped into Skills. 3. Re-render, re-parse, re-score (internal + independent), and report before/after keyword coverage + scores. This module is model-independent: any provider produces the regeneration; the deterministic layer + scorers decide whether it actually improved. """ from __future__ import annotations import os import re import json from datetime import datetime from typing import Dict, List, Optional from .resume_model import Resume from .ats_scorer import _kw_in_text # Decision vocabulary (spec #8) ALREADY_PRESENT = "already_present_parser_issue" ADD_SKILLS = "add_to_skills" ADD_EXPERIENCE = "add_to_experience_bullet" ADD_SUMMARY = "add_to_summary" MEDIUM_AUTO = "medium_review_auto_include" HIGH_CONFIRM = "high_risk_needs_confirmation" BLOCKED = "blocked_do_not_include" def classify_jobalytics_keywords(pasted_keywords: List[str], jd_text: str, base_resume_text: str, final_resume_text: str, maximum_ats_mode: bool = False, confirmed_terms: List[str] = None) -> List[dict]: """Classify each pasted keyword into a placement decision (spec #8). In Maximum ATS Mode, normal PM/Product/AI/SaaS/B2B/agile vocabulary is treated as user-confirmed (LOW risk) — see `candidate_fit._is_max_ats_safe`. Hard blocks (credentials/seniority/employers/engineering) are unaffected. """ from .jd_analyzer import analyze_jd, Requirement, _categorize from .candidate_fit import ( classify_fit, infer_seniority, severity, _candidate_years, LOW, MEDIUM, HIGH, BLOCKED as SEV_BLOCKED, ) from .candidate_vault import user_confirmed_terms, user_blocked_terms jd_low = (jd_text or "").lower() final_low = (final_resume_text or "").lower() req = analyze_jd(jd_text) req_by_term = {r.term.lower(): r for r in req.all_requirements()} seniority = infer_seniority(base_resume_text or "") cand_years = _candidate_years(base_resume_text or "") try: confirmed, blocked = user_confirmed_terms(), user_blocked_terms() except Exception: confirmed, blocked = set(), set() if confirmed_terms: confirmed = set(confirmed) | {t.lower().strip() for t in confirmed_terms} confirmed -= set(blocked or set()) rows: List[dict] = [] for kw in pasted_keywords: kw = (kw or "").strip() if not kw: continue kl = kw.lower() # Already in the EXPORTED resume -> external checker parser miss. if _kw_in_text(kw, final_low): rows.append({"keyword": kw, "in_jd": kl in jd_low or kl in req_by_term, "decision": ALREADY_PRESENT, "reason": "already present in the exported resume text"}) continue r = req_by_term.get(kl) or Requirement(term=kw, category=_categorize(kl)) v = classify_fit(r, base_resume_text or "", seniority=seniority, confirmed=confirmed, blocked=blocked, candidate_years=cand_years, maximum_ats_mode=maximum_ats_mode) sev = severity(v) if v.action == "block" or sev == SEV_BLOCKED: decision = BLOCKED elif v.action == "ask_user" or sev == HIGH: decision = HIGH_CONFIRM elif sev == MEDIUM: decision = MEDIUM_AUTO else: # LOW / explicit / adjacent -> choose a natural placement placement = [p.lower() for p in (v.recommended_placement or [])] if "experience" in placement or r.category in ("responsibility", "hard_skill"): decision = ADD_EXPERIENCE elif "summary" in placement: decision = ADD_SUMMARY else: decision = ADD_SKILLS rows.append({ "keyword": kw, "in_jd": kl in jd_low or kl in req_by_term, "category": r.category, "fit_status": v.fit_status, "severity": sev, "decision": decision, "reason": v.reason, "placement": v.recommended_placement, }) return rows def _placement_guidance(rows: List[dict]) -> str: return "\n".join(f"- {r['keyword']}: {r['decision']}" for r in rows) def _coverage(keywords: List[str], text: str) -> dict: low = (text or "").lower() present = [k for k in keywords if _kw_in_text(k, low)] total = len(keywords) or 1 return {"present": present, "missing": [k for k in keywords if k not in present], "covered": len(present), "total": len(keywords), "pct": round(100 * len(present) / total)} def regenerate_from_jobalytics(job: dict, pasted_keywords: List[str], llm=None, provider=None, base_resume: Resume = None, output_dir: str = None, maximum_ats_mode: bool = False, confirmed_terms: List[str] = None) -> dict: """Regenerate `job`'s resume to address pasted Jobalytics missing keywords. Returns a result dict with classifications, before/after coverage, scores, status, and the new resume filepath. Never fabricates: BLOCKED terms are skipped, HIGH_CONFIRM terms are only added when already supported/confirmed. Maximum ATS Mode treats normal PM/AI/SaaS craft terms as user-confirmed and flows that flag into the generation pipeline (via the job dict). """ from .resume_customizer import _read_docx_text, ResumeCustomizer jd_text = job.get("description", "") or "" job_title = job.get("title", "") company = job.get("company", "") # Base resume: explicit > parsed cache > real PDF. if base_resume is None: base_resume = _load_base_resume(job) if base_resume is None: return {"error": "no base resume available for regeneration"} base_text = base_resume.to_flat_text() # Current (pre-repair) exported text for before-coverage + classification. cur_path = job.get("resume_path", "") final_text_before = "" if cur_path and os.path.exists(cur_path): try: final_text_before = _read_docx_text(cur_path) except Exception: final_text_before = "" if not final_text_before: final_text_before = base_text rows = classify_jobalytics_keywords(pasted_keywords, jd_text, base_text, final_text_before, maximum_ats_mode=maximum_ats_mode, confirmed_terms=confirmed_terms) before = _coverage(pasted_keywords, final_text_before) # Addable terms = everything except BLOCKED and unconfirmed HIGH_CONFIRM. confirmed_high = _confirmed_terms() | {t.lower().strip() for t in (confirmed_terms or [])} addable = [r["keyword"] for r in rows if r["decision"] in (ADD_SKILLS, ADD_EXPERIENCE, ADD_SUMMARY, MEDIUM_AUTO)] addable += [r["keyword"] for r in rows if r["decision"] == HIGH_CONFIRM and r["keyword"].lower() in confirmed_high] if provider is None: provider = _first_provider(llm) # Run the FULL proven v4 pipeline (deterministic fit-expansion + repair loop # reaches ~90 and honestly excludes risky/blocked terms), but in # external_checker_mode='jobalytics_repair': the provider uses the jobalytics # prompt and the pasted addable keywords are merged into the include pool so # they get woven/placed. This keeps the existing tailoring quality AND # addresses the external checker's keywords. out_dir = output_dir or os.path.dirname(cur_path) or os.path.join( "data", "output", "resumes", datetime.now().strftime("%Y-%m-%d")) os.makedirs(out_dir, exist_ok=True) new_path = os.path.join(out_dir, _safe_name(company, job_title) + "_jobalytics.docx") job2 = dict(job) existing_kw = [k.strip() for k in (job.get("ats_keywords", "") or "").split(",") if k.strip()] job2["ats_keywords"] = ", ".join(dict.fromkeys(existing_kw + addable)) job2["_jobalytics_keywords"] = addable job2["_jobalytics_placement"] = _placement_guidance(rows) job2["_maximum_ats_mode"] = maximum_ats_mode job2["_confirmed_terms"] = (confirmed_terms or []) + addable cust = ResumeCustomizer.__new__(ResumeCustomizer) cust.llm = llm cust.resume_text = base_text cust.fast_model_cfg = None cust.output_dir = out_dir try: cust._generate_resume_v4(job2, cfg=None, filepath=new_path, provider=provider, base_resume_override=base_resume) except Exception as e: return {"error": f"regeneration failed: {e}", "classifications": rows, "before_coverage": before} report = job2.get("_v2_report", {}) or {} est = report.get("estimated_scores", {}) or {} parsed_after = _read_docx_text(new_path) if os.path.exists(new_path) else "" after = _coverage(pasted_keywords, parsed_after) return { "classifications": rows, "before_coverage": before, "after_coverage": after, "scores": { "internal_jd_match": est.get("jd_match", 0), "independent_jd_match": report.get("independent_jd_match", 0), "ats_readability": est.get("ats_readability", 0), }, "status": report.get("status", job2.get("_v2_status", "")), "download_allowed": bool(report.get("download_allowed")), "provider_used": report.get("provider_used", getattr(provider, "name", "")), "provider_response_quality": report.get("provider_response_quality", ""), "resume_path": new_path, "decisions_summary": _decision_counts(rows), } # ── External ATS feedback (paste from Jobalytics/Simplify) ─────────────────── def parse_external_feedback(text: str) -> dict: """Best-effort parse of pasted external-checker feedback into {external_score, missing_keywords, matched_keywords, notes}. Lenient: handles a raw comma/newline list of keywords, or a fuller panel paste with 'missing'/'matched' sections and an 'NN%' score. Never fabricates — it only reads what the user pasted. """ text = text or "" out = {"external_score": None, "missing_keywords": [], "matched_keywords": [], "notes": ""} m = re.search(r"(\d{1,3})\s*%", text) if m: try: v = int(m.group(1)) if 0 <= v <= 100: out["external_score"] = v except ValueError: pass def _split(chunk: str) -> List[str]: parts = re.split(r"[\n,;•·|]+", chunk) seen, terms = set(), [] for p in parts: t = re.sub(r"^[\s\-\*✓✗>]+", "", p).strip() t = re.sub(r"\s*\(\d+\)\s*$", "", t).strip() # drop trailing counts t = t.strip(" .;:•-").strip() # drop trailing punctuation tl = t.lower() if 2 <= len(t) <= 40 and tl not in seen and not t.endswith(":"): seen.add(tl) terms.append(t) return terms low = text.lower() # Try to isolate a "missing" section; else treat the whole paste as missing. miss_idx = re.search(r"missing\b[^:\n]*:?", low) match_idx = re.search(r"matched\b[^:\n]*:?|present\b[^:\n]*:?", low) if miss_idx: start = miss_idx.end() end = match_idx.start() if (match_idx and match_idx.start() > start) else len(text) out["missing_keywords"] = _split(text[start:end]) if match_idx: out["matched_keywords"] = _split(text[match_idx.end():]) else: # No explicit section header — assume the whole paste is a keyword list, # but drop obvious feedback-prose tokens so a prose paste with no real # keywords ("great resume, no notes") yields nothing rather than junk. _PROSE = {"resume", "great", "good", "no", "not", "notes", "note", "nice", "looks", "look", "strong", "weak", "section", "sections", "score", "match", "overall", "your", "the", "is", "are", "was", "improve", "add", "present", "missing", "keyword", "keywords"} out["missing_keywords"] = [ t for t in _split(text) if not (set(re.findall(r"[a-z]+", t.lower())) & _PROSE) ] return out def build_coverage_report(rows: List[dict], missing_keywords: List[str], before: dict, after: dict, confirmed_high: set) -> dict: """Rich per-keyword ATS coverage report (spec: debug/reporting). For every pasted external keyword: its category, risk classification, inclusion decision, target resume section, and (for excluded terms) why. Plus before/after coverage and a category breakdown. """ after_present = {k.lower() for k in after.get("present", [])} _SECTION = { ADD_SUMMARY: "Summary", ADD_SKILLS: "Skills", ADD_EXPERIENCE: "Experience bullets", MEDIUM_AUTO: "Skills + Experience", ALREADY_PRESENT: "Already present", } _DISPOSITION = { ADD_SKILLS: "included_auto", ADD_EXPERIENCE: "included_auto", ADD_SUMMARY: "included_auto", MEDIUM_AUTO: "included_review", HIGH_CONFIRM: "needs_confirmation", BLOCKED: "blocked", ALREADY_PRESENT: "already_present", } keywords: List[dict] = [] by_category: Dict[str, int] = {} for r in rows: kw = r["keyword"] decision = r["decision"] disposition = _DISPOSITION.get(decision, "review") if decision == HIGH_CONFIRM and kw.lower() in confirmed_high: disposition = "user_confirmed" placed = kw.lower() in after_present cat = r.get("category", "unknown") by_category[cat] = by_category.get(cat, 0) + 1 keywords.append({ "keyword": kw, "category": cat, "in_jd": r.get("in_jd", False), "risk": r.get("severity", ""), "disposition": disposition, "placed_in_resume": placed, "section": _SECTION.get(decision, "—") if placed or decision == ALREADY_PRESENT else "(not placed)", "reason": r.get("reason", ""), }) return { "external_keywords_total": len(missing_keywords), "before": {"covered": before.get("covered", 0), "total": before.get("total", 0), "pct": before.get("pct", 0)}, "after": {"covered": after.get("covered", 0), "total": after.get("total", 0), "pct": after.get("pct", 0)}, "coverage_count": f"{after.get('covered', 0)}/{after.get('total', 0)}", "by_category": by_category, "keywords": keywords, } def repair_with_external_feedback(job: dict, feedback_text: str = None, missing_keywords: List[str] = None, external_score: int = None, llm=None, provider=None, base_resume: Resume = None, output_dir: str = None, maximum_ats_mode: bool = False, confirmed_terms: List[str] = None, target_external_score: int = None) -> dict: """External ATS Feedback Repair Mode. Re-tailors `job`'s resume to address externally-reported missing keywords (Jobalytics/Simplify), honestly: every keyword goes through candidate_fit risk classification; BLOCKED and unconfirmed HIGH-risk terms are NEVER added. Scores come from the re-parsed exported file. In Maximum ATS Mode, normal PM/Product/AI/SaaS/B2B/agile vocabulary is treated as user-confirmed and aggressively woven across sections, and the loop keeps repairing toward `target_external_score` (default from config) as long as the remaining gaps are LOW/MEDIUM. Statuses: READY_MAX_ATS_95_PLUS / READY_90_PLUS_EXTERNAL_ALIGNED / READY_95_EXTERNAL_ALIGNED / NEEDS_USER_CONFIRMATION / BELOW_TARGET_REPAIRABLE / base status. """ from .fit_gate import ( READY, READY_REVIEW, READY_95_EXTERNAL_ALIGNED, READY_MAX_ATS_95_PLUS, READY_90_PLUS_EXTERNAL_ALIGNED, NEEDS_USER_CONFIRMATION, BELOW_TARGET_REPAIRABLE, NEEDS_USER_INPUT, ) try: from config import MAXIMUM_ATS as _MAX_CFG except Exception: _MAX_CFG = {"target_external_score": 95, "min_external_score": 90, "max_repair_iterations": 4} if target_external_score is None: target_external_score = _MAX_CFG.get("target_external_score", 95) min_external = _MAX_CFG.get("min_external_score", 90) max_iters = _MAX_CFG.get("max_repair_iterations", 4) if maximum_ats_mode else 1 parsed_fb = parse_external_feedback(feedback_text) if feedback_text else {} if missing_keywords is None: missing_keywords = parsed_fb.get("missing_keywords", []) if external_score is None: external_score = parsed_fb.get("external_score") if not missing_keywords: return {"error": "no_missing_keywords", "detail": "Could not find any missing keywords in the pasted feedback."} # Iterate the honest re-tailor until external-style coverage hits the target # or only blocked/high-risk terms remain (max mode); single pass otherwise. result = None for _ in range(max(1, max_iters)): result = regenerate_from_jobalytics( job, missing_keywords, llm=llm, provider=provider, base_resume=base_resume, output_dir=output_dir, maximum_ats_mode=maximum_ats_mode, confirmed_terms=confirmed_terms) if "error" in result: return result after_pct = result.get("after_coverage", {}).get("pct", 0) rows = result.get("classifications", []) # Stop if target reached, or remaining missing are only high/blocked. still_missing = set(k.lower() for k in result.get("after_coverage", {}).get("missing", [])) repairable_left = [r for r in rows if r["keyword"].lower() in still_missing and r["decision"] in (ADD_SKILLS, ADD_EXPERIENCE, ADD_SUMMARY, MEDIUM_AUTO)] if after_pct >= target_external_score or not repairable_left: break rows = result.get("classifications", []) confirmed_high = _confirmed_terms() | {t.lower().strip() for t in (confirmed_terms or [])} after_cov = result.get("after_coverage", {}) before_cov = result.get("before_coverage", {}) after_present = set(k.lower() for k in after_cov.get("present", [])) added = [r["keyword"] for r in rows if r["decision"] in (ADD_SKILLS, ADD_EXPERIENCE, ADD_SUMMARY, MEDIUM_AUTO) and r["keyword"].lower() in after_present] review_flags = [r["keyword"] for r in rows if r["decision"] == MEDIUM_AUTO] unresolved_high = [r["keyword"] for r in rows if r["decision"] == HIGH_CONFIRM and r["keyword"].lower() not in confirmed_high] blocked = [r["keyword"] for r in rows if r["decision"] == BLOCKED] already_present = [r["keyword"] for r in rows if r["decision"] == ALREADY_PRESENT] # Terms still missing that ARE repairable (LOW/MEDIUM) — should be empty in # max mode at the end; if not, we stayed BELOW_TARGET_REPAIRABLE. still_missing = set(k.lower() for k in after_cov.get("missing", [])) still_missing_repairable = [r["keyword"] for r in rows if r["keyword"].lower() in still_missing and r["decision"] in (ADD_SKILLS, ADD_EXPERIENCE, ADD_SUMMARY, MEDIUM_AUTO)] base_status = result.get("status", "") after_pct = after_cov.get("pct", 0) internal = result.get("scores", {}).get("internal_jd_match", 0) independent = result.get("scores", {}).get("independent_jd_match", 0) readability = result.get("scores", {}).get("ats_readability", 0) gates_pass = (base_status in (READY, READY_REVIEW) and internal >= 90 and independent >= 90 and readability >= 90) # External coverage measure: prefer the pasted external score if it's higher # confidence; otherwise our re-parsed coverage of the pasted gaps. ext_measure = max(after_pct, external_score or 0) if gates_pass and ext_measure >= target_external_score and not review_flags: final_status = READY_MAX_ATS_95_PLUS elif gates_pass and ext_measure >= min_external: final_status = (READY_95_EXTERNAL_ALIGNED if not maximum_ats_mode else READY_90_PLUS_EXTERNAL_ALIGNED) elif still_missing_repairable: # Below target but the remaining gaps are LOW/MEDIUM → keep repairing. final_status = BELOW_TARGET_REPAIRABLE elif unresolved_high: # Only high-risk-but-supportable terms remain → ask the user to confirm. final_status = NEEDS_USER_CONFIRMATION if maximum_ats_mode else NEEDS_USER_INPUT else: final_status = base_status coverage_report = build_coverage_report(rows, missing_keywords, before_cov, after_cov, confirmed_high) # Plain-English explanation for the UI when we're below 90. explanation = "" if final_status not in (READY_MAX_ATS_95_PLUS, READY_90_PLUS_EXTERNAL_ALIGNED, READY_95_EXTERNAL_ALIGNED, READY, READY_REVIEW): bits = [] if internal < 90: bits.append(f"internal {internal} < 90") if independent < 90: bits.append(f"independent {independent} < 90") if readability < 90: bits.append(f"readability {readability} < 90") if ext_measure < min_external: bits.append(f"external coverage {ext_measure}% < {min_external}%") if unresolved_high: bits.append(f"{len(unresolved_high)} term(s) need your confirmation: " + ", ".join(unresolved_high[:6])) if blocked: bits.append(f"{len(blocked)} blocked (won't fake): " + ", ".join(blocked[:6])) explanation = "; ".join(bits) result.update({ "status": final_status, "external_score": external_score, "target_external_score": target_external_score, "maximum_ats_mode": maximum_ats_mode, "added_terms": added, "review_flag_terms": review_flags, "unresolved_high_risk_terms": unresolved_high, "blocked_terms": blocked, "already_present_terms": already_present, "still_missing_repairable": still_missing_repairable, "missing_keywords_input": missing_keywords, "coverage_report": coverage_report, "below_target_explanation": explanation, }) return result # ── helpers ────────────────────────────────────────────────────────────────── def _safe_name(company: str, title: str) -> str: base = f"{company}_{title}" return re.sub(r'[\\/*?:"<>|]', "", base)[:100] or "resume" def _decision_counts(rows: List[dict]) -> dict: out: dict = {} for r in rows: out[r["decision"]] = out.get(r["decision"], 0) + 1 return out def _load_base_resume(job: dict) -> Optional[Resume]: # 1. Parsed cache cache = os.path.join("data", "resume", "_parsed.json") if os.path.exists(cache): try: with open(cache, encoding="utf-8") as f: data = json.load(f) return Resume.from_dict(data.get("resume", data)) except Exception: pass # 2. Real PDF pdf = os.path.join("data", "resume", "resume.pdf") if os.path.exists(pdf): try: from .resume_parser_v2 import parse_resume_pdf_cached return parse_resume_pdf_cached(pdf) except Exception: pass return None def _first_provider(llm=None): from .providers import build_provider_chain chain = build_provider_chain(llm) return chain[0] def _confirmed_terms() -> set: try: from .candidate_vault import user_confirmed_terms return user_confirmed_terms() except Exception: return set()