pinion-pdf 1.0.0: rebuild a CSV from a spreadsheet printed to PDF
Derives page layout from the file itself (batch size, column edges, row grid, heading font), places text by start-x to survive cell overflow, detects blackout redactions including filled-path bars and single-character bars, and self-verifies via Excel's printed row-number chain. Output byte-identical to four independently written implementations across a 1,048,576-row reference file. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012bQ28wRBiQYcSR9FzV9U7c
This commit is contained in:
daveandClaude Fable 5 committed
commit
74b06be742
11 files changed
+666
No files matched your search
@@ -0,0 +1,14 @@
|
||||
"""pinion-pdf: turn a spreadsheet that was printed to PDF back into a CSV.
|
||||
|
||||
Someone took a spreadsheet, hit "Print", and handed out the PDF. pinion
|
||||
measures the PDF before it reads it -- page size, column widths, column and
|
||||
row counts are all derived from the file -- rebuilds every row, detects
|
||||
blacked-out cells, and then checks its own working.
|
||||
|
||||
MAP: page.py - reading what the PDF drew on a page
|
||||
measure.py - measuring the layout
|
||||
convert.py - rebuilding the rows
|
||||
cli.py - run it, then check our own working
|
||||
"""
|
||||
|
||||
__version__ = "1.0.0"
|
||||
@@ -0,0 +1,3 @@
|
||||
from .cli import main
|
||||
|
||||
main()
|
||||
@@ -0,0 +1,78 @@
|
||||
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 .measure import measure
|
||||
from .convert import convert
|
||||
|
||||
|
||||
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, for a trial run")
|
||||
args = ap.parse_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.")
|
||||
@@ -0,0 +1,119 @@
|
||||
from .page import all_runs, run_start, run_text, row_numbers, solid_regions
|
||||
|
||||
|
||||
def which_column(edges, x):
|
||||
"""Which column does text starting at x belong to?
|
||||
Checked against text start with a slight fudge factor."""
|
||||
for i in range(len(edges) - 1):
|
||||
if edges[i] - 0.6 <= x < edges[i + 1] - 0.6:
|
||||
return i
|
||||
return None
|
||||
|
||||
|
||||
def convert(doc, M, out, n_batches):
|
||||
"""Rebuild every row and write it to `out`."""
|
||||
rpp, pitch = M["rows_per_page"], M["pitch"]
|
||||
stats = {"text_no_column": 0, "text_no_row": 0, "bars_no_row": 0,
|
||||
"pieces": 0, "gutter_disagreed": 0, "gutter_unreadable": 0,
|
||||
"n_rows": 0, "chain_ok": True,
|
||||
"blackouts": [], "readable": []}
|
||||
next_expected = 1
|
||||
|
||||
for batch in range(n_batches):
|
||||
if batch and batch % max(1, n_batches // 10) == 0:
|
||||
print(f" {100 * batch // n_batches}% ...")
|
||||
cells, redacted, known_rows = {}, {}, set()
|
||||
|
||||
for slice_no in range(M["ppb"]):
|
||||
page = doc[batch * M["ppb"] + slice_no]
|
||||
edges, offset = M["bands"][slice_no], M["offsets"][slice_no]
|
||||
runs = all_runs(page)
|
||||
data = [r for r in runs if r["font"] != M["font"]]
|
||||
if not data:
|
||||
continue
|
||||
|
||||
# STEP 1 - Read this page's printed row numbers.
|
||||
printed = row_numbers(runs, M["font"], M["table_left"])
|
||||
stats["gutter_unreadable"] += rpp - len(printed)
|
||||
|
||||
# STEP 2 - Vertical position -> row number.
|
||||
# This makes sure redaction-covered row numbers don't mess us up.
|
||||
# (Yes, it happens)
|
||||
if printed:
|
||||
anchor_y = min(printed)
|
||||
anchor_n = printed[anchor_y]
|
||||
else:
|
||||
anchor_y = M["first_row_y"]
|
||||
anchor_n = batch * rpp + 1
|
||||
|
||||
def row_at(y):
|
||||
n = anchor_n + round((y - anchor_y) / pitch)
|
||||
lo = anchor_n - (anchor_n - 1) % rpp
|
||||
return n if lo <= n < lo + rpp else None
|
||||
|
||||
# If we can read the number, might as well validate it.
|
||||
for y, n in printed.items():
|
||||
if row_at(y) != n:
|
||||
stats["gutter_disagreed"] += 1
|
||||
known_rows.add(n)
|
||||
|
||||
# STEP 3 - Text -> cell.
|
||||
for run in data:
|
||||
x, y = run_start(run)
|
||||
row = row_at(y)
|
||||
if row is None:
|
||||
stats["text_no_row"] += 1
|
||||
continue
|
||||
col = which_column(edges, x)
|
||||
if col is None:
|
||||
stats["text_no_column"] += 1
|
||||
continue
|
||||
slot = cells.setdefault(row, [[] for _ in range(M["n_cols"])])
|
||||
slot[offset + col].append((x, run_text(run)))
|
||||
|
||||
# STEP 4 - Blackout bars (Trap 2).
|
||||
for box in solid_regions(page):
|
||||
if box.width <= M["bar_min_w"] or box.height <= M["bar_min_h"]:
|
||||
continue
|
||||
row = row_at((box.y0 + box.y1) / 2)
|
||||
if row is None:
|
||||
stats["bars_no_row"] += 1
|
||||
continue
|
||||
col = which_column(edges, box.x0)
|
||||
letters = M["strips"][slice_no]
|
||||
label = letters[col] if col is not None else "?"
|
||||
stats["blackouts"].append((row, label, round(box.width, 1)))
|
||||
redacted[row] = max(redacted.get(row, 0), round(box.width, 1))
|
||||
|
||||
# Did any writing survive underneath the bar?
|
||||
# If so, point and laugh. And also report it.
|
||||
for run in data:
|
||||
if row_at(run_start(run)[1]) != row:
|
||||
continue
|
||||
under = [chr(c[0]) for c in run["chars"]
|
||||
if box.x0 - 0.5 <= c[2][0] <= box.x1 - 1.5]
|
||||
if under:
|
||||
stats["readable"].append((row, label, "".join(under)))
|
||||
|
||||
# STEP 5 - Write it all out.
|
||||
for row in sorted(set(cells) | known_rows):
|
||||
slot = cells.get(row) or [[] for _ in range(M["n_cols"])]
|
||||
line = []
|
||||
for cell in slot:
|
||||
if len(cell) > 1:
|
||||
stats["pieces"] += 1
|
||||
# sorting by x rejoins a cell drawn in several pieces
|
||||
line.append("".join(t for _, t in sorted(cell)).strip())
|
||||
if row == 1:
|
||||
# header row names our two extra redaction columns
|
||||
out.writerow(line + ["reason_redacted", "redaction_bar_pt"])
|
||||
else:
|
||||
bar = redacted.get(row)
|
||||
out.writerow(line + ["TRUE" if bar else "FALSE", bar or ""])
|
||||
stats["n_rows"] += 1
|
||||
# Row numbers must count 1,2,3... with no gaps or repeats. Duh.
|
||||
if row != next_expected:
|
||||
stats["chain_ok"] = False
|
||||
next_expected = row + 1
|
||||
|
||||
return stats
|
||||
@@ -0,0 +1,155 @@
|
||||
import sys
|
||||
import statistics
|
||||
|
||||
import pymupdf
|
||||
|
||||
from .page import all_runs, run_start, run_text, row_numbers, vertical_rules
|
||||
|
||||
|
||||
def find_heading_font(page0):
|
||||
"""Excel draws its labelling (row numbers, letter strip) in a different
|
||||
font from the data, and the row numbers sit leftmost on the page. So:
|
||||
whatever font the leftmost text uses is the labelling font."""
|
||||
leftmost = min(all_runs(page0), key=lambda r: run_start(r)[0])
|
||||
if not run_text(leftmost).strip().isdigit():
|
||||
sys.exit("Expected a row number at the far left of page 1, found "
|
||||
f"{run_text(leftmost)!r}. Was this printed from a spreadsheet "
|
||||
"with row and column headings switched on?")
|
||||
return leftmost["font"]
|
||||
|
||||
|
||||
def find_table_left(page0, font):
|
||||
"""Where cell data begins. Left of this: margin. Right of this: table."""
|
||||
return min(run_start(r)[0] for r in all_runs(page0) if r["font"] != font)
|
||||
|
||||
|
||||
def column_letters(page, font, table_left):
|
||||
"""The 'A B C D' strip above the table, read left to right."""
|
||||
got = []
|
||||
for run in all_runs(page):
|
||||
if run["font"] != font:
|
||||
continue
|
||||
for ch in run["chars"]:
|
||||
if ch[2][0] >= table_left and chr(ch[0]).isalpha():
|
||||
got.append((ch[2][0], chr(ch[0])))
|
||||
return "".join(c for _, c in sorted(got))
|
||||
|
||||
|
||||
def find_batch_shape(doc, font, table_left):
|
||||
"""How many pages hold one batch of rows, and which columns each shows.
|
||||
We follow the letter strip page by page until it starts over at 'A'."""
|
||||
strips = [column_letters(doc[p], font, table_left)
|
||||
for p in range(min(40, doc.page_count))]
|
||||
if not strips[0].startswith("A"):
|
||||
sys.exit(f"Page 1 starts at column {strips[0][:1]!r}, not 'A'. This "
|
||||
"does not look like the first page of a printed spreadsheet.")
|
||||
ppb = next((i for i in range(1, len(strips))
|
||||
if strips[i].startswith("A")), None)
|
||||
if ppb is None:
|
||||
sys.exit("Could not find where the column letters start over at 'A'.")
|
||||
if doc.page_count % ppb:
|
||||
sys.exit(f"The file has {doc.page_count} pages, which does not divide "
|
||||
f"evenly into batches of {ppb}. The layout is not uniform "
|
||||
"and this tool cannot safely rebuild it.")
|
||||
return strips[:ppb], ppb, doc.page_count // ppb
|
||||
|
||||
|
||||
def find_row_grid(page0_runs, font, table_left):
|
||||
"""Rows per page, row height, and where the first row sits -- read off
|
||||
the printed row numbers, which is exact. Assumes rows are evenly spaced
|
||||
and the same height throughout the file."""
|
||||
gutter = row_numbers(page0_runs, font, table_left)
|
||||
if len(gutter) < 2:
|
||||
sys.exit("Could not read Excel's printed row numbers from the left "
|
||||
"margin of page 1. They are required to place rows exactly.")
|
||||
ys = sorted(gutter)
|
||||
if gutter[ys[0]] != 1:
|
||||
sys.exit(f"Page 1 begins at spreadsheet row {gutter[ys[0]]}, not "
|
||||
"row 1. This may not be the start of the document.")
|
||||
pitch = statistics.median(b - a for a, b in zip(ys, ys[1:]))
|
||||
return len(gutter), pitch, ys[0]
|
||||
|
||||
|
||||
def find_column_edges(doc, font, strips, ppb, n_batches):
|
||||
"""Column boundaries for each page type. Excel only draws a boundary
|
||||
where a cell has a border, so no single page shows all of them -- we
|
||||
sample many pages of each type and combine. Edges closer than 2pt are
|
||||
the same line drawn twice; anything left of the first cell is margin,
|
||||
except the last such line, which is the table's own left edge."""
|
||||
step = max(1, n_batches // 80)
|
||||
bands = []
|
||||
for slice_no in range(ppb):
|
||||
found, starts = set(), []
|
||||
for b in range(0, n_batches, step):
|
||||
page = doc[b * ppb + slice_no]
|
||||
found |= vertical_rules(page)
|
||||
starts += [run_start(r)[0] for r in all_runs(page)
|
||||
if r["font"] != font]
|
||||
if not starts:
|
||||
sys.exit(f"No cell text found anywhere on page type {slice_no + 1}.")
|
||||
leftmost_cell = min(starts)
|
||||
edges = []
|
||||
for x in sorted(found):
|
||||
if x > leftmost_cell:
|
||||
if edges and x - edges[-1] < 2.0:
|
||||
continue
|
||||
edges.append(x)
|
||||
else:
|
||||
edges = [x]
|
||||
wanted = len(strips[slice_no]) + 1
|
||||
if len(edges) != wanted:
|
||||
sys.exit(f"On page type {slice_no + 1} the column letters say "
|
||||
f"there are {len(strips[slice_no])} columns "
|
||||
f"({strips[slice_no]}), which needs {wanted} edges, but "
|
||||
f"{len(edges)} were found.")
|
||||
bands.append(edges)
|
||||
return bands
|
||||
|
||||
|
||||
def measure(path):
|
||||
"""Establish every fact the converter needs, narrating as we go."""
|
||||
doc = pymupdf.open(path)
|
||||
page0_runs = all_runs(doc[0])
|
||||
|
||||
print("MEASURING THE PAGE")
|
||||
print("-" * 62)
|
||||
font = find_heading_font(doc[0])
|
||||
print(f" heading font ......... {font}")
|
||||
|
||||
table_left = find_table_left(doc[0], font)
|
||||
print(f" table starts at x .... {table_left:.1f}pt "
|
||||
f"(left of this is the row-number margin)")
|
||||
|
||||
strips, ppb, n_batches = find_batch_shape(doc, font, table_left)
|
||||
print(f" pages per batch ...... {ppb} column strips {strips}")
|
||||
print(f" batches in the file .. {n_batches:,}")
|
||||
|
||||
rows_per_page, pitch, first_row_y = find_row_grid(page0_runs, font, table_left)
|
||||
print(f" rows per page ........ {rows_per_page}")
|
||||
print(f" row height ........... {pitch:.2f}pt "
|
||||
f"(first row sits at y={first_row_y:.1f})")
|
||||
|
||||
bands = find_column_edges(doc, font, strips, ppb, n_batches)
|
||||
for i, edges in enumerate(bands):
|
||||
print(f" page type {i + 1} ({strips[i]:<8}) {len(edges) - 1} columns, "
|
||||
f"edges {' '.join(f'{e:.0f}' for e in edges)}")
|
||||
|
||||
# Telling a blackout bar from a grid line: separate by SHAPE, not size.
|
||||
# A single-character blackout is only ~3pt wide, but it is always as
|
||||
# tall as the writing it hides; a rule is long but paper-thin.
|
||||
bar_min_w, bar_min_h = 1.2, pitch * 0.5
|
||||
print(f" blackout bars ........ solid shapes over {bar_min_w}pt wide "
|
||||
f"and {bar_min_h:.1f}pt tall")
|
||||
print(f" spreadsheet shape .... {n_batches * rows_per_page:,} rows "
|
||||
f"x {sum(len(s) for s in strips)} columns\n")
|
||||
|
||||
doc.close()
|
||||
return {
|
||||
"font": font, "table_left": table_left, "strips": strips,
|
||||
"ppb": ppb, "n_batches": n_batches, "bands": bands,
|
||||
"rows_per_page": rows_per_page, "pitch": pitch,
|
||||
"first_row_y": first_row_y, "bar_min_w": bar_min_w,
|
||||
"bar_min_h": bar_min_h,
|
||||
"n_cols": sum(len(s) for s in strips),
|
||||
"offsets": [sum(len(s) for s in strips[:i]) for i in range(ppb)],
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Reading what the PDF actually drew on a page.
|
||||
|
||||
Small views onto the raw page. Everything downstream is built from these.
|
||||
The key one is all_runs(): it returns each piece of text AS THE PDF DREW IT,
|
||||
one run per spreadsheet cell, which is what dodges Trap 1 -- overlapping
|
||||
cells stay separate because the PDF drew them separately.
|
||||
"""
|
||||
|
||||
|
||||
def all_runs(page):
|
||||
"""Every piece of text on the page, one run per spreadsheet cell."""
|
||||
return [r for r in page.get_texttrace() if r["chars"]]
|
||||
|
||||
|
||||
def run_start(run):
|
||||
"""Where a piece of text begins, as (across, down)."""
|
||||
return run["chars"][0][2]
|
||||
|
||||
|
||||
def run_text(run):
|
||||
"""The text of a run. Word Wrap would really screw this up; don't let
|
||||
the PDF boomers know."""
|
||||
return "".join(chr(c[0]) for c in run["chars"])
|
||||
|
||||
|
||||
def solid_regions(page):
|
||||
"""Every solid filled region, taken as a whole. A blackout bar can be a
|
||||
plain rectangle OR a filled four-sided path -- visually identical,
|
||||
different in the PDF -- so we measure whole regions, never segments."""
|
||||
return [d["rect"] for d in page.get_drawings() if d["type"] == "f"]
|
||||
|
||||
|
||||
def vertical_rules(page):
|
||||
"""x-positions of thin vertical lines: <1.2pt wide, >8pt tall, drawn
|
||||
either as skinny rectangles or as literal line segments."""
|
||||
xs = set()
|
||||
for d in page.get_drawings():
|
||||
for item in d["items"]:
|
||||
if item[0] == "re" and item[1].width < 1.2 and item[1].height > 8:
|
||||
xs.add(round(item[1].x0 + item[1].width / 2, 1))
|
||||
elif item[0] == "l":
|
||||
a, b = item[1], item[2]
|
||||
if abs(a.x - b.x) < 1.2 and abs(a.y - b.y) > 8:
|
||||
xs.add(round((a.x + b.x) / 2, 1))
|
||||
return xs
|
||||
|
||||
|
||||
def row_numbers(runs, heading_font, table_left):
|
||||
"""Excel's printed row numbers -> {vertical position: row number}.
|
||||
Digits of one number are drawn separately, so group them by line and
|
||||
read left to right."""
|
||||
lines = {}
|
||||
for run in runs:
|
||||
if run["font"] != heading_font:
|
||||
continue
|
||||
for ch in run["chars"]:
|
||||
x, y = ch[2]
|
||||
if x < table_left:
|
||||
lines.setdefault(round(y, 1), []).append((x, chr(ch[0])))
|
||||
out = {}
|
||||
for y, chars in lines.items():
|
||||
text = "".join(c for _, c in sorted(chars))
|
||||
if text.isdigit():
|
||||
out[y] = int(text)
|
||||
return out
|
||||
Reference in new issue
Block a user