import copy import json import re import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from docx import Document from docx.oxml import OxmlElement from docx.table import _Cell, Table from docx.text.paragraph import Paragraph from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet from reportlab.lib.units import cm from reportlab.pdfbase import pdfmetrics from reportlab.pdfbase.ttfonts import TTFont from reportlab.platypus import Paragraph as PdfParagraph, SimpleDocTemplate, Spacer BANNED_PHRASE_PATTERNS = [ re.compile(r"\bif accurate\b", re.I), re.compile(r"\bif relevant\b", re.I), re.compile(r"\bif applicable\b", re.I), re.compile(r"\badd a bullet\b", re.I), re.compile(r"\bsuggested\b", re.I), re.compile(r"\brecommendation\b", re.I), re.compile(r"\bevidence-backed\b", re.I), re.compile(r"\bconsider adding\b", re.I), re.compile(r"\bimprove proof of\b", re.I), re.compile(r"\btailor this\b", re.I), re.compile(r"\bthe cv should\b", re.I), re.compile(r"\bthe candidate should\b", re.I), re.compile(r"\bmanual review\b", re.I), re.compile(r"\bneeds review\b", re.I), re.compile(r"\bno evidence found\b", re.I), re.compile(r"\bedit plan\b", re.I), re.compile(r"\bplanner\b", re.I), re.compile(r"\bcoaching\b", re.I), re.compile(r"\banalysis\b", re.I), re.compile(r"\bapply recommendations?\b", re.I), re.compile(r"\brecruiter-ready\b", re.I), re.compile(r"\bאם זה נכון\b"), re.compile(r"\bאם רלוונטי\b"), re.compile(r"\bאם קיים\b"), re.compile(r"\bאם קיימת\b"), re.compile(r"\bמומלץ\b"), re.compile(r"\bהמלצ"), re.compile(r"\bהוסף\b"), re.compile(r"\bלהוסיף\b"), re.compile(r"\bיש להוסיף\b"), re.compile(r"\bכדאי להוסיף\b"), re.compile(r"\bשפר\b"), re.compile(r"\bחזק\b"), re.compile(r"\bהמועמד צריך\b"), re.compile(r"\bקורות החיים צריכים\b"), re.compile(r"\bלא נמצאה ראיה\b"), re.compile(r"\bנדרשת בדיקה\b"), re.compile(r"\bבדיקה ידנית\b"), re.compile(r"\bsi es cierto\b", re.I), re.compile(r"\bsi aplica\b", re.I), re.compile(r"\bse recomienda\b", re.I), re.compile(r"\bagregue\b", re.I), re.compile(r"\bañada\b", re.I), re.compile(r"\brequiere revisión\b", re.I), re.compile(r"\bsi c'est vrai\b", re.I), re.compile(r"\bil est recommandé\b", re.I), re.compile(r"\bajoutez\b", re.I), re.compile(r"\bfalls zutreffend\b", re.I), re.compile(r"\bempfehlung\b", re.I), re.compile(r"\bfügen sie hinzu\b", re.I), ] BANNED_LINE_PATTERNS = [ re.compile(r"^if\b", re.I), re.compile(r"^suggested\b", re.I), re.compile(r"^recommendation\b", re.I), re.compile(r"^consider\b", re.I), re.compile(r"^tailor\b", re.I), re.compile(r"^the cv should\b", re.I), re.compile(r"^the candidate should\b", re.I), ] UNICODE_FONT_NAME = "PulseUnicode" BOLD_UNICODE_FONT_NAME = "PulseUnicodeBold" def register_unicode_fonts() -> Tuple[str, str]: regular_candidates = [ Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"), Path("C:/Windows/Fonts/arial.ttf"), ] bold_candidates = [ Path("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"), Path("C:/Windows/Fonts/arialbd.ttf"), ] regular = next((candidate for candidate in regular_candidates if candidate.exists()), None) bold = next((candidate for candidate in bold_candidates if candidate.exists()), None) if not regular: return "Helvetica", "Helvetica-Bold" if UNICODE_FONT_NAME not in pdfmetrics.getRegisteredFontNames(): pdfmetrics.registerFont(TTFont(UNICODE_FONT_NAME, str(regular))) if bold and BOLD_UNICODE_FONT_NAME not in pdfmetrics.getRegisteredFontNames(): pdfmetrics.registerFont(TTFont(BOLD_UNICODE_FONT_NAME, str(bold))) return UNICODE_FONT_NAME, BOLD_UNICODE_FONT_NAME if bold else UNICODE_FONT_NAME def normalize_text(value: str) -> str: return re.sub(r"\s+", " ", value or "").strip().lower() def set_paragraph_text(paragraph: Paragraph, text: str) -> None: style = paragraph.style first_run_style = None if paragraph.runs: first = paragraph.runs[0] first_run_style = { "bold": first.bold, "italic": first.italic, "underline": first.underline, "font_name": first.font.name, "font_size": first.font.size, } p = paragraph._element for child in list(p): if child.tag.endswith("}pPr"): continue p.remove(child) run = paragraph.add_run(text) paragraph.style = style if first_run_style: run.bold = first_run_style["bold"] run.italic = first_run_style["italic"] run.underline = first_run_style["underline"] if first_run_style["font_name"]: run.font.name = first_run_style["font_name"] if first_run_style["font_size"]: run.font.size = first_run_style["font_size"] def insert_paragraph_after(paragraph: Paragraph, text: str) -> Paragraph: new_p = OxmlElement("w:p") if paragraph._p.pPr is not None: new_p.append(copy.deepcopy(paragraph._p.pPr)) paragraph._p.addnext(new_p) new_para = Paragraph(new_p, paragraph._parent) try: new_para.style = paragraph.style except Exception: pass run = new_para.add_run(text) if paragraph.runs: ref = paragraph.runs[0] run.bold = ref.bold run.italic = ref.italic run.underline = ref.underline if ref.font.name: run.font.name = ref.font.name if ref.font.size: run.font.size = ref.font.size return new_para def paragraph_contains(paragraph: Paragraph, target: str) -> bool: return normalize_text(target) in normalize_text(paragraph.text) def iter_table_paragraphs(table: Table) -> List[Paragraph]: paragraphs: List[Paragraph] = [] for row in table.rows: for cell in row.cells: paragraphs.extend(iter_cell_paragraphs(cell)) return paragraphs def iter_cell_paragraphs(cell: _Cell) -> List[Paragraph]: paragraphs: List[Paragraph] = list(cell.paragraphs) for table in cell.tables: paragraphs.extend(iter_table_paragraphs(table)) return paragraphs def iter_document_paragraphs(document: Document) -> List[Paragraph]: """Return editable paragraphs from body, tables, headers, and footers.""" paragraphs: List[Paragraph] = list(document.paragraphs) for table in document.tables: paragraphs.extend(iter_table_paragraphs(table)) for section in document.sections: paragraphs.extend(section.header.paragraphs) paragraphs.extend(section.footer.paragraphs) for table in section.header.tables: paragraphs.extend(iter_table_paragraphs(table)) for table in section.footer.tables: paragraphs.extend(iter_table_paragraphs(table)) return paragraphs def find_paragraph_by_text(document: Document, target: str) -> Optional[Paragraph]: for paragraph in iter_document_paragraphs(document): if paragraph_contains(paragraph, target): return paragraph return None def all_document_text(document: Document) -> str: return "\n".join(p.text for p in iter_document_paragraphs(document) if p.text) def extract_contact_lines(text: str) -> List[str]: lines = [line.strip() for line in text.splitlines() if line.strip()] contacts = [] for line in lines[:8]: if "@" in line or re.search(r"\+?\d[\d\-\s]{7,}", line): contacts.append(line) return contacts def sanitize_recruiter_text(value: str) -> Tuple[str, List[str], bool]: warnings: List[str] = [] if not value or not value.strip(): return "", warnings, True kept_lines: List[str] = [] for raw_line in value.splitlines(): line = raw_line.strip() if not line: continue banned = any(pattern.search(line) for pattern in BANNED_LINE_PATTERNS) or any(pattern.search(line) for pattern in BANNED_PHRASE_PATTERNS) if banned: warnings.append(f"Removed internal instruction line: {line}") continue kept_lines.append(line) text = "\n".join(kept_lines).strip() rejected = not text or any(pattern.search(text) for pattern in BANNED_PHRASE_PATTERNS) if rejected: warnings.append("Rejected contaminated recruiter-facing text before writing output.") return text, warnings, rejected def validate_final_cv_output(text: str) -> None: violations = [pattern.pattern for pattern in BANNED_PHRASE_PATTERNS if pattern.search(text or "")] if violations: raise ValueError(f"Final CV output contains banned internal instruction language: {', '.join(violations)}") def build_redline(report_path: Path, instructions: List[Dict[str, Any]], before_after: List[Dict[str, str]], warnings: List[str]) -> None: lines = ["# Pulse CV Change Report", ""] if warnings: lines.append("## Warnings") lines.extend([f"- {warning}" for warning in warnings]) lines.append("") lines.append("## Applied changes") lines.append("") for index, instruction in enumerate(instructions, start=1): delta = next((item for item in before_after if item["id"] == instruction["id"]), None) lines.append(f"### {index}. {instruction['sectionLabel']} — {instruction['action']}") lines.append(f"- Rationale: {instruction['rationale']}") if instruction.get("linkedRequirementIds"): lines.append(f"- Linked requirements: {', '.join(instruction['linkedRequirementIds'])}") if delta: lines.append("- Before:") lines.append("") lines.append(f"```text\n{delta['before']}\n```") lines.append("- After:") lines.append("") lines.append(f"```text\n{delta['after']}\n```") lines.append("") report_path.write_text("\n".join(lines), encoding="utf-8") def apply_docx(request: Dict[str, Any]) -> Tuple[str, str, List[str]]: source_path = Path(request["sourcePath"]) output_path = Path(request["outputPath"]) document = Document(str(source_path)) instructions = request["editPlan"]["instructions"] before_after: List[Dict[str, str]] = [] warnings: List[str] = [] for instruction in instructions: if not instruction.get("safeToApply", False): warnings.append(f"Skipped unsafe instruction in section {instruction['sectionLabel']}.") continue action = instruction["action"] target = instruction.get("targetText") or "" replacement = instruction.get("replacementText") or "" anchor = instruction.get("insertionAnchor") or target replacement, sanitize_warnings, replacement_rejected = sanitize_recruiter_text(replacement) warnings.extend(sanitize_warnings) if instruction.get("replacementText") and replacement_rejected: warnings.append(f"Skipped contaminated instruction in section {instruction['sectionLabel']}.") continue if action in {"replace_phrase", "rewrite_bullet"} and target: paragraph = find_paragraph_by_text(document, target) if paragraph is None: warnings.append(f"Could not locate target text for instruction in section {instruction['sectionLabel']}.") continue before_after.append({"id": instruction["id"], "before": paragraph.text, "after": replacement}) set_paragraph_text(paragraph, replacement) elif action == "insert_bullet": paragraphs = iter_document_paragraphs(document) anchor_paragraph = find_paragraph_by_text(document, anchor) or (paragraphs[-1] if paragraphs else None) if anchor_paragraph is None: warnings.append(f"Could not locate anchor for insertion in section {instruction['sectionLabel']}.") continue inserted = insert_paragraph_after(anchor_paragraph, replacement) before_after.append({"id": instruction["id"], "before": "", "after": inserted.text}) elif action == "tighten_heading": paragraph = find_paragraph_by_text(document, target) if paragraph is None: warnings.append(f"Could not locate heading to tighten in section {instruction['sectionLabel']}.") continue before_after.append({"id": instruction["id"], "before": paragraph.text, "after": replacement}) set_paragraph_text(paragraph, replacement) original_text = request["resumeText"] updated_text = all_document_text(document) validate_final_cv_output(updated_text) for contact in extract_contact_lines(original_text): if normalize_text(contact) not in normalize_text(updated_text): warnings.append(f"Contact line may require manual verification: {contact}") document.save(str(output_path)) build_redline(Path(request["redlinePath"]), instructions, before_after, warnings) return output_path.name, "application/vnd.openxmlformats-officedocument.wordprocessingml.document", warnings def build_pdf_story(text: str) -> List[Any]: styles = getSampleStyleSheet() body_font, heading_font = register_unicode_fonts() body = ParagraphStyle( "PulseBody", parent=styles["BodyText"], fontName=body_font, fontSize=10.5, leading=14, spaceAfter=8, ) heading = ParagraphStyle( "PulseHeading", parent=styles["Heading2"], fontName=heading_font, fontSize=12.5, leading=16, spaceBefore=10, spaceAfter=8, ) story: List[Any] = [] for raw_line in text.splitlines(): line = raw_line.strip() if not line: story.append(Spacer(1, 0.18 * cm)) continue if len(line) < 40 and line == line.upper(): story.append(PdfParagraph(line, heading)) continue if re.match(r"^(summary|profile|experience|skills|education|projects|certifications|languages|תקציר|ניסיון|מיומנויות|השכלה)$", line, re.I): story.append(PdfParagraph(line, heading)) continue escaped = ( line.replace("&", "&") .replace("<", "<") .replace(">", ">") ) if line.startswith(("-", "•", "*")): escaped = f"• {escaped.lstrip('-•* ').strip()}" story.append(PdfParagraph(escaped, body)) return story def normalize_pdf_resume_lines(text: str) -> List[str]: section_markers = [ "תקציר מקצועי", "מוקדי ניסיון", "ניסיון מקצועי", "השכלה", "שפות", "Professional Summary", "Experience", "Education", "Skills", "Languages", ] normalized = text.replace("\uf0b7", "\n• ") for marker in section_markers: normalized = re.sub(rf"\s+({re.escape(marker)})\s+", rf"\n\1\n", normalized) lines: List[str] = [] for raw_line in normalized.splitlines(): line = raw_line.strip() if not line: lines.append("") continue if len(line) <= 420: lines.append(line) continue chunks = re.split(r"(?<=[.;:])\s+|(?<=\.)\s+|(?<=\|)\s+", line) buffer = "" for chunk in chunks: candidate = f"{buffer} {chunk}".strip() if len(candidate) > 320 and buffer: lines.append(buffer) buffer = chunk.strip() else: buffer = candidate if buffer: lines.append(buffer) return [line for line in lines if line or lines.count("") < 12] def build_docx_from_text(output_path: Path, text: str) -> None: document = Document() for raw_line in normalize_pdf_resume_lines(text): line = raw_line.strip() if not line: document.add_paragraph("") continue paragraph = document.add_paragraph(line) if len(line) < 45: for run in paragraph.runs: run.bold = True document.save(str(output_path)) def apply_pdf(request: Dict[str, Any]) -> Tuple[str, str, List[str]]: output_path = Path(request["outputPath"]).with_suffix(".docx") text = request["resumeText"] instructions = request["editPlan"]["instructions"] warnings = [ "Source PDF was converted to an ATS-safe DOCX because Hebrew PDF regeneration can damage text extraction order." ] before_after: List[Dict[str, str]] = [] updated_text = text for instruction in instructions: if not instruction.get("safeToApply", False): warnings.append(f"Skipped unsafe instruction in section {instruction['sectionLabel']}.") continue action = instruction["action"] target = instruction.get("targetText") or "" replacement = instruction.get("replacementText") or "" replacement, sanitize_warnings, replacement_rejected = sanitize_recruiter_text(replacement) warnings.extend(sanitize_warnings) if instruction.get("replacementText") and replacement_rejected: warnings.append(f"Skipped contaminated instruction in section {instruction['sectionLabel']}.") continue if action in {"replace_phrase", "rewrite_bullet"} and target and normalize_text(target) in normalize_text(updated_text): before_after.append({"id": instruction["id"], "before": target, "after": replacement}) updated_text = re.sub(re.escape(target), replacement, updated_text, count=1) elif action == "insert_bullet": anchor = instruction.get("insertionAnchor") or "" if anchor and normalize_text(anchor) in normalize_text(updated_text): before_after.append({"id": instruction["id"], "before": "", "after": replacement}) updated_text = updated_text.replace(anchor, f"{anchor}\n• {replacement}", 1) else: before_after.append({"id": instruction["id"], "before": "", "after": replacement}) updated_text += f"\n• {replacement}" validate_final_cv_output(updated_text) build_docx_from_text(output_path, updated_text) build_redline(Path(request["redlinePath"]), instructions, before_after, warnings) return output_path.name, "application/vnd.openxmlformats-officedocument.wordprocessingml.document", warnings def main() -> None: request_path = Path(sys.argv[1]) request = json.loads(request_path.read_text(encoding="utf-8")) source_format = request["sourceFormat"] if source_format == "docx": output_file_name, mime_type, warnings = apply_docx(request) else: output_file_name, mime_type, warnings = apply_pdf(request) print(json.dumps({ "outputFileName": output_file_name, "outputMimeType": mime_type, "outputPath": str(Path(request["outputPath"]).with_suffix(".docx")) if source_format == "pdf" else str(Path(request["outputPath"])), "warnings": warnings, })) if __name__ == "__main__": main()