jobtrackingapp/docs/cv-integration-qa/benchmarks/generate.py

"""Generate a separate synthetic QA corpus; never replace the original import benchmark."""
import argparse
from hashlib import sha256
import json
from pathlib import Path
import re

from docx import Document
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from reportlab import __file__ as reportlab_file
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.pdfgen import canvas


FONT = "CvQaSynthetic"
FONT_PATH = Path(reportlab_file).parent / "fonts/Vera.ttf"


def load_corpus(path):
    cases = json.loads(path.read_text(encoding="utf-8"))
    ids = [case["id"] for case in cases]
    if len(ids) != len(set(ids)):
        raise ValueError("Duplicate logical fixture ID.")
    for case in cases:
        if not re.fullmatch(r"[a-z0-9-]+", case["id"]) or case["split"] not in {"development", "frozen"}:
            raise ValueError("Invalid fixture ID or split.")
        if not isinstance(case["expected"], dict) or not case["blocks"]:
            raise ValueError("Every case requires independent expectations and source blocks.")
    return cases


def fragments(case):
    """Source units for literal text recovery, before interpretation or field assignment."""
    result = []
    for block in case["blocks"]:
        if block.get("type") == "table":
            result.extend(block["headers"])
            result.extend(cell for row in block["rows"] for cell in row if cell)
        elif block.get("type") != "page_break":
            result.append(block["text"])
    return result


def generate_case(case, output):
    identifier = case["id"]
    if not re.fullmatch(r"[a-z0-9-]+", identifier):
        raise ValueError("Unsafe fixture ID.")
    paths = {kind: output / f"{identifier}.{kind}" for kind in ("pdf", "docx")}
    if any(path.exists() for path in paths.values()):
        raise FileExistsError("Select fresh fixture output; historical bytes are immutable.")
    output.mkdir(parents=True, exist_ok=True)
    pdfmetrics.registerFont(TTFont(FONT, str(FONT_PATH)))
    pdf = canvas.Canvas(str(paths["pdf"]), pagesize=(595, 842), invariant=1)
    pdf.setTitle("Synthetic CV integration QA: " + identifier)
    pdf.setAuthor("Jobjakt synthetic fixture")
    word = Document()
    y = 800

    def page():
        nonlocal y
        pdf.showPage()
        y = 800

    def wrap(text, width, size=10):
        lines, line = [], ""
        for token in text.split():
            proposed = (line + " " + token).strip()
            if line and pdfmetrics.stringWidth(proposed, FONT, size) > width:
                lines.append(line)
                line = token
            else:
                line = proposed
        return lines + [line]

    for block in case["blocks"]:
        kind = block.get("type", "paragraph")
        if kind == "page_break":
            word.add_page_break()
            page()
            continue
        if kind == "table":
            headers, rows = block["headers"], block["rows"]
            if not headers or any(len(row) != len(headers) for row in rows):
                raise ValueError("Table rows must match actual header columns.")
            table = word.add_table(rows=1, cols=len(headers))
            table.style = "Table Grid"
            for cell, text in zip(table.rows[0].cells, headers):
                cell.text = text
            repeat_header = OxmlElement("w:tblHeader")
            table.rows[0]._tr.get_or_add_trPr().append(repeat_header)
            for row in rows:
                for cell, text in zip(table.add_row().cells, row):
                    cell.text = text
            width = 515 / len(headers)
            for row in [headers] + rows:
                wrapped = [wrap(cell, width - 10) for cell in row]
                height = max(len(lines) for lines in wrapped) * 14 + 10
                if y - height < 40:
                    page()
                pdf.setFont(FONT, 10)
                for column, lines in enumerate(wrapped):
                    x = 40 + column * width
                    pdf.rect(x, y - height, width, height)
                    for i, line in enumerate(lines):
                        pdf.drawString(x + 5, y - 15 - 14 * i, line)
                y -= height
            y -= 18
            continue
        if kind not in {"paragraph", "heading", "list"}:
            raise ValueError("Unsupported fixture block kind.")
        text = block["text"]
        paragraph = word.add_paragraph(text)
        if kind == "heading":
            paragraph.style = word.styles["Heading 2"]
        elif kind == "list":
            paragraph.style = word.styles["List Bullet"]
        # PDF lists use a visible explicit marker; DOCX lists use actual paragraph list styling.
        size = 12 if kind == "heading" else 10
        lines = wrap(("- " if kind == "list" else "") + text, 515, size)
        height = len(lines) * 15 + (14 if kind == "heading" else 9)
        if y - height < 40:
            page()
        pdf.setFont(FONT, size)
        for line in lines:
            pdf.drawString(40, y, line)
            y -= 15
        y -= 14 if kind == "heading" else 9
    pdf.save()
    word.save(paths["docx"])
    return paths


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--corpus", type=Path, default=Path(__file__).with_name("corpus.json"))
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    if args.output.exists():
        raise FileExistsError("Select a fresh directory; do not overwrite a frozen manifest.")
    cases = load_corpus(args.corpus)
    manifest = {"corpusSha256": sha256(args.corpus.read_bytes()).hexdigest(), "logicalCases": len(cases),
                "generatorSha256": sha256(Path(__file__).read_bytes()).hexdigest(), "documents": []}
    for case in cases:
        for kind, path in generate_case(case, args.output).items():
            manifest["documents"].append({"caseId": case["id"], "split": case["split"], "format": kind,
                                          "file": path.name, "sha256": sha256(path.read_bytes()).hexdigest(),
                                          "bytes": path.stat().st_size})
    (args.output / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({"logicalCases": len(cases), "documents": len(manifest["documents"]), "output": str(args.output)}))


if __name__ == "__main__":
    main()