"""Run-local, offline experiment; writes only beside this file. prepare: construct controlled PDF variants and retain inspected source. recipes: record the fixed-column intervention and page-level probes. """ import argparse import hashlib import importlib.metadata as metadata import inspect import json import platform import shutil import time from datetime import datetime from pathlib import Path import pdfplumber import pymupdf from markitdown.converters import _pdf_converter LAB = Path(__file__).resolve().parent CORPUS = LAB / "corpus" BASE = ["prose-single", "table-simple", "table-merged-header", "two-column", "scan-image", "mixed", "equation", "rotated-header"] EXTRA = ["columns-with-table", "wide-columns", "wide-columns-with-table"] SAMPLES = BASE + EXTRA def save_json(path, value): path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def prepare(): truth = json.loads((CORPUS / "two-column.truth.json").read_text()) # Widen the gutter without changing paragraph text or its expected order. doc = pymupdf.open() page = doc.new_page(width=900, height=720) for rect, paragraphs in [ (pymupdf.Rect(45, 55, 360, 675), truth["paragraphs"][:3]), (pymupdf.Rect(540, 55, 855, 675), truth["paragraphs"][3:]), ]: remaining = page.insert_textbox(rect, "\n\n".join(paragraphs), fontsize=11, fontname="helv", lineheight=1.35) if remaining < 0: raise RuntimeError("Text overflow in controlled fixture") doc.save(CORPUS / "wide-columns.pdf") doc.close() save_json(CORPUS / "wide-columns.truth.json", truth) for first, output in [("two-column", "columns-with-table"), ("wide-columns", "wide-columns-with-table")]: combined = pymupdf.open() for stem in [first, "table-simple"]: with pymupdf.open(CORPUS / (stem + ".pdf")) as source: combined.insert_pdf(source) combined.save(CORPUS / (output + ".pdf")) combined.close() # Scorer deliberately evaluates the unchanged column content only. save_json(CORPUS / (output + ".truth.json"), truth) source_dir = LAB / "source" source_dir.mkdir(exist_ok=True) objects = { "pdfplumber-page-extract-text.py": pdfplumber.page.Page.extract_text, "pdfplumber-page-get-textmap.py": pdfplumber.page.Page._get_textmap, "pdfplumber-wordmap-to-textmap.py": pdfplumber.utils.text.WordMap.to_textmap, "pdfplumber-chars-to-textmap.py": pdfplumber.utils.text.chars_to_textmap, "markitdown-pdf-convert.py": _pdf_converter.PdfConverter.convert, "markitdown-form-detection.py": _pdf_converter._extract_form_content_from_words, } sources = [] for name, obj in objects.items(): path = source_dir / name path.write_text(inspect.getsource(obj), encoding="utf-8") sources.append({"saved": str(path.relative_to(LAB)), "installed_path": inspect.getfile(obj), "function": obj.__qualname__, "first_line": inspect.getsourcelines(obj)[1], "sha256": digest(path)}) versions = {pkg: metadata.version(pkg) for pkg in ["pdfplumber", "pdfminer.six", "markitdown", "pymupdf4llm", "PyMuPDF"]} scorer = LAB.parents[2] / "tools" / "lab_score.py" shutil.copyfile(scorer, LAB / "scorer.py") save_json(LAB / "source-provenance.json", { "recorded_at": datetime.now().astimezone().isoformat(), "python": platform.python_version(), "versions": versions, "sources": sources, "scorer_sha256": digest(LAB / "scorer.py"), "fixtures": [{"file": f"corpus/{s}.pdf", "sha256": digest(CORPUS / (s + ".pdf"))} for s in SAMPLES], "construction": "original column page unchanged via insert_pdf; wide fixture reflows same truth with a larger gutter; combined truths score column paragraphs only", }) (CORPUS / "controlled-fixtures.md").write_text( "# Controlled additions\n\n" "Generated by ../experiment.py, not public documents.\n\n" "- columns-with-table.pdf: unchanged two-column.pdf followed by table-simple.pdf\n" "- wide-columns.pdf: same paragraph truth reflowed into a wider page and gutter\n" "- wide-columns-with-table.pdf: unchanged wide-columns.pdf followed by table-simple.pdf\n\n" "The combined truths score only the column text, not the appended table. This isolates whether an unrelated table page changes that text. The source pages are inserted without rasterizing. No Markovo request was made for these additions.\n", encoding="utf-8") def recipes(): profiles, probes, results = {}, {}, [] out_dir = LAB / "pdfplumber-columns" out_dir.mkdir(exist_ok=True) for stem in SAMPLES: path = CORPUS / (stem + ".pdf") start = time.perf_counter() parts, page_profiles, page_probes = [], [], [] with pdfplumber.open(path) as pdf: for page in pdf.pages: mid = page.width / 2 parts.extend([page.crop((0, 0, mid, page.height)).extract_text() or "", page.crop((mid, 0, page.width, page.height)).extract_text() or ""]) form = _pdf_converter._extract_form_content_from_words(page) page_profiles.append({"page": page.page_number, "chars": len(page.chars), "images": len(page.images), "width": page.width, "height": page.height, "form_detected": form is not None}) page_probes.append({"default": page.extract_text() or "", "layout": page.extract_text(layout=True) or "", "text_flow": page.extract_text(use_text_flow=True) or ""}) output = out_dir / (stem + ".md") output.write_text("\n\n".join(parts), encoding="utf-8") results.append({"tool": "pdfplumber-columns", "sample": stem, "format": "pdf", "elapsed_s": round(time.perf_counter() - start, 4), "output": str(output.relative_to(LAB)), "error": None}) profiles[stem] = page_profiles probes[stem] = page_probes save_json(LAB / "profiles.json", profiles) save_json(LAB / "parameter-probes.json", probes) runs_path = LAB / "runs.json" runs = json.loads(runs_path.read_text()) runs["results"] = [r for r in runs["results"] if r["tool"] != "pdfplumber-columns"] + results runs["tools"]["pdfplumber-columns"] = { "version": metadata.version("pdfplumber"), "device": "cpu", "configuration": "fixed midpoint crop, left then right; deliberately applied to all selected PDFs to retain negative controls"} save_json(runs_path, runs) print(json.dumps({"samples": len(SAMPLES), "python": platform.python_version(), "form_pages": {s: [p["form_detected"] for p in profiles[s]] for s in SAMPLES}}, indent=2)) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("mode", choices=["prepare", "recipes"]) args = parser.parse_args() {"prepare": prepare, "recipes": recipes}[args.mode]()