Bank Statement → Excel
You are extracting the transaction detail from a statement PDF into a workbook that a
preparer, reviewer, or auditor can rely on without re-reading the PDF.
The standard is not "did you get most of the rows." The standard is: the extraction
mathematically ties to the statement's own opening and closing balance. If it doesn't
tie, the file is not delivered — it's flagged. A silently incomplete statement extract is
worse than no extract at all, because it will be relied on.
Non-negotiables
- Never invent, estimate, or interpolate a number. If a figure is illegible, missing,
or ambiguous, it goes in the Exceptions tab with the page number and the reason. You do
not fill the gap by back-solving from the balance column.
- Every output row carries its source page. A reviewer must be able to jump from any
row in Excel to the exact page of the PDF.
- Debits and credits get separate columns. Never a single signed amount column — sign
conventions get misread downstream and reverse the reconciliation.
- Money is decimal, never float. Use Python's
Decimal. Do not round anything; carry
two decimal places as presented.
- The proof runs before delivery, not after. See Step 4.
Workflow
Step 1 — Inventory the source
List every PDF in scope. For each, capture: file name, page count, account number (last 4
only in filenames and tab names — never write a full account number into an output
filename), statement period, and whether the page has a real text layer or is a scan.
python3 scripts/extract_text.py --input <folder-or-file> --out work/
This writes, per PDF:
<name>.text.txt — page-delimited text layer<name>.words.jsonl — every word with x0/x1/top/bottom coordinates (needed when columns
collide or descriptions wrap)<name>.meta.json — page count, per-page character count, scanned: true|false
A page with near-zero extractable characters is a scan. Route those through Step 2. Do not
attempt to read numbers off a scan from the text layer — you will get silent blanks.
Step 2 — OCR only if needed (optional, local)
If scanned: true, and ocrmypdf/tesseract are available locally:
python3 scripts/extract_text.py --input <file.pdf> --out work/ --ocr
If OCR tooling is not installed, say so plainly and stop — offer the install command
(brew install ocrmypdf / apt install ocrmypdf tesseract-ocr). Do not send the document
to a cloud OCR service. These are client bank records.
OCR-specific vigilance. On OCR'd output, treat these as suspect until confirmed against
the balance column: 0/O, 1/l/I, 5/S, 8/B, 6/G, comma-vs-period decimal separators,
and dropped leading digits (1,240.00 read as 240.00). The balance proof in Step 4 is
what catches these — which is exactly why it is mandatory on OCR'd statements.
Step 3 — Normalize the transactions
Read the text and word-coordinate output and build a normalized table. You are doing the
layout reasoning here; the scripts do the arithmetic and the file writing.
Columns:
| Field | Rule |
|---|
row_id | Sequential from 1, in statement order. Never re-sorted. |
source_file | PDF filename |
source_page | 1-based page number |
txn_date | ISO YYYY-MM-DD. See year-assignment rule below. |
post_date | ISO, if the statement shows a separate posting date; else blank |
description | Full description, wrapped lines joined with a single space |
check_no | If present |
reference | Bank reference / trace / confirmation number if present |
|
Year assignment. Most statements print MM/DD. Derive the year from the statement
period, and handle the December/January rollover explicitly: on a statement period of
12/16/2025–01/15/2026, a 12/28 row is 2025 and a 01/03 row is 2026. Getting this wrong
silently misstates a fiscal year cutoff. If a date cannot be resolved to a year with
certainty, it is an exception — not a guess.
Lines that are not transactions. Exclude, and do not count toward totals: Balance Forward, Beginning Balance, Ending Balance, Total Deposits, Total Withdrawals,
Continued on next page, subtotals, column headers repeated on continuation pages, daily
balance summary tables, interest-rate disclosures, and the check-image grid. These are the
single most common source of double-counting. If you include a subtotal as a transaction,
your proof will fail by exactly that subtotal — which is a useful tell.
Wrapped descriptions. A description continuing on the next physical line has no date and
no amount. Join it to the transaction above. Use the word coordinates to confirm it starts
in the description column, not the date column.
Multi-account statements. Some PDFs contain several accounts. Split by account into
separate normalized tables and separate worksheet tabs. Never commingle — each account gets
its own independent balance proof.
Write the normalized table to work/<name>.transactions.csv.
Step 4 — Prove it (mandatory gate)
python3 scripts/build_workbook.py \
--transactions work/<name>.transactions.csv \
--opening <opening balance from statement> \
--closing <closing balance from statement> \
--out "<Client> - <Bank> x<last4> - <Period> - Transactions.xlsx"
Three independent tests run, all of which must pass:
- Test A — Balance roll-forward.
opening + Σcredits − Σdebits = closing, using the
opening and closing balances printed on the statement. Any difference is reported to the
cent.
- Test B — Statement control totals. Where the statement prints "Total Deposits" and
"Total Withdrawals," extracted sums must agree to those figures.
- Test C — Line-balance continuity. Where the statement prints a per-line running
balance: for every row,
prior balance ± amount = printed balance. This isolates which
row is wrong rather than just telling you the total is off — it is the fastest debugging
tool you have and it catches OCR digit errors precisely.
If any test fails, the script exits non-zero and writes a break report. Do not hand the
workbook over. Go find the row. In practice the cause is one of five things, in
descending order of frequency: a subtotal captured as a transaction, a wrapped description
line captured as a zero-amount transaction, a transaction on a continuation page missed
entirely, a debit classified as a credit, or an OCR digit misread. Test C tells you which.
Step 5 — Deliver
The workbook has four tabs, in this order:
- Transactions — the normalized rows. Frozen header, autofilter, dates as real dates,
amounts as accounting-format numbers (not text),
source_page visible. A Recomputed Balance column sits beside the printed balance so a reviewer can see they agree.
- Proof — the three tests, showing opening balance, total credits, total debits,
computed closing, statement closing, and the difference (which reads
0.00). This is
the tab a reviewer looks at first. It is what makes the file signable.
- Exceptions — every illegible figure, unresolved date, ambiguous classification, and
judgment call, with page reference. An empty Exceptions tab is a valid and good outcome;
a missing Exceptions tab is not.
- Source Map — one row per PDF page: file, page, transactions extracted from that page,
page subtotal. Lets a reviewer sample-check any page in seconds.
Then report in chat, briefly: statements processed, period covered, transaction count,
total debits and credits, proof result, and the exception count. If anything failed, lead
with that.
Multi-statement engagements
For a 12-month or multi-account set, run each statement independently — each gets its own
proof — then add a Continuity tab: each statement's closing balance must equal the next
statement's opening balance, in date order. A gap here means a missing statement month, and
that is worth flagging loudly. It is the failure that most often makes it all the way to a
lender or an auditor unnoticed.
Security posture
Everything runs locally: pdfplumber for extraction, optional local ocrmypdf/tesseract,
openpyxl for output. No network calls, no cloud OCR, no uploads, no telemetry. Original
PDFs are opened read-only and never modified. Intermediate files land in work/ and can be
deleted after delivery — tell the user they exist so they can shred them under their
document-retention policy.
Dependencies
pip install pdfplumber openpyxl
# optional, for scanned statements:
# macOS: brew install ocrmypdf
# Debian: sudo apt install ocrmypdf tesseract-ocr
Built by Adopt AI — free to use and modify.
Appendix — bundled files
If you installed the .skill package these files are already in place and you can ignore this appendix. If you copied the skill as text, create the files below at the paths shown, alongside your SKILL.md. The skill will not run without them.
scripts/build_workbook.py
#!/usr/bin/env python3
"""
Prove a normalized statement extract and write the deliverable workbook.
Runs three independent completeness tests and REFUSES to produce a clean
workbook if any of them fail:
Test A Balance roll-forward: opening + credits - debits == closing
Test B Statement control totals (if supplied)
Test C Line-balance continuity, where the statement prints a running balance
Usage:
python3 build_workbook.py \
--transactions work/jan.transactions.csv \
--opening 42150.22 --closing 38904.71 \
--out "Acme - First National x4471 - 2026-01 - Transactions.xlsx"
Optional:
--stmt-deposits 18200.00 --stmt-withdrawals 21445.51 (enables Test B)
--account-label "First National x4471"
--period "2026-01-01 to 2026-01-31"
--force write the workbook even if a test fails (still marked FAILED)
Input CSV columns (blank where not applicable):
row_id, source_file, source_page, txn_date, post_date, description,
check_no, reference, debit, credit, balance, section
Money is handled as Decimal throughout. Nothing is rounded.
"""
from __future__ import annotations
import argparse
import csv
import sys
from datetime import date, datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
try:
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
except ImportError:
sys.exit("openpyxl is required. pip install openpyxl")
ZERO = Decimal("0.00")
MONEY_FMT = '#,##0.00;[Red](#,##0.00)'
DATE_FMT = "yyyy-mm-dd"
HDR_FILL = PatternFill("solid", fgColor="1F3864")
HDR_FONT = Font(bold=True, color="FFFFFF")
PASS_FONT = Font(bold=True, color="006100")
FAIL_FONT = Font(bold=True, color="9C0006")
FAIL_FILL = PatternFill("solid", fgColor="FFC7CE")
COLUMNS = [
"row_id", "source_file", "source_page", "txn_date", "post_date",
"description", "check_no", "reference", "section",
"debit", "credit", "balance",
]
def dec(raw) -> Decimal | None:
"""Parse a money string to Decimal. Blank -> None. Never guesses."""
if raw is None:
return None
s = str(raw).strip()
if s in ("", "-", "--", "n/a", "N/A", "None"):
return None
neg = s.startswith("(") and s.endswith(")")
s = s.strip("()").replace(",", "").replace("$", "").strip()
if s.endswith("-"): # trailing-minus convention
neg, s = True, s[:-1]
try:
v = Decimal(s)
except InvalidOperation:
raise ValueError(f"Unparseable amount: {raw!r}")
return -v if neg else v
def parse_date(raw) -> date | None:
s = str(raw or "").strip()
if not s:
return None
try:
return datetime.strptime(s, "%Y-%m-%d").date()
except ValueError:
raise ValueError(f"Dates must be ISO YYYY-MM-DD, got {raw!r}")
def load_rows(path: Path) -> list[dict]:
with path.open(newline="", encoding="utf-8-sig") as fh:
rows = list(csv.DictReader(fh))
if not rows:
sys.exit(f"No rows in {path}")
out = []
for i, r in enumerate(rows, start=1):
d, c = dec(r.get("debit")), dec(r.get("credit"))
if d is not None and c is not None:
raise ValueError(
f"Row {i}: both debit and credit populated. A transaction is one "
f"or the other. ({r.get('description','')!r})"
)
if d is None and c is None:
raise ValueError(
f"Row {i}: neither debit nor credit populated. This is usually a "
f"wrapped description line or a subtotal that should not be a "
f"transaction row. ({r.get('description','')!r})"
)
out.append({
"row_id": int(r.get("row_id") or i),
"source_file": (r.get("source_file") or "").strip(),
"source_page": int(r["source_page"]) if str(r.get("source_page") or "").strip() else None,
"txn_date": parse_date(r.get("txn_date")),
"post_date": parse_date(r.get("post_date")),
"description": " ".join((r.get("description") or "").split()),
"check_no": (r.get("check_no") or "").strip(),
"reference": (r.get("reference") or "").strip(),
"section": (r.get("section") or "").strip(),
"debit": d, "credit": c,
"balance": dec(r.get("balance")),
})
return out
# --------------------------------------------------------------------------- tests
def test_a(rows, opening: Decimal, closing: Decimal) -> dict:
deb = sum((r["debit"] for r in rows if r["debit"] is not None), ZERO)
cre = sum((r["credit"] for r in rows if r["credit"] is not None), ZERO)
computed = opening + cre - deb
diff = computed - closing
return {
"name": "Test A - Balance roll-forward",
"detail": "opening + total credits - total debits = statement closing balance",
"debits": deb, "credits": cre,
"computed_closing": computed, "stmt_closing": closing,
"difference": diff, "passed": diff == ZERO,
}
def test_b(rows, stmt_dep, stmt_wd) -> dict | None:
if stmt_dep is None and stmt_wd is None:
return None
deb = sum((r["debit"] for r in rows if r["debit"] is not None), ZERO)
cre = sum((r["credit"] for r in rows if r["credit"] is not None), ZERO)
parts, ok = [], True
if stmt_dep is not None:
d = cre - stmt_dep
ok &= d == ZERO
parts.append(("Total deposits / credits", cre, stmt_dep, d))
if stmt_wd is not None:
d = deb - stmt_wd
ok &= d == ZERO
parts.append(("Total withdrawals / debits", deb, stmt_wd, d))
return {
"name": "Test B - Statement control totals",
"detail": "extracted sums agree to the totals printed on the statement",
"parts": parts, "passed": ok,
}
def test_c(rows, opening: Decimal) -> dict:
"""Walk the printed running balance. Isolates the exact offending row."""
with_bal = [r for r in rows if r["balance"] is not None]
if not with_bal:
return {
"name": "Test C - Line-balance continuity",
"detail": "statement does not print a per-line running balance - not applicable",
"breaks": [], "passed": True, "applicable": False,
}
breaks, prior = [], opening
for r in rows:
amt = (r["credit"] or ZERO) - (r["debit"] or ZERO)
expected = prior + amt
if r["balance"] is not None:
if r["balance"] != expected:
breaks.append({
"row_id": r["row_id"], "page": r["source_page"],
"date": r["txn_date"], "description": r["description"][:70],
"prior": prior, "amount": amt,
"expected": expected, "printed": r["balance"],
"difference": r["balance"] - expected,
})
prior = r["balance"] # resync so one bad row doesn't cascade
else:
prior = expected
return {
"name": "Test C - Line-balance continuity",
"detail": "prior balance +/- amount = printed balance, for every row",
"breaks": breaks, "passed": not breaks, "applicable": True,
}
def recomputed_balances(rows, opening: Decimal) -> list[Decimal]:
out, run = [], opening
for r in rows:
run += (r["credit"] or ZERO) - (r["debit"] or ZERO)
out.append(run)
return out
# --------------------------------------------------------------------------- excel
def style_header(ws, ncols: int, row: int = 1) -> None:
for c in range(1, ncols + 1):
cell = ws.cell(row=row, column=c)
cell.fill, cell.font = HDR_FILL, HDR_FONT
cell.alignment = Alignment(vertical="center", wrap_text=True)
ws.row_dimensions[row].height = 28
def autosize(ws, widths: dict[int, int]) -> None:
for col, w in widths.items():
ws.column_dimensions[get_column_letter(col)].width = w
def write_transactions(wb, rows, opening) -> None:
ws = wb.active
ws.title = "Transactions"
headers = [
"Row", "Source File", "Pg", "Txn Date", "Post Date", "Description",
"Check No", "Reference", "Section", "Debit", "Credit",
"Balance (printed)", "Balance (recomputed)", "Tie",
]
ws.append(headers)
style_header(ws, len(headers))
recomp = recomputed_balances(rows, opening)
for r, rb in zip(rows, recomp):
tie = "" if r["balance"] is None else ("OK" if r["balance"] == rb else "BREAK")
ws.append([
r["row_id"], r["source_file"], r["source_page"], r["txn_date"],
r["post_date"], r["description"], r["check_no"], r["reference"],
r["section"],
float(r["debit"]) if r["debit"] is not None else None,
float(r["credit"]) if r["credit"] is not None else None,
float(r["balance"]) if r["balance"] is not None else None,
float(rb), tie,
])
last = ws.max_row
for row in ws.iter_rows(min_row=2, max_row=last):
for idx in (4, 5):
row[idx - 1].number_format = DATE_FMT
for idx in (10, 11, 12, 13):
row[idx - 1].number_format = MONEY_FMT
if row[13].value == "BREAK":
row[13].font, row[13].fill = FAIL_FONT, FAIL_FILL
# totals strip
ws.append([])
t = ws.max_row + 1
ws.cell(row=t, column=9, value="TOTALS").font = Font(bold=True)
ws.cell(row=t, column=10, value=f"=SUBTOTAL(109,J2:J{last})").number_format = MONEY_FMT
ws.cell(row=t, column=11, value=f"=SUBTOTAL(109,K2:K{last})").number_format = MONEY_FMT
for c in (10, 11):
ws.cell(row=t, column=c).font = Font(bold=True)
ws.freeze_panes = "D2"
ws.auto_filter.ref = f"A1:N{last}"
autosize(ws, {1: 6, 2: 26, 3: 5, 4: 12, 5: 12, 6: 54, 7: 10, 8: 20,
9: 20, 10: 14, 11: 14, 12: 17, 13: 19, 14: 8})
def write_proof(wb, results, meta, overall_pass: bool) -> None:
ws = wb.create_sheet("Proof")
ws.column_dimensions["A"].width = 44
ws.column_dimensions["B"].width = 18
ws.column_dimensions["C"].width = 18
ws.column_dimensions["D"].width = 16
ws.column_dimensions["E"].width = 46
r = 1
ws.cell(row=r, column=1, value="COMPLETENESS PROOF").font = Font(bold=True, size=14)
r += 1
verdict = ws.cell(
row=r, column=1,
value="RESULT: PASS - extraction ties to the statement"
if overall_pass else
"RESULT: FAILED - DO NOT RELY ON THIS FILE",
)
verdict.font = PASS_FONT if overall_pass else FAIL_FONT
if not overall_pass:
verdict.fill = FAIL_FILL
r += 2
for k, v in meta.items():
ws.cell(row=r, column=1, value=k).font = Font(bold=True)
ws.cell(row=r, column=2, value=v)
r += 1
r += 1
a = results["a"]
ws.cell(row=r, column=1, value=a["name"]).font = Font(bold=True, size=12)
ws.cell(row=r, column=5, value=a["detail"]).font = Font(italic=True, size=9)
r += 1
for label, val in [
("Opening balance per statement", meta["Opening balance"]),
("Add: total credits / deposits", a["credits"]),
("Less: total debits / withdrawals", -a["debits"]),
("= Computed closing balance", a["computed_closing"]),
("Closing balance per statement", a["stmt_closing"]),
("Difference", a["difference"]),
]:
ws.cell(row=r, column=1, value=label)
cell = ws.cell(row=r, column=2, value=float(val))
cell.number_format = MONEY_FMT
if label.startswith(("=", "Difference")):
cell.font = Font(bold=True)
if label == "Difference":
cell.font = PASS_FONT if a["passed"] else FAIL_FONT
if not a["passed"]:
cell.fill = FAIL_FILL
r += 1
ws.cell(row=r, column=1, value="Test A verdict").font = Font(bold=True)
vc = ws.cell(row=r, column=2, value="PASS" if a["passed"] else "FAIL")
vc.font = PASS_FONT if a["passed"] else FAIL_FONT
r += 2
b = results.get("b")
if b:
ws.cell(row=r, column=1, value=b["name"]).font = Font(bold=True, size=12)
ws.cell(row=r, column=5, value=b["detail"]).font = Font(italic=True, size=9)
r += 1
for h, col in zip(["Item", "Extracted", "Per statement", "Difference"], range(1, 5)):
hc = ws.cell(row=r, column=col, value=h)
hc.fill, hc.font = HDR_FILL, HDR_FONT
r += 1
for label, extracted, stmt, diff in b["parts"]:
ws.cell(row=r, column=1, value=label)
for col, val in ((2, extracted), (3, stmt), (4, diff)):
c = ws.cell(row=r, column=col, value=float(val))
c.number_format = MONEY_FMT
dc = ws.cell(row=r, column=4)
dc.font = PASS_FONT if diff == ZERO else FAIL_FONT
if diff != ZERO:
dc.fill = FAIL_FILL
r += 1
ws.cell(row=r, column=1, value="Test B verdict").font = Font(bold=True)
vc = ws.cell(row=r, column=2, value="PASS" if b["passed"] else "FAIL")
vc.font = PASS_FONT if b["passed"] else FAIL_FONT
r += 2
else:
ws.cell(row=r, column=1, value="Test B - Statement control totals").font = Font(bold=True, size=12)
ws.cell(row=r, column=2, value="NOT PERFORMED - statement totals not supplied")
r += 2
c = results["c"]
ws.cell(row=r, column=1, value=c["name"]).font = Font(bold=True, size=12)
ws.cell(row=r, column=5, value=c["detail"]).font = Font(italic=True, size=9)
r += 1
if not c["applicable"]:
ws.cell(row=r, column=2, value="NOT APPLICABLE - no per-line balance printed")
r += 1
elif c["passed"]:
vc = ws.cell(row=r, column=2, value="PASS - every printed balance agrees")
vc.font = PASS_FONT
r += 1
else:
ws.cell(row=r, column=1, value=f"{len(c['breaks'])} break(s) - each row below is where "
f"the extract diverges from the statement").font = FAIL_FONT
r += 1
heads = ["Row", "Pg", "Date", "Description", "Prior bal", "Amount",
"Expected", "Printed", "Difference"]
for col, h in enumerate(heads, start=1):
hc = ws.cell(row=r, column=col, value=h)
hc.fill, hc.font = HDR_FILL, HDR_FONT
r += 1
for brk in c["breaks"]:
ws.cell(row=r, column=1, value=brk["row_id"])
ws.cell(row=r, column=2, value=brk["page"])
dc = ws.cell(row=r, column=3, value=brk["date"])
dc.number_format = DATE_FMT
ws.cell(row=r, column=4, value=brk["description"])
for col, key in ((5, "prior"), (6, "amount"), (7, "expected"),
(8, "printed"), (9, "difference")):
mc = ws.cell(row=r, column=col, value=float(brk[key]))
mc.number_format = MONEY_FMT
ws.cell(row=r, column=9).font = FAIL_FONT
r += 1
r += 1
ws.cell(row=r, column=1, value="Most likely causes when a test fails, in order:").font = Font(bold=True)
for line in [
"1. A subtotal or 'Total Deposits' line captured as a transaction",
"2. A wrapped description line captured as its own row",
"3. A transaction on a continuation page missed entirely",
"4. A debit classified as a credit (or vice versa)",
"5. An OCR digit misread (0/O, 1/l, 5/S, 8/B) or a dropped leading digit",
"Test C above identifies the specific row - start there, not with the totals.",
]:
r += 1
ws.cell(row=r, column=1, value=line)
def write_exceptions(wb, exceptions_csv: Path | None) -> None:
ws = wb.create_sheet("Exceptions")
headers = ["#", "Source File", "Pg", "Type", "Description of issue",
"As presented", "Treatment applied", "Needs preparer decision"]
ws.append(headers)
style_header(ws, len(headers))
n = 0
if exceptions_csv and exceptions_csv.exists():
with exceptions_csv.open(newline="", encoding="utf-8-sig") as fh:
for i, row in enumerate(csv.DictReader(fh), start=1):
ws.append([i] + [row.get(k, "") for k in
["source_file", "source_page", "type", "issue",
"as_presented", "treatment", "needs_decision"]])
n = i
if n == 0:
ws.cell(row=2, column=1, value="None")
ws.cell(row=2, column=5,
value="No illegible figures, unresolved dates, or ambiguous "
"classifications were encountered.")
autosize(ws, {1: 5, 2: 26, 3: 5, 4: 20, 5: 56, 6: 18, 7: 34, 8: 22})
def write_source_map(wb, rows) -> None:
ws = wb.create_sheet("Source Map")
headers = ["Source File", "Pg", "Transactions", "Debits on page",
"Credits on page", "Net"]
ws.append(headers)
style_header(ws, len(headers))
agg: dict[tuple, dict] = {}
for r in rows:
key = (r["source_file"], r["source_page"])
a = agg.setdefault(key, {"n": 0, "d": ZERO, "c": ZERO})
a["n"] += 1
a["d"] += r["debit"] or ZERO
a["c"] += r["credit"] or ZERO
for (f, p), a in sorted(agg.items(), key=lambda kv: (kv[0][0], kv[0][1] or 0)):
ws.append([f, p, a["n"], float(a["d"]), float(a["c"]), float(a["c"] - a["d"])])
for row in ws.iter_rows(min_row=2, max_row=ws.max_row):
for idx in (4, 5, 6):
row[idx - 1].number_format = MONEY_FMT
ws.freeze_panes = "A2"
autosize(ws, {1: 30, 2: 6, 3: 14, 4: 16, 5: 16, 6: 16})
# --------------------------------------------------------------------------- main
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--transactions", required=True)
ap.add_argument("--opening", required=True)
ap.add_argument("--closing", required=True)
ap.add_argument("--out", required=True)
ap.add_argument("--stmt-deposits")
ap.add_argument("--stmt-withdrawals")
ap.add_argument("--exceptions", help="optional exceptions CSV")
ap.add_argument("--account-label", default="")
ap.add_argument("--period", default="")
ap.add_argument("--force", action="store_true",
help="write the workbook even if a test fails")
args = ap.parse_args()
rows = load_rows(Path(args.transactions))
opening, closing = dec(args.opening), dec(args.closing)
if opening is None or closing is None:
sys.exit("--opening and --closing are required and must be numeric.")
results = {
"a": test_a(rows, opening, closing),
"b": test_b(rows, dec(args.stmt_deposits), dec(args.stmt_withdrawals)),
"c": test_c(rows, opening),
}
overall = results["a"]["passed"] and results["c"]["passed"] and (
results["b"]["passed"] if results["b"] else True
)
meta = {
"Account": args.account_label or "(not supplied)",
"Statement period": args.period or "(not supplied)",
"Source file(s)": ", ".join(sorted({r["source_file"] for r in rows if r["source_file"]})) or "(not supplied)",
"Transactions extracted": len(rows),
"Opening balance": opening,
"Prepared": datetime.now().strftime("%Y-%m-%d %H:%M"),
}
# ---- console report
print("=" * 68)
print("COMPLETENESS PROOF")
print("=" * 68)
a = results["a"]
print(f"Transactions extracted : {len(rows)}")
print(f"Opening balance : {opening:>14,.2f}")
print(f"Total credits : {a['credits']:>14,.2f}")
print(f"Total debits : {a['debits']:>14,.2f}")
print(f"Computed closing : {a['computed_closing']:>14,.2f}")
print(f"Statement closing : {a['stmt_closing']:>14,.2f}")
print(f"Difference : {a['difference']:>14,.2f} "
f"{'PASS' if a['passed'] else '*** FAIL ***'}")
if results["b"]:
print(f"\nTest B (control totals): "
f"{'PASS' if results['b']['passed'] else '*** FAIL ***'}")
for label, ex, st, df in results["b"]["parts"]:
print(f" {label:<28} extracted {ex:>12,.2f} stmt {st:>12,.2f} diff {df:>10,.2f}")
c = results["c"]
if not c["applicable"]:
print("\nTest C (line continuity): N/A - no per-line balance printed")
elif c["passed"]:
print("\nTest C (line continuity): PASS")
else:
print(f"\nTest C (line continuity): *** FAIL - {len(c['breaks'])} break(s) ***")
for brk in c["breaks"][:15]:
print(f" row {brk['row_id']:>4} pg {brk['page']} {brk['date']} "
f"{brk['description'][:40]:<40} expected {brk['expected']:>12,.2f} "
f"printed {brk['printed']:>12,.2f} diff {brk['difference']:>10,.2f}")
if len(c["breaks"]) > 15:
print(f" ... and {len(c['breaks']) - 15} more")
if not overall and not args.force:
print("\n" + "=" * 68)
print("WORKBOOK NOT WRITTEN. The extract does not tie to the statement.")
print("Fix the extract and re-run. Do not deliver an unproven file.")
print("Check, in order: subtotals captured as transactions; wrapped")
print("description lines; missed continuation pages; debit/credit flipped;")
print("OCR digit misreads. Test C above names the offending row.")
print("Use --force only to produce a marked-FAILED file for debugging.")
print("=" * 68)
return 1
wb = Workbook()
write_transactions(wb, rows, opening)
write_proof(wb, results, meta, overall)
write_exceptions(wb, Path(args.exceptions) if args.exceptions else None)
write_source_map(wb, rows)
wb.active = 0
out = Path(args.out)
if not overall:
out = out.with_name(out.stem + " [FAILED PROOF]" + out.suffix)
out.parent.mkdir(parents=True, exist_ok=True)
wb.save(str(out))
print("\n" + ("PROOF PASSED. " if overall else "PROOF FAILED (forced). "))
print(f"Workbook: {out}")
return 0 if overall else 1
if __name__ == "__main__":
sys.exit(main())
scripts/extract_text.py
#!/usr/bin/env python3
"""
Extract the text layer and word geometry from statement PDFs, fully locally.
Writes, per input PDF:
<stem>.text.txt page-delimited text
<stem>.words.jsonl one JSON object per word with coordinates
<stem>.meta.json page count, per-page char counts, scanned flag
Usage:
python3 extract_text.py --input statements/ --out work/
python3 extract_text.py --input jan.pdf --out work/ --ocr
No network access. Originals are opened read-only and never modified.
"""
from __future__ import annotations
import argparse
import json
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
try:
import pdfplumber
except ImportError:
sys.exit("pdfplumber is required. pip install pdfplumber")
# A page with fewer than this many extractable characters is treated as a scan.
SCAN_CHAR_THRESHOLD = 60
def ocr_pdf(src: Path) -> Path:
"""Run local OCR, returning a path to a new searchable PDF. Never touches src."""
ocrmypdf = shutil.which("ocrmypdf")
if not ocrmypdf:
sys.exit(
"This PDF has no text layer and 'ocrmypdf' is not installed.\n"
"Install it locally, then re-run with --ocr:\n"
" macOS: brew install ocrmypdf\n"
" Debian: sudo apt install ocrmypdf tesseract-ocr\n"
"Do not upload client bank records to a cloud OCR service."
)
# Resolve to an absolute input path and confirm it is a real PDF before handing
# it to another process. No shell is involved (list argv, shell=False), and the
# binary is the absolute path resolved above rather than a PATH lookup at spawn.
src = src.resolve(strict=True)
if not src.is_file() or src.suffix.lower() != ".pdf":
sys.exit(f"Not a PDF file: {src}")
out = Path(tempfile.mkdtemp(prefix="ocr_")) / f"{src.stem}.ocr.pdf"
print(f" OCR (local): {src.name} -> {out.name}", flush=True)
subprocess.run(
[
ocrmypdf,
"--force-ocr",
"--deskew",
"--optimize", "0",
"--quiet",
str(src),
str(out),
],
check=True,
)
return out
def extract(pdf_path: Path, out_dir: Path, use_ocr: bool) -> dict:
stem = pdf_path.stem
source = pdf_path
if use_ocr:
source = ocr_pdf(pdf_path)
text_lines: list[str] = []
words_out: list[str] = []
page_meta: list[dict] = []
with pdfplumber.open(str(source)) as pdf:
for pageno, page in enumerate(pdf.pages, start=1):
txt = page.extract_text() or ""
text_lines.append(f"\n===== PAGE {pageno} =====\n{txt}")
for w in page.extract_words(
use_text_flow=False,
keep_blank_chars=False,
extra_attrs=["size"],
):
words_out.append(
json.dumps(
{
"page": pageno,
"text": w["text"],
"x0": round(float(w["x0"]), 2),
"x1": round(float(w["x1"]), 2),
"top": round(float(w["top"]), 2),
"bottom": round(float(w["bottom"]), 2),
"size": round(float(w.get("size", 0)), 1),
},
ensure_ascii=False,
)
)
page_meta.append(
{
"page": pageno,
"chars": len(txt),
"scanned": len(txt.strip()) < SCAN_CHAR_THRESHOLD,
"width": round(float(page.width), 1),
"height": round(float(page.height), 1),
}
)
out_dir.mkdir(parents=True, exist_ok=True)
(out_dir / f"{stem}.text.txt").write_text("\n".join(text_lines), encoding="utf-8")
(out_dir / f"{stem}.words.jsonl").write_text(
"\n".join(words_out) + "\n", encoding="utf-8"
)
scanned_pages = [p["page"] for p in page_meta if p["scanned"]]
meta = {
"source_file": pdf_path.name,
"ocr_applied": use_ocr,
"page_count": len(page_meta),
"scanned_pages": scanned_pages,
"scanned": bool(scanned_pages),
"pages": page_meta,
}
(out_dir / f"{stem}.meta.json").write_text(
json.dumps(meta, indent=2), encoding="utf-8"
)
return meta
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--input", required=True, help="PDF file or folder of PDFs")
ap.add_argument("--out", default="work", help="output directory (default: work)")
ap.add_argument(
"--ocr",
action="store_true",
help="run local OCR first (required for scanned statements)",
)
args = ap.parse_args()
src = Path(args.input).expanduser()
out_dir = Path(args.out).expanduser()
if src.is_dir():
pdfs = sorted(src.glob("*.pdf")) + sorted(src.glob("*.PDF"))
elif src.is_file():
pdfs = [src]
else:
sys.exit(f"Not found: {src}")
if not pdfs:
sys.exit(f"No PDFs found in {src}")
print(f"Extracting {len(pdfs)} PDF(s) -> {out_dir}/")
needs_ocr: list[str] = []
for p in pdfs:
print(f"- {p.name}")
meta = extract(p, out_dir, args.ocr)
flag = ""
if meta["scanned"] and not args.ocr:
needs_ocr.append(p.name)
flag = f" [SCANNED pages {meta['scanned_pages']} - re-run with --ocr]"
print(f" {meta['page_count']} page(s){flag}")
if needs_ocr:
print(
"\nThese files have pages with no usable text layer and MUST be re-run "
"with --ocr before you read any figures from them:"
)
for n in needs_ocr:
print(f" - {n}")
print(
"Reading amounts off a scanned page without OCR produces silent blanks, "
"not errors."
)
return 2
print("\nDone. Text layer looks usable on every page.")
return 0
if __name__ == "__main__":
sys.exit(main())
