""" converter.py — hardened "Python-first" bank-statement PDF -> Excel/CSV converter. Free path, no LLM. Handles: - gridlined tables (pdfplumber lines) - borderless tables (word-position reconstruction) - scanned/image PDFs (OCR: PyMuPDF rasterize + Tesseract) - renamed headers (synonym map) - single signed Amount column - US (1,234.56) and EU (1.234,56) numbers - relabeled opening/closing balances - multiple accounts in one PDF (sectioning) Reconciliation (opening + Σcredits - Σdebits == closing) gates every result. Usage: python converter.py [statement.pdf] (default ./sample_statement.pdf) """ from __future__ import annotations import csv import math import re import statistics import sys from dataclasses import dataclass, field from pathlib import Path import pdfplumber from openpyxl import Workbook # ---------------- number / date parsing ---------------- def parse_money(value) -> float | None: if value is None: return None s = str(value).strip() if not s or s in {"-", "—"}: return None up = s.upper() neg = ("(" in s and ")" in s) or "DR" in up or "-" in s # '-' anywhere, e.g. "$-2,942.71" t = re.sub(r"[^0-9.,]", "", s) if not t: return None if "." in t and "," in t: # both separators present t = t.replace(",", "") if t.rfind(".") > t.rfind(",") else t.replace(".", "").replace(",", ".") elif "," in t: # comma only t = t.replace(",", ".") if (t.count(",") == 1 and re.search(r",\d{2}$", t)) else t.replace(",", "") try: v = float(t) except ValueError: return None return -abs(v) if neg else v _MONTHS = {m: i for i, m in enumerate( ["jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec"], start=1)} def norm_date(s: str) -> str: s = (s or "").strip() m = re.match(r"(\d{1,2})[/\-.](\d{1,2})[/\-.](\d{2,4})", s) if m: d, mo, y = m.groups() y = "20" + y if len(y) == 2 else y return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}" m = re.match(r"(\d{4})-(\d{2})-(\d{2})", s) if m: return s[:10] m = re.match(r"(\d{1,2})\s+([A-Za-z]{3,})\s+(\d{4})", s) if m: d, mon, y = m.groups() mo = _MONTHS.get(mon[:3].lower()) if mo: return f"{int(y):04d}-{mo:02d}-{int(d):02d}" return s # ---------------- header synonyms ---------------- _CANON = { "date": ["transaction date", "txn date", "value date", "posting date", "trans date", "date"], "balance": ["running balance", "closing balance", "balance amount", "balance"], "debit": ["debit amount", "withdrawals", "withdrawal", "money out", "paid out", "debit", "dr"], "credit": ["credit amount", "deposits", "deposit", "money in", "paid in", "credit", "cr"], "amount": ["amount", "value"], "description": ["description", "details", "narrative", "particulars", "memo", "reference", "remarks", "transaction"], } def map_header(cell: str) -> str | None: c = (cell or "").strip().lower() if not c: return None # whole-word match so short tokens like "cr"/"dr" don't hit "des(cr)iption"/"a(dr)ess" for key in ("date", "balance", "debit", "credit", "amount", "description"): for syn in _CANON[key]: if re.search(r"\b" + re.escape(syn) + r"\b", c): return key return None # ---------------- model ---------------- @dataclass class Transaction: date: str description: str debit: float | None credit: float | None balance: float | None @dataclass class ExtractedStatement: account: str | None = None account_holder: str | None = None period: str | None = None opening_balance: float | None = None closing_balance: float | None = None transactions: list[Transaction] = field(default_factory=list) reconciles: bool | None = None reconciliation_detail: str = "" # structured reconciliation result (each check is True/False/None=absent) running_balance_ok: bool | None = None stated_summary_ok: bool | None = None # "verified" (all present checks pass) | "review" (checks disagree) | # "failed" (all present checks fail) | "unverified" (nothing to check) reconcile_status: str = "unverified" @property def empty(self) -> bool: return not (self.transactions or self.opening_balance is not None or self.closing_balance is not None or self.account) # ---------------- the converter ---------------- class BankStatementConverter: def __init__(self, pdf_path): self.pdf_path = Path(pdf_path) self.statements: list[ExtractedStatement] = [] def extract(self) -> list[ExtractedStatement]: events = [] with pdfplumber.open(self.pdf_path) as pdf: for i, page in enumerate(pdf.pages): base = i * 100000.0 # offset so global order == reading order text = page.extract_text() or "" page_ev = self._events_from_text_page(page) if text.strip() else self._events_from_ocr(i) events += [(top + base, pri, kind, val) for (top, pri, kind, val) in page_ev] events.sort(key=lambda e: (e[0], e[1])) # by (top, priority) self._assemble(events) for st in self.statements: self._reconcile(st) return self.statements # ----- text page (gridlined or borderless) ----- def _events_from_text_page(self, page): ev = [] for ln in page.extract_text_lines(): kind, val = self._classify(ln["text"]) if kind: ev.append((ln["top"], 0, kind, val)) tables = page.find_tables() if tables: for tb in tables: ev.append((tb.bbox[1], 1, "table", tb.extract())) else: # borderless -> reconstruct from word positions rows = self._rows_from_words(page.extract_words(), float(page.width)) if rows: ev.append((rows[0][1], 1, "table", rows[0][0])) return ev # ----- scanned page (OCR) ----- def _events_from_ocr(self, page_index): # pytesseract has a Py3.14 stderr-decode bug -> call tesseract CLI (TSV) directly import csv as _csv import io import subprocess import fitz doc = fitz.open(self.pdf_path) try: # 200 DPI is enough for OCR; 300 rendered ~30MB/page and spiked RAM -> OOM on the host. pix = doc[page_index].get_pixmap(dpi=200) # write the temp image next to the PDF (tesseract/leptonica can't read /tmp here) png = self.pdf_path.parent / f".ocr_p{page_index}.png" png.write_bytes(pix.tobytes("png")) finally: doc.close() # release the page/pixmap immediately so RAM doesn't grow page-by-page try: proc = subprocess.run(["tesseract", str(png), "stdout", "tsv"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) tsv = proc.stdout.decode("utf-8", "ignore") # don't decode tesseract's binary stderr except FileNotFoundError: return [] finally: png.unlink(missing_ok=True) words = [] for row in _csv.DictReader(io.StringIO(tsv), delimiter="\t", quoting=_csv.QUOTE_NONE): txt = (row.get("text") or "").strip() try: conf = float(row.get("conf", "-1")) except (TypeError, ValueError): conf = -1 if txt and conf > 30: x, y = int(row["left"]), int(row["top"]) w, h = int(row["width"]), int(row["height"]) words.append({"text": txt, "x0": x, "x1": x + w, "top": y, "bottom": y + h}) ev = [] # markers from OCR text lines for line_text, top in self._lines_from_words(words): kind, val = self._classify(line_text) if kind: ev.append((top, 0, kind, val)) rows = self._rows_from_words(words, float(pix.width)) if rows: ev.append((rows[0][1], 1, "table", rows[0][0])) return ev # ----- marker classification ----- @staticmethod def _classify(text): if ":" not in text: return None, None label, _, val = text.partition(":") L = label.strip().lower() val = val.strip() if "holder" in L: return "holder", val if "period" in L: return "period", val if any(k in L for k in ("opening balance", "balance brought forward", "brought forward", "opening", "beginning balance", "b/f")): return "opening", val if any(k in L for k in ("closing balance", "balance carried forward", "carried forward", "closing", "ending balance", "c/f")): return "closing", val if "account" in L: return "account", val return None, None # ----- word-position table reconstruction (borderless + OCR) ----- def _lines_from_words(self, words): if not words: return [] ws = sorted(words, key=lambda w: (w["top"], w["x0"])) heights = [w["bottom"] - w["top"] for w in ws if w["bottom"] > w["top"]] ytol = (statistics.median(heights) if heights else 8) * 0.7 lines, cur, top = [], [ws[0]], ws[0]["top"] for w in ws[1:]: if abs(w["top"] - top) <= ytol: cur.append(w) else: lines.append(cur) cur, top = [w], w["top"] lines.append(cur) out = [] for ln in lines: ln.sort(key=lambda w: w["x0"]) out.append((" ".join(w["text"] for w in ln), ln[0]["top"])) return out def _rows_from_words(self, words, page_width): """Reconstruct a table from word positions. Columns are found from the vertical whitespace gaps in the table region (robust for right-aligned numbers). Returns [(rows, table_top)] or [].""" if not words: return [] ws = sorted(words, key=lambda w: (w["top"], w["x0"])) heights = [w["bottom"] - w["top"] for w in ws if w["bottom"] > w["top"]] ytol = (statistics.median(heights) if heights else 8) * 0.7 lines, cur, top = [], [ws[0]], ws[0]["top"] for w in ws[1:]: if abs(w["top"] - top) <= ytol: cur.append(w) else: lines.append(sorted(cur, key=lambda w: w["x0"])) cur, top = [w], w["top"] lines.append(sorted(cur, key=lambda w: w["x0"])) hdr_idx = None for i, ln in enumerate(lines): t = " ".join(w["text"] for w in ln).lower() if "date" in t and any(k in t for k in ("balance", "amount", "credit", "debit", "deposit", "withdraw", "description", "details")): hdr_idx = i break if hdr_idx is None: return [] region = lines[hdr_idx:] # header + data rows only # merge word x-intervals into column "bands" separated by whitespace gaps intervals = sorted((w["x0"], w["x1"]) for ln in region for w in ln) merge_tol = max(4.0, 0.006 * page_width) # < inter-column gap, > intra-word spaces bands = [] bx0, bx1 = intervals[0] for x0, x1 in intervals[1:]: if x0 - bx1 <= merge_tol: bx1 = max(bx1, x1) else: bands.append((bx0, bx1)); bx0, bx1 = x0, x1 bands.append((bx0, bx1)) ncol = len(bands) bounds = [-math.inf] + [(bands[j][1] + bands[j + 1][0]) / 2 for j in range(ncol - 1)] + [math.inf] def assign(line): row = [""] * ncol for w in line: cx = (w["x0"] + w["x1"]) / 2 for j in range(ncol): if bounds[j] <= cx < bounds[j + 1]: row[j] = (row[j] + " " + w["text"]).strip() break return row rows = [assign(region[0])] for ln in region[1:]: r = assign(ln) if any(r): rows.append(r) return [(rows, region[0][0]["top"])] @staticmethod def _norm_acct(s): return re.sub(r"\(cid:\d+\)", "", (s or "")).strip().lower() # ----- assemble sections into statements ----- def _assemble(self, events): cur = None def fresh(): return ExtractedStatement() for _top, _pri, kind, val in events: if kind == "account": same = (cur is not None and cur.account is not None and self._norm_acct(cur.account) == self._norm_acct(val)) # a repeated header for the SAME account is a page continuation, not a new statement if cur and not cur.empty and not same and (cur.transactions or cur.closing_balance is not None): self.statements.append(cur); cur = fresh() if cur is None: cur = fresh() if cur.account is None or not same: cur.account = val elif kind == "holder": cur = cur or fresh(); cur.account_holder = val elif kind == "period": cur = cur or fresh(); cur.period = val elif kind == "opening": # only split when the previous statement actually closed; a repeated # "brought forward" mid-run is a page continuation, not a new statement if cur and cur.closing_balance is not None: self.statements.append(cur); cur = fresh() cur = cur or fresh() if cur.opening_balance is None: cur.opening_balance = parse_money(val) elif kind == "closing": cur = cur or fresh(); cur.closing_balance = parse_money(val) elif kind == "table": summ = self._summary_balances(val) # Beginning/Ending balance table? if summ: if cur and cur.transactions: # new summary block => new statement self.statements.append(cur); cur = fresh() cur = cur or fresh() cur.opening_balance, cur.closing_balance = summ else: cur = cur or fresh() cur.transactions += self._parse_rows(val) if cur and not cur.empty: self.statements.append(cur) @staticmethod def _summary_balances(table): """Detect a Beginning/Ending-balance summary table -> (opening, closing) or None.""" if not table or len(table) < 2: return None header = [(c or "").strip().lower() for c in table[0]] if any("date" in h for h in header): return None o = next((i for i, h in enumerate(header) if "beginning balance" in h or "opening balance" in h), None) cl = next((i for i, h in enumerate(header) if "ending balance" in h or "closing balance" in h), None) if o is None or cl is None: return None vals = table[1] op = parse_money(vals[o]) if o < len(vals) else None clo = parse_money(vals[cl]) if cl < len(vals) else None return (op, clo) if op is not None and clo is not None else None def _parse_rows(self, table): if not table or len(table) < 2: return [] header = [(c or "").strip() for c in table[0]] idx = {} for j, h in enumerate(header): key = map_header(h) if key and key not in idx: idx[key] = j if "date" not in idx or not ({"balance", "amount", "debit", "credit"} & idx.keys()): return [] out = [] for row in table[1:]: cells = [(c or "").strip() for c in row] if "date" not in idx or idx["date"] >= len(cells): continue raw_date = cells[idx["date"]] if not raw_date or not re.search(r"\d", raw_date): continue debit = credit = None if "debit" in idx or "credit" in idx: debit = parse_money(cells[idx["debit"]]) if "debit" in idx and idx["debit"] < len(cells) else None credit = parse_money(cells[idx["credit"]]) if "credit" in idx and idx["credit"] < len(cells) else None elif "amount" in idx and idx["amount"] < len(cells): amt = parse_money(cells[idx["amount"]]) if amt is not None: if amt < 0: debit = -amt else: credit = amt balance = parse_money(cells[idx["balance"]]) if "balance" in idx and idx["balance"] < len(cells) else None out.append(Transaction(norm_date(raw_date), cells[idx["description"]] if "description" in idx and idx["description"] < len(cells) else "", debit, credit, balance)) return out def _reconcile(self, st, tol=0.02): txns = st.transactions credits = sum(t.credit or 0.0 for t in txns) debits = sum(t.debit or 0.0 for t in txns) parts = [] # (1) running-balance integrity — validates the balance column directly (label-free) integrity = None if txns and all(t.balance is not None for t in txns): derived_open = round(txns[0].balance - ((txns[0].credit or 0) - (txns[0].debit or 0)), 2) prev, bad = derived_open, 0 for t in txns: if abs(round(prev + (t.credit or 0) - (t.debit or 0), 2) - t.balance) > tol: bad += 1 prev = t.balance integrity = (bad == 0) parts.append(f"running-balance {'✓' if integrity else f'✗ ({bad} bad)'} " f"(derived {derived_open:,.2f}→{txns[-1].balance:,.2f})") # (2) stated-summary match — opening + credits - debits == closing summary = None if st.opening_balance is not None and st.closing_balance is not None and txns: summary = abs((st.opening_balance + credits - debits) - st.closing_balance) <= tol parts.append(f"stated-summary {'✓' if summary else '✗'} " f"({st.opening_balance:,.2f}+{credits:,.2f}-{debits:,.2f} vs {st.closing_balance:,.2f})") st.running_balance_ok = integrity st.stated_summary_ok = summary present = [c for c in (integrity, summary) if c is not None] if not present: st.reconciles = None st.reconcile_status = "unverified" st.reconciliation_detail = "No balance column or stated totals to reconcile." else: # `reconciles` stays "either proof is enough" for backwards-compat; # `reconcile_status` is the precise signal the UI uses. st.reconciles = any(present) if all(present): st.reconcile_status = "verified" elif any(present): st.reconcile_status = "review" # one check passes, another disagrees else: st.reconcile_status = "failed" st.reconciliation_detail = " | ".join(parts) # ----- export ----- def to_excel(self, out_path): wb = Workbook(); wb.remove(wb.active) for n, st in enumerate(self.statements, 1): ws = wb.create_sheet(f"Account{n}") ws.append(["Date", "Description", "Debit", "Credit", "Balance"]) for t in st.transactions: ws.append([t.date, t.description, t.debit, t.credit, t.balance]) wb.save(out_path); return Path(out_path) def to_csv(self, out_path): with open(out_path, "w", newline="") as f: w = csv.writer(f) w.writerow(["Account", "Date", "Description", "Debit", "Credit", "Balance"]) for st in self.statements: for t in st.transactions: w.writerow([st.account or "", t.date, t.description, t.debit, t.credit, t.balance]) return Path(out_path) # ---------------- demo ---------------- def _print(statements): flag = {True: "✅ PASS", False: "❌ FAIL", None: "⚠️ N/A"} print("=" * 60) print(f"STATEMENTS FOUND: {len(statements)}") for n, st in enumerate(statements, 1): print("-" * 60) print(f"[{n}] account={st.account} holder={st.account_holder}") print(f" open/close: {st.opening_balance} / {st.closing_balance} txns: {len(st.transactions)}") print(f" reconcile : {flag[st.reconciles]} | {st.reconciliation_detail}") print("=" * 60) if __name__ == "__main__": pdf = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(__file__).parent / "sample_statement.pdf" conv = BankStatementConverter(pdf) sts = conv.extract() _print(sts) conv.to_excel(pdf.with_suffix(".out.xlsx")) conv.to_csv(pdf.with_suffix(".out.csv"))