Sniffs the format on page 1: printed row numbers -> gridded mode (unchanged); UUID-keyed rows -> new headerless mode, which anchors rows on per-record UUIDs and takes the column map from the page-1 header labels. Handles fused cells (ID+Name+Org+networks), mid-value line wraps, two-timestamp Time Frame ranges, wrapped header labels, and per-export schema drift. Validated against three monthly JC PD audits: 120/120 cross-engine spot checks, 0 anomalies, field-identical to the standalone converter it replaces. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012bQ28wRBiQYcSR9FzV9U7c
138 lines
5.7 KiB
Python
138 lines
5.7 KiB
Python
import sys
|
|
import csv
|
|
import argparse
|
|
|
|
try:
|
|
import pymupdf # noqa: F401 (probed here for a friendly error)
|
|
except ImportError:
|
|
sys.exit("Missing dependency. Run: pip install pymupdf")
|
|
|
|
from .page import all_runs, run_start, run_text
|
|
from .measure import measure
|
|
from .convert import convert
|
|
from .headerless import looks_headerless, convert_headerless
|
|
|
|
|
|
def report(label, value, ok=None, bad=""):
|
|
"""One line of the results table. ok=None means purely informational."""
|
|
tail = "" if ok is None else (" ok" if ok else f" <-- {bad}")
|
|
print(f" {label + ' ':.<28} {value}{tail}")
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(
|
|
prog="pinion",
|
|
description="Rebuild a CSV from a spreadsheet that was printed to PDF.")
|
|
ap.add_argument("pdf")
|
|
ap.add_argument("out_csv")
|
|
ap.add_argument("--limit", type=int, default=0,
|
|
help="only convert the first N batches (gridded) or "
|
|
"pages (headerless), for a trial run")
|
|
args = ap.parse_args()
|
|
|
|
# Which of the two known print formats is this? A gridded Excel print
|
|
# puts its row numbers leftmost on the page; the headerless audit
|
|
# variant has no headings at all but keys every row with a UUID.
|
|
doc = pymupdf.open(args.pdf)
|
|
runs0 = all_runs(doc[0])
|
|
if not runs0:
|
|
sys.exit("Page 1 has no text at all. Is this a scan?")
|
|
leftmost = min(runs0, key=lambda r: run_start(r)[0])
|
|
if run_text(leftmost).strip().isdigit():
|
|
doc.close()
|
|
run_gridded(args)
|
|
elif looks_headerless(doc[0]):
|
|
run_headerless(doc, args)
|
|
else:
|
|
sys.exit("Unrecognised layout: no printed row numbers (gridded "
|
|
"Excel print) and no UUID-keyed rows (headerless audit "
|
|
"print). pinion doesn't know this format yet.")
|
|
|
|
|
|
def run_gridded(args):
|
|
M = measure(args.pdf)
|
|
n_batches = min(args.limit, M["n_batches"]) if args.limit else M["n_batches"]
|
|
|
|
print("CONVERTING (takes a few minutes)")
|
|
print("-" * 62)
|
|
doc = pymupdf.open(args.pdf)
|
|
with open(args.out_csv, "w", newline="", encoding="utf-8") as fh:
|
|
stats = convert(doc, M, csv.writer(fh), n_batches)
|
|
doc.close()
|
|
|
|
expected = n_batches * M["rows_per_page"] - 1
|
|
misplaced = stats["text_no_column"] + stats["text_no_row"]
|
|
tally = {}
|
|
for _row, letter, _width in stats["blackouts"]:
|
|
tally[letter] = tally.get(letter, 0) + 1
|
|
|
|
print(f"\nWritten to {args.out_csv}\n")
|
|
print("-" * 62)
|
|
report("rows written", f"{stats['n_rows']:,}")
|
|
report("rows expected", f"{expected:,}", stats["n_rows"] == expected, "MISMATCH")
|
|
report("row numbers 1..N unbroken", "yes" if stats["chain_ok"] else "NO",
|
|
stats["chain_ok"], "GAPS OR DUPLICATES")
|
|
report("unplaced text",
|
|
f"{stats['text_no_column']:,} by column, {stats['text_no_row']:,} by row",
|
|
misplaced == 0, "SOME TEXT LOST")
|
|
report("unplaced bars", f"{stats['bars_no_row']:,}",
|
|
stats["bars_no_row"] == 0, "REDACTIONS LOST")
|
|
report("disagreed numbers", f"{stats['gutter_disagreed']:,}",
|
|
stats["gutter_disagreed"] == 0, "LAYOUT WRONG")
|
|
report("row numbers hidden by bars",
|
|
f"{stats['gutter_unreadable']:,} (placed by spacing instead)")
|
|
report("cells drawn in pieces",
|
|
f"{stats['pieces']:,} (rejoined; nothing lost)")
|
|
print()
|
|
report("blackout bars found", f"{len(stats['blackouts']):,}")
|
|
report("rows affected", f"{len(set(b[0] for b in stats['blackouts'])):,}")
|
|
report("which columns", tally or "none")
|
|
if stats["readable"]:
|
|
report("TEXT READABLE UNDER BARS", f"{len(stats['readable']):,}",
|
|
False, "REDACTION IS COSMETIC, HAHA! TEXT RECOVERABLE")
|
|
for row, letter, text in stats["readable"][:10]:
|
|
print(f" row {row:,} column {letter}: {text!r}")
|
|
else:
|
|
report("text readable under bars", "none (deleted)")
|
|
|
|
if n_batches * M["rows_per_page"] == 1048576:
|
|
print("\n WARNING: exactly 1,048,576 rows. That's an Excel magic number")
|
|
print(" The source was almost certainly truncated.")
|
|
print(" Go tell the source they owe you more data.")
|
|
|
|
|
|
def run_headerless(doc, args):
|
|
n_pages = min(args.limit, doc.page_count) if args.limit else doc.page_count
|
|
print("HEADERLESS AUDIT PRINT (no row/column headings; anchoring on")
|
|
print("the page-1 header labels and per-row UUIDs)")
|
|
print("-" * 62)
|
|
with open(args.out_csv, "w", newline="", encoding="utf-8") as fh:
|
|
r = convert_headerless(doc, csv.writer(fh), n_pages)
|
|
doc.close()
|
|
s = r["stats"]
|
|
|
|
print(f"\nWritten to {args.out_csv}\n")
|
|
print("-" * 62)
|
|
report("rows written", f"{r['n_rows']:,}")
|
|
report("columns in this export", f"{len(r['names'])}/13",
|
|
not r["missing"],
|
|
f"absent from the PDF itself: {r['missing']}")
|
|
report("org name (derived)", repr(r["org"]))
|
|
report("time-frame ranges",
|
|
f"{r['tf_ok']:,} valid pairs, {r['tf_single']} single",
|
|
r["tf_bad"] == 0, f"{r['tf_bad']} MALFORMED")
|
|
report("rows with a bad uuid", s["rows_bad_uuid"],
|
|
s["rows_bad_uuid"] == 0, "ROWS DROPPED")
|
|
report("unplaced runs", s["runs_no_column"] + s["runs_below_last_anchor"],
|
|
s["runs_no_column"] + s["runs_below_last_anchor"] == 0,
|
|
"SOME TEXT LOST")
|
|
report("strays in name/org windows", s["strays_in_name_org_windows"],
|
|
s["strays_in_name_org_windows"] == 0, "COLUMN MAPPED WRONG")
|
|
report("fused SearchTime split", s["searchtime_splits"])
|
|
report("blackout bars found", len(r["bars"]))
|
|
if r["missing"]:
|
|
print(f"\n NOTE: this export has no {', '.join(r['missing'])} "
|
|
"column(s).")
|
|
print(" That data was never in the PDF; the CSV column is left "
|
|
"empty.")
|