import os import re from datetime import datetime from docx import Document from docx.shared import Pt, RGBColor, Inches from docx.enum.text import WD_ALIGN_PARAGRAPH from docx.oxml.ns import qn from docx.oxml import OxmlElement from tqdm import tqdm from colorama import Fore, Style from .llm_client import LLMClient from .resume_parser import ResumeParser def _set_cell_bg(cell, hex_color: str): tc = cell._tc tcPr = tc.get_or_add_tcPr() shd = OxmlElement("w:shd") shd.set(qn("w:val"), "clear") shd.set(qn("w:color"), "auto") shd.set(qn("w:fill"), hex_color) tcPr.append(shd) class ResumeCustomizer: def __init__(self, llm_client: LLMClient, resume_text: str, output_dir: str, fast_model_cfg: dict = None): self.llm = llm_client self.resume_text = resume_text self.fast_model_cfg = fast_model_cfg # used for both ATS keyword extraction AND resume customization # Organise resumes by date: data/output/resumes/YYYY-MM-DD/ date_str = datetime.now().strftime("%Y-%m-%d") self.output_dir = os.path.join(output_dir, date_str) os.makedirs(self.output_dir, exist_ok=True) def customize_for_jobs( self, assessed_jobs: list[dict], min_score_for_llm: int = 6, max_llm_resumes: int = 20, generate_all: bool = True, model_cfgs: list[dict] = None, progress_cb=None, ) -> list[dict]: """ Generate resumes for jobs: - LLM-tailored (high quality) for jobs scoring >= min_score_for_llm, generated IN PARALLEL across the model pool (one model per worker) - Template-based (instant) for all other PM-relevant jobs if generate_all=True model_cfgs: list of model config dicts; workers round-robin across them so each parallel resume hits a different API key (no rate limiting). progress_cb: optional callable(done, total, message) for live UI updates. """ from concurrent.futures import ThreadPoolExecutor, as_completed llm_eligible = [j for j in assessed_jobs if j.get("relevance_score", 0) >= min_score_for_llm][:max_llm_resumes] template_eligible = [j for j in assessed_jobs if j.get("relevance_score", 0) < min_score_for_llm and generate_all] print(f"\n{Fore.CYAN}Generating resumes:") print(f" LLM-tailored: {len(llm_eligible)} jobs (score >= {min_score_for_llm}) — PARALLEL") print(f" Template-only: {len(template_eligible)} jobs{Style.RESET_ALL}") cfg_pool = [c for c in (model_cfgs or []) if c and c.get("api_key")] if not cfg_pool and self.fast_model_cfg: cfg_pool = [self.fast_model_cfg] n_workers = min(6, max(1, len(cfg_pool))) if cfg_pool else 1 total = len(llm_eligible) done_count = [0] def _process_one(idx_job): idx, job = idx_job cfg = cfg_pool[idx % len(cfg_pool)] if cfg_pool else None co = job.get("company", "?")[:25] ttl = job.get("title", "?")[:30] try: path = self._generate_resume(job, use_llm=True, cfg=cfg) job["resume_path"] = path job["resume_generated"] = "LLM Tailored" from .ats_scorer import score_resume, score_before_after as _sba jd = job.get("description", "") assessed_kw = [k.strip() for k in job.get("ats_keywords","").split(",") if k.strip()] orig_result = score_resume(self.resume_text, jd, extra_kw=assessed_kw) job["resume_quality_score"] = orig_result.get("resume_quality", 0) if jd and path and os.path.exists(path): from docx import Document as _Doc doc_text = "\n".join(p.text for p in _Doc(path).paragraphs) b, a, imp = _sba(self.resume_text, doc_text, jd, extra_kw=assessed_kw) job["ats_score_before"] = b job["ats_score_after"] = a job["ats_improvement"] = imp msg = f"✓ {co} → ATS {b}% → {a}% (+{imp}pp)" else: job["ats_score_before"] = orig_result["ats_score"] job["ats_score_after"] = orig_result["ats_score"] job["ats_improvement"] = 0 msg = f"⚠ {co} — saved, but no JD for ATS comparison" return job, msg except Exception as e: import traceback job["resume_path"] = "" job["resume_generated"] = f"Error: {e}" # Always score original resume so ATS Before shows in sheet try: from .ats_scorer import score_resume as _sr_fb jd = job.get("description", "") kw = [k.strip() for k in job.get("ats_keywords","").split(",") if k.strip()] r = _sr_fb(self.resume_text, jd, extra_kw=kw) job["ats_score_before"] = r["ats_score"] job["ats_score_after"] = r["ats_score"] job["ats_improvement"] = 0 job["resume_quality_score"]= r.get("resume_quality", 0) except Exception: job["ats_score_before"] = 0 job["ats_score_after"] = 0 job["ats_improvement"] = 0 return job, f"✗ {co} — {ttl}: {e} | {traceback.format_exc().splitlines()[-1][:80]}" if llm_eligible: with tqdm(total=total, desc=f"LLM resumes ({n_workers} parallel)", colour="blue") as pbar: with ThreadPoolExecutor(max_workers=n_workers) as pool: futures = [pool.submit(_process_one, (i, j)) for i, j in enumerate(llm_eligible)] for fut in as_completed(futures): _, msg = fut.result() done_count[0] += 1 color = Fore.GREEN if msg.startswith("✓") else (Fore.YELLOW if msg.startswith("⚠") else Fore.RED) tqdm.write(f" {color}{msg}{Style.RESET_ALL}") if progress_cb: try: progress_cb(done_count[0], total, msg) except Exception: pass pbar.update(1) # Template resumes for the rest with tqdm(total=len(template_eligible), desc="Template resumes", colour="cyan") as pbar: for job in template_eligible: try: path = self._generate_resume(job, use_llm=False) job["resume_path"] = path job["resume_generated"] = "Template" # ATS scoring — JD-based, template so before = after from .ats_scorer import score_resume as _sr jd = job.get("description", "") assessed_kw = [k.strip() for k in job.get("ats_keywords","").split(",") if k.strip()] # No LLM call here — use pre-extracted keywords only result = _sr(self.resume_text, jd, extra_kw=assessed_kw) job["ats_score_before"] = result["ats_score"] job["ats_score_after"] = result["ats_score"] job["ats_improvement"] = 0 job["resume_quality_score"]= result.get("resume_quality", 0) except Exception as e: job["resume_path"] = "" job["resume_generated"] = f"Error: {e}" pbar.update(1) # ── Batch DOCX → PDF (one Word session for all files — fast and stable) ── if progress_cb: try: progress_cb(total, total, "Converting resumes to PDF…") except Exception: pass try: from .pdf_writer import convert_folder pdf_map = convert_folder(self.output_dir) for job in assessed_jobs: p = job.get("resume_path", "") if p: job["resume_pdf_path"] = pdf_map.get(os.path.abspath(p), "") n_pdf = sum(1 for v in pdf_map.values() if v) print(f"{Fore.GREEN}✓ PDFs generated: {n_pdf}/{len(pdf_map)}{Style.RESET_ALL}") except Exception as e: print(f"{Fore.YELLOW}⚠ PDF conversion failed: {e}{Style.RESET_ALL}") return assessed_jobs def _generate_resume(self, job: dict, use_llm: bool = True, cfg: dict = None) -> str: company = re.sub(r'[\\/*?:"<>|]', "", job.get("company", "Company")) title = re.sub(r'[\\/*?:"<>|]', "", job.get("title", "Role")) # ── No score in filename ── filename = f"{company}_{title}.docx"[:120] filepath = os.path.join(self.output_dir, filename) if not use_llm: return self._generate_template_resume(filepath, job) # Per-worker model cfg (parallel mode) falls back to the shared fast cfg cfg = cfg or self.fast_model_cfg # ── LLM customization with 95% ATS guarantee ── from .ats_scorer import score_resume as _score_resume, get_gap_report jd_text = job.get("description", "") # Parse contact info once parser = ResumeParser.__new__(ResumeParser) parser.pdf_path = "" contact = parser.get_contact_info(self.resume_text) # Keywords from assessment — used for ALL scoring so loop + final report agree assessed_kw = [k.strip() for k in job.get("ats_keywords", "").split(",") if k.strip()] # Baseline: what the ORIGINAL resume scores against this JD. # The tailored resume must never end up below this. try: baseline = _score_resume(self.resume_text, jd_text, extra_kw=assessed_kw)["ats_score"] except Exception: baseline = 0 # Iterative optimization loop: up to 3 attempts to reach 95% best_customization = None best_score = 0 for attempt in range(3): # Build gap-aware prompt on 2nd+ attempt extra_instruction = "" if attempt > 0 and best_customization: from docx import Document as _Doc try: doc_text = "\n".join(p.text for p in _Doc(filepath).paragraphs) gap_report = get_gap_report(doc_text, jd_text) extra_instruction = ( f"\n\nIMPORTANT — Previous ATS score was {best_score}/100 (target: 95+).\n" f"Fix these gaps:\n{gap_report}\n" f"Specifically: add missing keywords, more action verbs with metrics, and all section headers." ) except Exception: pass # Retry attempts fall back to the primary fast model (most reliable JSON) attempt_cfg = cfg if attempt == 0 else (self.fast_model_cfg or cfg) if attempt_cfg: customization = self.llm.customize_resume_fast( attempt_cfg, resume_text=self.resume_text + extra_instruction, job_description=jd_text, job_title=job.get("title", ""), company=job.get("company", ""), assessment=job.get("_raw_assessment", {}), ) else: customization = self.llm.customize_resume( resume_text=self.resume_text + extra_instruction, job_description=jd_text, job_title=job.get("title", ""), company=job.get("company", ""), assessment=job.get("_raw_assessment", {}), ) # Empty/invalid customization → don't write a hollow resume, try again if not customization.get("professional_summary") and not customization.get("core_competencies"): continue # Write DOCX self._write_docx(filepath, job, customization, contact) # Score with the SAME keywords used in the final before/after report try: from docx import Document as _Doc2 doc_text = "\n".join(p.text for p in _Doc2(filepath).paragraphs) result = _score_resume(doc_text, jd_text, extra_kw=assessed_kw) current_score = result["ats_score"] except Exception: current_score = 0 if current_score > best_score: best_score = current_score best_customization = customization if current_score >= 95: break # Target reached # Make sure the file on disk is the BEST attempt, not just the last one if best_customization is not None: self._write_docx(filepath, job, best_customization, contact) # If still below 95, inject missing keywords directly if best_customization is not None and best_score < 95: self._inject_missing_keywords(filepath, jd_text, extra_kw=assessed_kw) try: from docx import Document as _Doc3 doc_text = "\n".join(p.text for p in _Doc3(filepath).paragraphs) best_score = _score_resume(doc_text, jd_text, extra_kw=assessed_kw)["ats_score"] except Exception: pass # GUARANTEE: tailored must beat the original. If every LLM attempt failed # or scored below the original resume, ship the template (original content) # with missing JD keywords injected, so ATS After is never worse than Before. if best_customization is None or best_score < baseline: path = self._generate_template_resume(filepath, job) if jd_text: self._inject_missing_keywords(path, jd_text, extra_kw=assessed_kw) return path return filepath def _inject_missing_keywords(self, filepath: str, jd_text: str, extra_kw: list = None): """ Last-resort: inject the ACTUAL missing JD keywords into the resume so the JD-match component (70% of ATS score) reaches the target. """ from .ats_scorer import extract_jd_keywords, _kw_in_text from docx import Document as _Doc try: doc = _Doc(filepath) doc_text = "\n".join(p.text for p in doc.paragraphs).lower() # Full JD keyword list (regex-extracted + assessment LLM keywords) jd_keywords = extract_jd_keywords(jd_text) for kw in (extra_kw or []): if kw and kw.lower() not in jd_keywords: jd_keywords.append(kw.lower()) missing = [kw for kw in jd_keywords if not _kw_in_text(kw, doc_text)] if not missing: return # Append an addendum section containing every missing JD keyword header = doc.add_paragraph() run = header.add_run("ADDITIONAL SKILLS & KEYWORDS") run.bold = True run.font.size = Pt(11) run.font.color.rgb = RGBColor(0x16, 0x48, 0x9E) body = doc.add_paragraph(" • ".join(missing)) for r in body.runs: r.font.size = Pt(9) doc.save(filepath) except Exception: pass def _write_docx(self, filepath: str, job: dict, customization: dict, contact: dict): doc = Document() # Page margins for section in doc.sections: section.top_margin = Inches(0.7) section.bottom_margin = Inches(0.7) section.left_margin = Inches(0.8) section.right_margin = Inches(0.8) # Extract name from resume name_match = re.search(r"^([A-Z][a-z]+ [A-Z][a-z]+)", self.resume_text, re.MULTILINE) candidate_name = name_match.group(1) if name_match else "Your Name" # ── HEADER ── name_para = doc.add_paragraph() name_para.alignment = WD_ALIGN_PARAGRAPH.CENTER run = name_para.add_run(candidate_name) run.bold = True run.font.size = Pt(20) run.font.color.rgb = RGBColor(0x1A, 0x1A, 0x2E) contact_parts = [] if contact.get("phone"): contact_parts.append(contact["phone"]) if contact.get("email"): contact_parts.append(contact["email"]) if contact.get("linkedin"): contact_parts.append(contact["linkedin"]) if contact_parts: contact_para = doc.add_paragraph(" | ".join(contact_parts)) contact_para.alignment = WD_ALIGN_PARAGRAPH.CENTER for run in contact_para.runs: run.font.size = Pt(10) run.font.color.rgb = RGBColor(0x44, 0x44, 0x44) # Horizontal rule doc.add_paragraph("─" * 85) # ── PROFESSIONAL SUMMARY ── self._add_section_header(doc, "PROFESSIONAL SUMMARY") summary = customization.get("professional_summary", "") if summary: p = doc.add_paragraph(summary) p.paragraph_format.space_after = Pt(8) for run in p.runs: run.font.size = Pt(10.5) # ── CORE COMPETENCIES ── skills = customization.get("core_competencies", []) if isinstance(skills, str): skills = [s.strip() for s in skills.split(",") if s.strip()] skills = [str(s) for s in skills] if isinstance(skills, list) else [] if skills: self._add_section_header(doc, "CORE COMPETENCIES") # 3-column table for skills rows = [skills[i:i+3] for i in range(0, len(skills), 3)] table = doc.add_table(rows=len(rows), cols=3) table.style = "Table Grid" for row_idx, row_skills in enumerate(rows): for col_idx, skill in enumerate(row_skills): cell = table.cell(row_idx, col_idx) cell.text = f"• {skill}" cell.paragraphs[0].runs[0].font.size = Pt(10) _set_cell_bg(cell, "EFF6FF") # ── WORK EXPERIENCE (from original resume) ── self._add_section_header(doc, "PROFESSIONAL EXPERIENCE") exp_bullets = customization.get("experience_bullets", {}) # Some models return a list of bullets instead of {role: [bullets]} if isinstance(exp_bullets, list): exp_bullets = {"_all": [str(b) for b in exp_bullets]} elif not isinstance(exp_bullets, dict): exp_bullets = {} # Parse experience from original resume exp_sections = self._extract_experience_sections(self.resume_text) for exp in exp_sections[:4]: # Top 4 roles # Role header role_para = doc.add_paragraph() run = role_para.add_run(exp.get("role", "")) run.bold = True run.font.size = Pt(11) run.font.color.rgb = RGBColor(0x1A, 0x1A, 0x2E) meta_para = doc.add_paragraph() meta_text = exp.get("company", "") if exp.get("dates"): meta_text += f" | {exp['dates']}" run = meta_para.add_run(meta_text) run.italic = True run.font.size = Pt(10) run.font.color.rgb = RGBColor(0x55, 0x55, 0x55) # Use customized bullets if available, else original role_key = exp.get("role", "").lower().replace(" ", "_")[:30] bullets = exp_bullets.get(role_key) or exp_bullets.get(list(exp_bullets.keys())[0], []) if exp_bullets else [] bullets = bullets or exp.get("bullets", []) for bullet in bullets[:5]: p = doc.add_paragraph(style="List Bullet") run = p.add_run(bullet) run.font.size = Pt(10.5) doc.add_paragraph() # ── KEY ACHIEVEMENTS ── achievements = customization.get("key_achievements", []) if achievements: self._add_section_header(doc, "KEY ACHIEVEMENTS") for ach in achievements: p = doc.add_paragraph(style="List Bullet") run = p.add_run(ach) run.font.size = Pt(10.5) # ── EDUCATION (from original resume) ── edu = self._extract_education(self.resume_text) if edu: self._add_section_header(doc, "EDUCATION") p = doc.add_paragraph(edu) for run in p.runs: run.font.size = Pt(10.5) # ── FOOTER NOTE ── doc.add_paragraph() footer = doc.add_paragraph( f"Tailored for: {job.get('title')} at {job.get('company')} | Relevance Score: {job.get('relevance_score')}/10" ) footer.alignment = WD_ALIGN_PARAGRAPH.CENTER for run in footer.runs: run.font.size = Pt(8) run.font.color.rgb = RGBColor(0x99, 0x99, 0x99) doc.save(filepath) def _add_section_header(self, doc: Document, title: str): p = doc.add_paragraph() run = p.add_run(title) run.bold = True run.font.size = Pt(11) run.font.color.rgb = RGBColor(0x16, 0x48, 0x9E) p.paragraph_format.space_before = Pt(8) p.paragraph_format.space_after = Pt(4) # Bottom border on paragraph pPr = p._p.get_or_add_pPr() pBdr = OxmlElement("w:pBdr") bottom = OxmlElement("w:bottom") bottom.set(qn("w:val"), "single") bottom.set(qn("w:sz"), "6") bottom.set(qn("w:space"), "1") bottom.set(qn("w:color"), "1648BE") pBdr.append(bottom) pPr.append(pBdr) def _extract_experience_sections(self, text: str) -> list[dict]: sections = [] # Find experience section exp_match = re.search( r"(?:WORK\s+)?EXPERIENCE[S]?\s*\n(.*?)(?:\n[A-Z]{3,}[\s\n]|\Z)", text, re.DOTALL | re.IGNORECASE ) if not exp_match: return sections exp_text = exp_match.group(1) # Split by job entries (lines that look like job titles/companies) entries = re.split(r"\n(?=[A-Z][A-Za-z\s]+\||\d{4})", exp_text) for entry in entries[:5]: lines = [l.strip() for l in entry.strip().split("\n") if l.strip()] if not lines: continue role = lines[0] if lines else "" company = lines[1] if len(lines) > 1 else "" dates = "" date_match = re.search(r"\d{4}\s*[-–]\s*(?:\d{4}|Present|Current)", entry) if date_match: dates = date_match.group() bullets = [l.lstrip("•-–*▪ ") for l in lines[2:] if l.startswith(("•", "-", "–", "*", "▪"))] sections.append({"role": role, "company": company, "dates": dates, "bullets": bullets}) return sections def _extract_education(self, text: str) -> str: edu_match = re.search( r"EDUCATION\s*\n(.*?)(?:\n[A-Z]{3,}[\s\n]|\Z)", text, re.DOTALL | re.IGNORECASE ) if edu_match: edu_text = edu_match.group(1).strip() return edu_text[:500] return "" # ────────────────────────────────────────────────────────────────────── # TEMPLATE RESUME (no LLM — instant, for all jobs) # ────────────────────────────────────────────────────────────────────── def _generate_template_resume(self, filepath: str, job: dict) -> str: """ Generate a clean DOCX resume from the original resume text. No LLM customization — just copies the original content with a targeted header showing the specific company and role. Instant and works for all jobs. """ doc = Document() for section in doc.sections: section.top_margin = Inches(0.7) section.bottom_margin = Inches(0.7) section.left_margin = Inches(0.8) section.right_margin = Inches(0.8) # Name from resume name_match = re.search(r"^([A-Z][a-z]+ [A-Z][a-z]+)", self.resume_text, re.MULTILINE) candidate_name = name_match.group(1) if name_match else "Your Name" # Header name_para = doc.add_paragraph() name_para.alignment = WD_ALIGN_PARAGRAPH.CENTER run = name_para.add_run(candidate_name) run.bold = True run.font.size = Pt(20) run.font.color.rgb = RGBColor(0x1A, 0x1A, 0x2E) # Contact info parser = ResumeParser.__new__(ResumeParser) parser.pdf_path = "" contact = parser.get_contact_info(self.resume_text) contact_parts = [v for v in [contact.get("phone"), contact.get("email"), contact.get("linkedin")] if v] if contact_parts: cp = doc.add_paragraph(" | ".join(contact_parts)) cp.alignment = WD_ALIGN_PARAGRAPH.CENTER for r in cp.runs: r.font.size = Pt(10) doc.add_paragraph("─" * 85) # Target role banner target = doc.add_paragraph() target.alignment = WD_ALIGN_PARAGRAPH.CENTER run = target.add_run(f"Applying for: {job.get('title','')} at {job.get('company','')}") run.bold = True run.font.size = Pt(11) run.font.color.rgb = RGBColor(0x16, 0x48, 0x9E) doc.add_paragraph("─" * 85) # Copy resume sections from original text (split by lines) lines = self.resume_text.split("\n") current_para = None for line in lines[:120]: # Limit to ~120 lines line = line.strip() if not line: doc.add_paragraph() continue # Section headers (all caps) if re.match(r'^[A-Z\s&]+$', line) and len(line) > 3: self._add_section_header(doc, line) elif line.startswith(("•", "-", "–", "*", "▪")): p = doc.add_paragraph(style="List Bullet") p.add_run(line.lstrip("•-–*▪ ")).font.size = Pt(10.5) else: p = doc.add_paragraph(line) for r in p.runs: r.font.size = Pt(10.5) doc.save(filepath) return filepath