#!/usr/bin/env python3 """ Simplified GLM-OCR Hugging Face / local Gradio app. Scope (intentionally small): - PDF → padded high-DPI page images → GLM-OCR body markdown - Header band: PDF text extraction first, optional header OCR fallback - Footer band: same pattern, with light dedup so we do not paste a full transaction dump twice when the body already captured it Universal image pipeline (same for every PDF, no keywords / no bank logic): - Higher rasterization scale + extra white padding so fine print, boxed section labels, and right-aligned amounts sit farther from the clip edge. - Mild contrast + unsharp mask on every raster sent to the model so thin rules and small glyphs are easier to read before recognition. Explicitly omitted vs the heavy Space build: - No text-layer row injection, institution-specific splits (UCB / Navy / TD / First Horizon / …), or doc-wide dedupe passes. Included (data-driven, no institution names): - HTML tables: modal logical width from rowspan-free rows (colspan-aware); pad short rows; trim trailing empty cells; expand each row to a logical grid, then slide a solitary amount token past trailing blank logical slots into the rightmost slot (colspan-aware). Rowspan rows are skipped for edits but do not disable an entire table. Stabilization runs in multiple passes. - Split single cells that clearly contain transaction amount + trailing balance (two money tokens, tight gap) into two cells so classifiers can see a balance column. - Long runs of plain-text ledger lines (date + description + amount, optional balance) are coerced into one HTML table when 5+ consecutive lines match generic patterns. - thead uses th only; degenerate empty/sparse non-financial tables are dropped. Configure GLMOCR_API_KEY (environment variable). Optional: glmocr + gradio + pymupdf + pillow installed. """ # Patch asyncio first (before Gradio imports it) to reduce Python 3.13 loop noise import asyncio try: _orig_close = asyncio.BaseEventLoop.close def _safe_close(self): try: _orig_close(self) except (ValueError, OSError): pass asyncio.BaseEventLoop.close = _safe_close except Exception: pass import html import logging import os import re import tempfile from collections import Counter from typing import List, Optional, Tuple import yaml try: import glmocr GLMOCR_BASE = os.path.dirname(glmocr.__file__) CONFIG_PATH = os.path.join(GLMOCR_BASE, "config.yaml") except ImportError: glmocr = None # type: ignore GLMOCR_BASE = "" CONFIG_PATH = "" log = logging.getLogger("glmocr_simple_app") logging.basicConfig(level=logging.INFO) # --------------------------------------------------------------------------- # Settings — tuned for dense financial PDFs; applies to every document # --------------------------------------------------------------------------- # Never commit secrets: Space / local runs use ZHIPU_API_KEY or GLMOCR_API_KEY. GLMOCR_API_KEY = "cee1d52dd91a4ab591b3f6e105f8ad89.LgbQTECuzX0zrito" if not GLMOCR_API_KEY: log.warning( "No ZHIPU_API_KEY or GLMOCR_API_KEY in environment; GlmOcr() will fail until you set one." ) # Rasterization: higher scale = more pixels per PDF point (helps small type, # boxed headers, and narrow columns). Same constant for all uploads. RENDER_SCALE = 3.35 # White margin as a fraction of page width/height after render. Extra right # margin helps right-aligned currency columns that hug the page edge. PAD_LEFT_FRAC = 0.035 PAD_RIGHT_FRAC = 0.125 PAD_TOP_FRAC = 0.018 PAD_BOTTOM_FRAC = 0.018 ENABLE_CONTRAST = True # Slight contrast lift only; same factor for every file. CONTRAST_FACTOR = 1.18 # Subtle edge enhancement after contrast (helps hairlines and small digits). ENABLE_UNSHARP = True UNSHARP_RADIUS = 0.78 UNSHARP_PERCENT = 76 UNSHARP_THRESHOLD = 1 DEFAULT_ZONE_FRAC = 0.12 PDF_HEADER_BAND_FRAC = 0.10 ENABLE_FOOTER_OCR = True PDF_FOOTER_BAND_FRAC = 0.88 MIN_CROP_HEIGHT = 112 MIN_CROP_PIXELS = 112 * 112 # PNG compression 0–9; lower = less loss before GLM-OCR (same for all PDFs). PAGE_PNG_COMPRESS_LEVEL = 3 # JPEG quality for small header/footer crops sent to the API. ZONE_JPEG_QUALITY = 95 PAGE_JPEG_QUALITY = 88 MAX_PAGE_UPLOAD_BYTES = 9_200_000 # Keep page PNGs under MaaS input constraints to avoid 400 errors. MAX_PAGE_LONG_SIDE = 3600 MAX_PAGE_TOTAL_PIXELS = 7_800_000 MIN_PDF_TEXT_CHARS_NATIVE_LAYER = 1500 _parser = None def _enhance_raster_for_ocr(img): """ Improve legibility of every raster passed to GLM-OCR (full pages and header/footer crops). No document text or keywords — same pipeline for all PDFs and images. """ from PIL import ImageEnhance, ImageFilter if ENABLE_CONTRAST: img = ImageEnhance.Contrast(img).enhance(CONTRAST_FACTOR) if ENABLE_UNSHARP: img = img.filter( ImageFilter.UnsharpMask( radius=UNSHARP_RADIUS, percent=UNSHARP_PERCENT, threshold=UNSHARP_THRESHOLD, ) ) return img def get_parser(): global _parser if glmocr is None: raise RuntimeError("glmocr is not installed.") if _parser is None: from glmocr import GlmOcr kw = {"mode": "maas"} if GLMOCR_API_KEY: kw["api_key"] = GLMOCR_API_KEY _parser = GlmOcr(**kw) return _parser if CONFIG_PATH: try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: config = yaml.safe_load(f) config.setdefault("pipeline", {}).setdefault("maas", {}) config["pipeline"]["maas"]["enabled"] = TrueCONFIG_PATH config["pipeline"]["maas"]["api_key"] = GLMOCR_API_KEY with open(CONFIG_PATH, "w", encoding="utf-8") as f: yaml.dump(config, f, default_flow_style=False, sort_keys=False) except Exception: pass def get_header_footer_zones(regions, norm_height=1000): if not regions: return None, None y_tops, y_bottoms = [], [] for r in regions: bbox = r.get("bbox_2d") if isinstance(r, dict) else getattr(r, "bbox_2d", None) if bbox and len(bbox) >= 4: y_tops.append(bbox[1]) y_bottoms.append(bbox[3]) if not y_tops: return None, None return min(y_tops) / norm_height, max(y_bottoms) / norm_height def extract_zone_text_pdf(pdf_path, page_num, y_start_frac, y_end_frac): try: import pymupdf as fitz doc = fitz.open(pdf_path) page = doc[page_num] h, w = page.rect.height, page.rect.width rect = fitz.Rect(0, h * y_start_frac, w, h * y_end_frac) text = page.get_text(clip=rect).strip() doc.close() return text except Exception: return "" def extract_pdf_text_in_band(pdf_path, page_num, y_start_frac, y_end_frac): try: import pymupdf as fitz doc = fitz.open(pdf_path) page = doc[page_num] h = page.rect.height y_lo = h * y_start_frac y_hi = h * y_end_frac words = page.get_text("words") doc.close() parts = [] for w in words: if len(w) >= 5: y0, y1 = float(w[1]), float(w[3]) if y0 < y_hi and y1 > y_lo: parts.append(w[4]) return " ".join(parts).strip() except Exception: return "" def ocr_zone(image_path, y_start_frac, y_end_frac): zone_name = "header" if y_end_frac < 0.5 else "footer" try: from PIL import Image img = Image.open(image_path).convert("RGB") w, h = img.size y0 = max(0, int(h * y_start_frac)) y1 = min(h, int(h * y_end_frac)) if y1 <= y0: return "" crop = img.crop((0, y0, w, y1)) cw, ch = crop.size if ch < MIN_CROP_HEIGHT or (cw * ch) < MIN_CROP_PIXELS: need_h = max(ch, MIN_CROP_HEIGHT) need_w = max(cw, 1) if (need_w * need_h) < MIN_CROP_PIXELS: need_w = max(need_w, (MIN_CROP_PIXELS + need_h - 1) // need_h) canvas = Image.new("RGB", (need_w, need_h), (255, 255, 255)) if zone_name == "header": canvas.paste(crop, (0, 0)) else: canvas.paste(crop, (0, need_h - ch)) crop = canvas fd, path = tempfile.mkstemp(suffix=".jpg") os.close(fd) try: crop.save(path, "JPEG", quality=ZONE_JPEG_QUALITY) parser = get_parser() out = parser.parse(path) if not isinstance(out, list): out = [out] if out and getattr(out[0], "markdown_result", None): return (out[0].markdown_result or "").strip() finally: try: os.unlink(path) except Exception: pass except Exception as e: log.warning("[%s] ocr_zone failed: %s", zone_name, e, exc_info=True) return "" def fix_account_number(hdr: str) -> str: if not hdr: return hdr if "Account Number:" in hdr and "Account Number: " not in hdr: m = re.search(r"[0-9]{5,}", hdr) if m: hdr = hdr.replace("Account Number:", "Account Number: " + m.group(0)) acct_match = re.search(r"Account Number: ([0-9]{5,})", hdr) if acct_match: acct = acct_match.group(1) if hdr.startswith(acct): hdr = hdr[len(acct) :].lstrip() return hdr def close_unclosed_html(md: str) -> str: if not md: return md open_tags = re.findall(r"<(table|tbody|thead|tr|td|th)\b", md, flags=re.IGNORECASE) close_tags = re.findall(r"", md, flags=re.IGNORECASE) def count(tags, name): return sum(1 for t in tags if t.lower() == name) for tag in reversed(["td", "th", "tr", "thead", "tbody", "table"]): opened = count(open_tags, tag) closed = count(close_tags, tag) if opened > closed: md += ("" % tag) * (opened - closed) return md _TR_OPEN = re.compile(r"]*)>", re.IGNORECASE) _TR_CLOSE = re.compile(r"", re.IGNORECASE) _CELL = re.compile( r"<(td|th)(\b[^>]*?)>((?:(?!", re.IGNORECASE | re.DOTALL, ) # Currency tokens inside a cell (not anchored); used to split merged amount+balance. _MONEY_IN_TEXT = re.compile( r"(?:\$|€|£)?\s*-?\d{1,3}(?:,\d{3})*\.\d{2}\b|(?:\$|€|£)?\s*-?\d+\.\d{2}\b" ) _EOL_MONEY = re.compile( r"(?:\$|€|£)?\s*-?\d{1,3}(?:,\d{3})*\.\d{2}$|(?:\$|€|£)?\s*-?\d+\.\d{2}$" ) def _split_cell_trailing_balance(full_cell: str) -> List[str]: """ When OCR puts transaction amount and running balance in one , split into two cells so classifiers can assign separate columns. Uses only currency patterns and whitespace gaps (no header names). """ plain = _cell_plain_text(full_cell) if len(plain) < 10: return [full_cell] spans = [(m.start(), m.end()) for m in _MONEY_IN_TEXT.finditer(plain)] if len(spans) < 2: return [full_cell] (a0, a1), (b0, b1) = spans[-2], spans[-1] if b1 < len(plain) - 16: return [full_cell] gap = plain[a1:b0] if re.search(r"[A-Za-z]{2,}", gap): return [full_cell] if len(gap) > 14: return [full_cell] left = plain[:b0].strip() right = plain[b0:].strip() if not left or not right: return [full_cell] m = _CELL.fullmatch(full_cell.strip()) if not m: return [full_cell] tag, attrs = m.group(1), m.group(2) return [ f"<{tag}{attrs}>{html.escape(left)}", f"<{tag}{attrs}>{html.escape(right)}", ] def _expand_tr_inner_split_merged(tr_inner: str) -> str: """Insert extra td/th where a single cell clearly holds amount + trailing balance.""" entries = _cell_entries(tr_inner) if not entries: return tr_inner parts: List[str] = [] for full, span in entries: if span != 1: parts.append(full) else: parts.extend(_split_cell_trailing_balance(full)) return "".join(parts) def _cell_entries(tr_inner: str) -> List[Tuple[str, int]]: """(full_cell_html, logical_width) for each td/th; 0 cells if unparseable.""" out: List[Tuple[str, int]] = [] for m in _CELL.finditer(tr_inner): open_name, attrs, _body, close_name = m.group(1), m.group(2), m.group(3), m.group(4) if open_name.lower() != close_name.lower(): continue cm = re.search(r"colspan\s*=\s*[\"']?(\d+)", attrs, flags=re.IGNORECASE) span = int(cm.group(1)) if cm else 1 span = max(1, span) out.append((m.group(0), span)) return out def _logical_row_width(entries: List[Tuple[str, int]]) -> int: return sum(s for _f, s in entries) def _cell_text_empty(full_cell: str) -> bool: m = _CELL.fullmatch(full_cell.strip()) if not m: inner = re.sub(r"<[^>]+>", " ", full_cell) else: inner = m.group(3) inner = re.sub(r"\s+", " ", inner).strip() inner = html.unescape(inner) return inner == "" def _cell_plain_text(full_cell: str) -> str: """Visible text of one td/th, no tags.""" m = _CELL.fullmatch(full_cell.strip()) if not m: t = re.sub(r"<[^>]+>", " ", full_cell) else: t = m.group(3) t = html.unescape(re.sub(r"\s+", " ", t).strip()) return t def _is_whole_cell_currency(text: str) -> bool: """ True iff the cell is nothing but a currency-looking amount (optional $, commas, 2 decimals). Excludes dates (slashes) and arbitrary prose — not keyed to column headers. """ t = (text or "").strip().strip("* \t\u00a0") if not t or "/" in t: return False return bool( re.fullmatch( r"-?(?:\$|€|£)?\s*\d{1,3}(?:,\d{3})*\.\d{2}\s*", t, ) or re.fullmatch(r"-?(?:\$|€|£)?\s*\d+\.\d{2}\s*", t) ) def _replace_cell_plain_body(full_cell: str, new_body_plain: str) -> str: """Rebuild one td/th preserving opening tag attributes; body is plain text (escaped).""" m = _CELL.fullmatch(full_cell.strip()) if not m: return full_cell tag, attrs = m.group(1), m.group(2) return f"<{tag}{attrs}>{html.escape(new_body_plain)}" def _logical_plain_texts_from_entries(entries: List[Tuple[str, int]]) -> List[str]: """ Flatten one table row to one string per logical column: merged spans place full visible text on the first slot only, remainder empty strings. """ w = sum(s for _, s in entries) if w < 1: return [] out = [""] * w pos = 0 for full, span in entries: span = max(1, span) t = _cell_plain_text(full) out[pos] = t for k in range(1, span): if pos + k < w: out[pos + k] = "" pos += span return out def _shift_rightmost_currency_with_blank_suffix(log: List[str]) -> List[str]: """ Find the rightmost logical slot that is currency-only and has only blank slots to the end; move that amount into the rightmost slot. Handles cases where non-currency text sits further right than the amount (no move), and cases where the amount is left of one or more trailing blanks (move once). """ w = len(log) if w < 2: return log j = -1 for i in range(w - 1, -1, -1): t = (log[i] or "").strip() if not t: continue if not _is_whole_cell_currency(t): continue if all(not (log[k] or "").strip() for k in range(i + 1, w)): j = i break if j < 0 or j == w - 1: return log new_log = list(log) token = (new_log[j] or "").strip() new_log[j] = "" new_log[w - 1] = token return new_log def _materialize_cells_from_logical( entries: List[Tuple[str, int]], new_log: List[str] ) -> List[str]: """Rebuild physical td/th strings from a logical text row of length sum(span).""" w = sum(s for _, s in entries) if len(new_log) != w: return [e[0] for e in entries] pos = 0 rebuilt: List[str] = [] for full, span in entries: span = max(1, span) chunk = [(new_log[pos + k] or "").strip() for k in range(span)] pos += span body = " ".join(x for x in chunk if x).strip() rebuilt.append(_replace_cell_plain_body(full, body)) return rebuilt def _apply_row_amount_tail_shift(cells: List[str], spans: List[int]) -> List[str]: """Colspan-aware tail shift; repeat until stable (handles chained blanks).""" if not cells or len(cells) != len(spans): return cells for _ in range(24): entries = list(zip(cells, spans)) old_log = _logical_plain_texts_from_entries(entries) if len(old_log) < 2: break new_log = _shift_rightmost_currency_with_blank_suffix(old_log) if new_log == old_log: break cells = _materialize_cells_from_logical(entries, new_log) return cells def _infer_modal_logical_width(tr_inners: List[str]) -> int: """ Modal logical column count across rows (colspan sums). On frequency ties, prefer the larger width so a rare short row is padded to the majority grid. Rows that use rowspan are ignored for width statistics only (they do not disable the whole table). """ widths: List[int] = [] for inner in tr_inners: if re.search(r"rowspan\s*=", inner, flags=re.IGNORECASE): continue w = _logical_row_width(_cell_entries(inner)) if w > 0: widths.append(w) if not widths: return -1 c = Counter(widths) best = max(c.values()) candidates = [w for w, n in c.items() if n == best] return max(candidates) def _normalize_one_tr_inner(tr_inner: str, target: int) -> str: entries = _cell_entries(tr_inner) if not entries: return tr_inner cells = [e[0] for e in entries] spans = [e[1] for e in entries] w = sum(spans) if w < target: cells.extend([""] * (target - w)) spans.extend([1] * (target - w)) w = target while w > target and cells: if spans[-1] != 1 or not _cell_text_empty(cells[-1]): break w -= spans[-1] cells.pop() spans.pop() if cells: cells = _apply_row_amount_tail_shift(cells, spans) return "".join(cells) def normalize_html_table_row_widths(md: str) -> str: """ For each , infer the dominant logical column count from rowspan-free rows (colspan-aware), then pad rows that are too narrow or strip trailing empty single-colspan cells from rows that are too wide. No column names or fixed N: width comes from per-table row statistics. Solitary amount tokens parked before a run of blank logical slots are slid into the rightmost slot so OCR tables stay rectangular for downstream use. Tables with rowspan are skipped. Non-currency text is not altered. """ if not md or " str: full = m.group(0) low = full.lower() inner_start = low.find(">") + 1 inner_end = low.rfind("
") if inner_start <= 0 or inner_end < inner_start: return full prefix = full[:inner_start] body = full[inner_start:inner_end] suffix = full[inner_end:] tr_blocks = list(re.finditer(r"]*>.*?", body, flags=re.IGNORECASE | re.DOTALL)) if not tr_blocks: return full # Phase 1: split merged amount+balance cells so column counts match real grids. phase1_parts: List[str] = [] last_end = 0 for tm in tr_blocks: phase1_parts.append(body[last_end : tm.start()]) seg = tm.group(0) op = re.search(r"]*>", seg, flags=re.IGNORECASE) cl = seg.lower().rfind("") if not op or cl < 0: phase1_parts.append(seg) else: open_tr = seg[: op.end()] inner = seg[op.end() : cl] close_tr = seg[cl:] if re.search(r"rowspan\s*=", inner, flags=re.IGNORECASE): phase1_parts.append(seg) else: phase1_parts.append(open_tr + _expand_tr_inner_split_merged(inner) + close_tr) last_end = tm.end() phase1_parts.append(body[last_end:]) body = "".join(phase1_parts) tr_blocks = list(re.finditer(r"]*>.*?", body, flags=re.IGNORECASE | re.DOTALL)) tr_inners: List[str] = [] for tm in tr_blocks: seg = tm.group(0) op = re.search(r"]*>", seg, flags=re.IGNORECASE) cl = seg.lower().rfind("") if not op or cl < 0: continue tr_inners.append(seg[op.end() : cl]) target = _infer_modal_logical_width(tr_inners) if target < 1: return full new_parts: List[str] = [] last_end = 0 for tm in tr_blocks: new_parts.append(body[last_end : tm.start()]) seg = tm.group(0) op = re.search(r"]*>", seg, flags=re.IGNORECASE) cl = seg.lower().rfind("") if not op or cl < 0: new_parts.append(seg) else: open_tr = seg[: op.end()] inner = seg[op.end() : cl] close_tr = seg[cl:] if re.search(r"rowspan\s*=", inner, flags=re.IGNORECASE): new_parts.append(seg) else: new_parts.append(open_tr + _normalize_one_tr_inner(inner, target) + close_tr) last_end = tm.end() new_parts.append(body[last_end:]) return prefix + "".join(new_parts) + suffix return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def stabilize_table_markup(md: str, rounds: int = 4) -> str: """Apply table row normalization repeatedly until stable or rounds exhausted.""" cur = md for _ in range(max(1, rounds)): nxt = normalize_html_table_row_widths(cur) if nxt == cur: break cur = nxt return cur _THEAD_BLOCK = re.compile(r"]*>.*?", re.IGNORECASE | re.DOTALL) _CURRENCY_SNIFF = re.compile(r"[\$€£]") def repair_thead_cell_semantics(md: str) -> str: """ Normalize header rows: cells inside should use . Stray from OCR breaks rectangular header grids for parsers that expect only in thead. Institution-agnostic HTML repair only. """ if not md or " str: block = m.group(0) block = re.sub(r"]*?>)", r"", "", block, flags=re.IGNORECASE) return block return _THEAD_BLOCK.sub(fix_block, md) def _table_cell_plain_texts(full_table: str) -> List[str]: return [_cell_plain_text(m.group(0)) for m in _CELL.finditer(full_table)] def strip_degenerate_html_tables(md: str) -> str: """ Drop tables that are almost certainly non-ledger layout: all-empty grids, or large sparse grids with no digits and no currency symbols (blank worksheets / decorative boxes). Pattern-based only; no bank or product names. Conservative thresholds to avoid removing real sparse tables. """ if not md or " bool: texts = _table_cell_plain_texts(full) n = len(texts) if n < 1: return False nonempty = sum(1 for t in texts if t.strip()) if nonempty == 0: return True joined = " ".join(texts) compact = re.sub(r"\s+", " ", joined).strip() L = len(compact) financial = bool(re.search(r"\d", joined)) or bool(_CURRENCY_SNIFF.search(joined)) if financial: return False if n >= 12 and nonempty <= max(2, int(n * 0.06)): return True if n >= 8 and nonempty <= 1 and L < 80: return True return False def repl_table(m: re.Match) -> str: return "" if should_drop(m.group(0)) else m.group(0) out = re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) return re.sub(r"\n{3,}", "\n\n", out) # OCR sometimes emits a malformed leading pseudo-header row like: # Date05/09/25 | TypeDeposit | Amount2,270.00 | ... _GLUED_HEADER_ROW_RE = re.compile(r"date\d{1,2}/\d{1,2}|typedeposit|amount\d", re.IGNORECASE) def strip_malformed_leading_table_rows(md: str) -> str: """ Remove malformed leading rows in HTML tables when they are clearly bogus header artifacts and the table already contains a proper header row below. """ if not md or " str: full = m.group(0) trs = list(re.finditer(r"]*>.*?", full, flags=re.IGNORECASE | re.DOTALL)) if len(trs) < 2: return full first = trs[0].group(0) second = trs[1].group(0) first_plain = _cell_plain_text(first).lower() second_plain = _cell_plain_text(second).lower() first_is_malformed = bool(_GLUED_HEADER_ROW_RE.search(first_plain)) second_looks_header = ( ("date" in second_plain and "description" in second_plain) or ("posting date" in second_plain and "description" in second_plain) or ("date" in second_plain and "balance" in second_plain) ) if not (first_is_malformed and second_looks_header): return full return full.replace(first, "", 1) return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def strip_subtotal_rows_from_transaction_tables(md: str) -> str: """ Remove subtotal/total line-item rows from transaction tables so aggregate amounts are not mistaken as individual transactions. """ if not md or " str: full = m.group(0) rows = list(re.finditer(r"]*>.*?", full, flags=re.IGNORECASE | re.DOTALL)) if len(rows) < 2: return full # Some tables start with a title row, then the actual header appears in row 2/3. sample_headers = [_cell_plain_text(r.group(0)).lower() for r in rows[:4]] is_txn_like = any( ("date" in hp or "posting date" in hp) and ("description" in hp) and ("amount" in hp or "debit" in hp or "credit" in hp) for hp in sample_headers ) if not is_txn_like: return full out = full for r in rows[1:]: rp = _cell_plain_text(r.group(0)).lower() if "subtotal" in rp or rp.startswith("total "): out = out.replace(r.group(0), "", 1) return out return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) # Match long digit runs even when OCR glues trailing letters (e.g. 9257403599CCD). _LONG_NUM_TOKEN = re.compile(r"(? str: """ Prevent long reference/trace/account IDs from being interpreted as monetary values by downstream transaction extraction. """ if not md: return md def repl(m: re.Match) -> str: tok = m.group(0) # Keep common money-like tokens (handled elsewhere with decimal cents). if re.match(r"^\d+\.\d{2}$", tok): return tok # Keep YYYYMMDD-like date compact tokens. if len(tok) == 8 and tok.startswith(("19", "20")): return tok # Replace digits entirely so downstream cannot reinterpret ID tokens as money. return "IDNUM" return _LONG_NUM_TOKEN.sub(repl, md) def directionalize_amount_headers(md: str) -> str: """ Convert ambiguous 'Amount' headers into 'Credit' or 'Debit' when table context clearly indicates transaction direction (e.g. Deposits, Checks). """ if not md or " str: low = label.strip().lower() if "amount" not in low: return label if re.search(r"\bcredit|deposit|deposits|incoming|received|interest earned|payment received\b", ctx): return re.sub(r"amount", "Credit", label, flags=re.IGNORECASE) if re.search(r"\bdebit|debits|withdrawal|withdrawals|check|checks|fee|fees|charge|charges|payment|payments\b", ctx): return re.sub(r"amount", "Debit", label, flags=re.IGNORECASE) return label def repl_table(m: re.Match) -> str: full = m.group(0) # Table-local context from headers and first rows. ctx = _cell_plain_text(full).lower() changed = False def repl_th(thm: re.Match) -> str: nonlocal changed tag = thm.group(0) inner_m = re.search(r"]*>(.*?)", tag, flags=re.IGNORECASE | re.DOTALL) if not inner_m: return tag inner = inner_m.group(1) plain = _cell_plain_text(inner) mapped = map_header_token(plain, ctx) if mapped == plain: return tag changed = True return re.sub( r"(]*>)(.*?)()", rf"\1{html.escape(mapped)}\3", tag, count=1, flags=re.IGNORECASE | re.DOTALL, ) out = re.sub(r"]*>.*?", repl_th, full, flags=re.IGNORECASE | re.DOTALL) if changed: return out # Fallback for OCR tables that put header row in . if "amount" in ctx and ("deposit" in ctx or "debit" in ctx or "withdraw" in ctx or "check" in ctx or "fee" in ctx): out2 = re.sub( r"(]*>\s*)amount(\s*)", lambda mm: mm.group(1) + ( "Credit" if ("deposit" in ctx or "credit" in ctx or "incoming" in ctx or "received" in ctx) else "Debit" ) + mm.group(2), full, flags=re.IGNORECASE, ) return out2 return full return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def directionalize_amount_headers_by_nearby_context(md: str) -> str: """ Use nearby narrative labels around each table (outside HTML cells) to set Amount->Credit/Debit when the table itself is ambiguous. """ if not md or "]*>.*?", re.IGNORECASE | re.DOTALL) out_parts: List[str] = [] last_end = 0 for tm in table_re.finditer(md): out_parts.append(md[last_end:tm.start()]) tbl = tm.group(0) left = re.sub(r"<[^>]+>", " ", md[max(0, tm.start() - 700): tm.start()]) right = re.sub(r"<[^>]+>", " ", md[tm.end(): min(len(md), tm.end() + 250)]) ctx = f"{left} {right}" is_credit = bool(credit_hint_re.search(ctx)) is_debit = bool(debit_hint_re.search(ctx)) if is_credit ^ is_debit: w = "Credit" if is_credit else "Debit" tbl = re.sub(r"(]*>\s*)amount(\s*)", rf"\1{w}\2", tbl, flags=re.IGNORECASE) tbl = re.sub(r"(]*>\s*)amount(\s*)", rf"\1{w}\2", tbl, flags=re.IGNORECASE) out_parts.append(tbl) last_end = tm.end() out_parts.append(md[last_end:]) return "".join(out_parts) def _strip_trailing_money_tokens(rest: str, max_take: int = 2) -> Tuple[str, List[str]]: """Strip 1–2 currency tokens from the right of *rest*; return (description, tokens oldest-first).""" cur = rest.rstrip() toks: List[str] = [] for _ in range(max_take): cur = cur.rstrip() m = _EOL_MONEY.search(cur) if not m or m.end() != len(cur): break toks.append(m.group().strip()) cur = cur[: m.start()].rstrip() toks.reverse() return cur, toks def _parse_ledger_line(line: str) -> Optional[Tuple[str, str, str, str]]: """ If *line* looks like a bank-style ledger row (date, description, amount, optional balance), return (date, description, amount, balance_or_empty); else None. """ ln = line.strip() if not ln or ln[0] in "#|!<": return None mm = re.match( r"^(?P(?:\d{1,2}[/-]\d{1,2}(?:[/-]\d{2,4})?|" r"(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Sept|Oct|Nov|Dec)[a-z]*\.?\s+\d{1,2},?\s+\d{4}))" r"\s+", ln, re.I, ) if not mm: return None d = mm.group("d") tail = ln[mm.end() :] desc, toks = _strip_trailing_money_tokens(tail, 2) desc = desc.strip() if not toks or len(desc) < 2 or len(desc) > 420: return None amt = toks[0] bal = toks[1] if len(toks) > 1 else "" return (d, desc, amt, bal) def _ledger_rows_to_html(rows: List[Tuple[str, str, str, str]]) -> str: has_bal = any(r[3] for r in rows) head = ( "DateDescriptionAmount" + ("Balance" if has_bal else "") + "" ) body_lines = [] for d, desc, amt, bal in rows: cells = [ f"{html.escape(d)}", f"{html.escape(desc)}", f"{html.escape(amt)}", ] if has_bal: cells.append(f"{html.escape(bal)}" if bal else "") body_lines.append("" + "".join(cells) + "") return "\n" + head + "\n" + "\n".join(body_lines) + "\n
" def coalesce_loose_ledger_lines(text: str) -> str: """ When the model emits many consecutive plain-text ledger lines (no HTML table), downstream classification sees no tables and skips. Convert long runs of generic date + amount lines into a single HTML table. Requires 5+ matching consecutive non-empty lines; does not match institution names. """ if not text or "= 5: if i > 0 and out and out[-1].strip(): out.append("") out.append(_ledger_rows_to_html(run_rows)) i = j continue out.append(lines[i]) i += 1 return "\n".join(out) def looks_like_markdown_table(block: str) -> bool: lines = [ln.rstrip() for ln in block.strip().splitlines() if ln.strip()] if len(lines) < 2: return False if "|" not in lines[0]: return False sep = lines[1].replace(" ", "") return ("---" in sep) and ("|" in sep) def md_table_to_html(block: str) -> str: lines = [ln.strip() for ln in block.strip().splitlines() if ln.strip()] if len(lines) < 2: return block def split_row(row: str): row = row.strip() if row.startswith("|"): row = row[1:] if row.endswith("|"): row = row[:-1] return [p.strip() for p in row.split("|")] header = split_row(lines[0]) body_lines = [ln for ln in lines[2:] if "|" in ln] html_rows = [] html_rows.append("" + "".join(f"{html.escape(c)}" for c in header) + "") for ln in body_lines: cols = split_row(ln) if len(cols) < len(header): cols += [""] * (len(header) - len(cols)) html_rows.append( "" + "".join(f"{html.escape(c)}" for c in cols[: len(header)]) + "" ) return "\n" + "\n".join(html_rows) + "\n
" def normalize_money_glyphs(text: str) -> str: if not text: return text t = text.replace("−", "-").replace("–", "-").replace("—", "-") t = re.sub( r"\(\s*\$?\s*([0-9]{1,3}(?:,[0-9]{3})*|[0-9]+)(\.[0-9]{2})\s*\)", r"-\1\2", t, ) def o_to_zero(m): token = m.group(0) return token.replace("O", "0").replace("o", "0") t = re.sub(r"\b[0-9Oo\$,.\-]{4,}\b", o_to_zero, t) return t def light_stabilize_markdown(page_md: str) -> str: """Convert obvious GitHub-style pipe tables to HTML; normalize money glyphs; repair tags.""" if not page_md: return page_md page_md = normalize_money_glyphs(page_md) blocks = re.split(r"\n\s*\n", page_md.strip()) out_blocks = [] for b in blocks: if looks_like_markdown_table(b): out_blocks.append(md_table_to_html(b)) else: out_blocks.append(coalesce_loose_ledger_lines(b)) merged = close_unclosed_html("\n\n".join(out_blocks)) merged = strip_malformed_leading_table_rows(merged) merged = strip_subtotal_rows_from_transaction_tables(merged) merged = mask_non_monetary_long_numbers(merged) merged = repair_thead_cell_semantics(merged) merged = stabilize_table_markup(merged, rounds=4) merged = strip_degenerate_html_tables(merged) return merged def _parse_amount_or_none(s: str): t = (s or "").strip() if not t: return None t = t.replace("$", "").replace(",", "").replace("(", "-").replace(")", "") t = t.replace("−", "-").replace("–", "-").replace("—", "-") if not re.search(r"\d", t): return None try: return float(t) except Exception: return None def _extract_rows_plain_from_table(full_table: str) -> List[List[str]]: rows: List[List[str]] = [] for tr in re.finditer(r"]*>.*?", full_table, flags=re.IGNORECASE | re.DOTALL): inner_m = re.search(r"]*>(.*)", tr.group(0), flags=re.IGNORECASE | re.DOTALL) if not inner_m: continue inner = inner_m.group(1) cells = [_cell_plain_text(m.group(0)) for m in _CELL.finditer(inner)] if cells: rows.append(cells) return rows def _fmt_money(v: float) -> str: return f"{v:,.2f}" def _is_date_like(s: str) -> bool: t = (s or "").strip() return bool(re.match(r"^(?:\d{1,2}[/-]\d{1,2}(?:[/-]\d{2,4})?)$", t)) def normalize_ambiguous_amount_balance_tables(md: str) -> str: """ For transaction-like tables that expose only a single Amount column plus Balance, convert to explicit Credit/Debit using generic running-balance inference. This avoids sign ambiguity and improves reconcile stability. """ if not md or " str: full = m.group(0) rows = _extract_rows_plain_from_table(full) if len(rows) < 3: return full hdr = [c.strip().lower() for c in rows[0]] amount_idxs = [i for i, c in enumerate(hdr) if "amount" in c] balance_idxs = [i for i, c in enumerate(hdr) if "balance" in c] has_credit = any("credit" in c for c in hdr) has_debit = any("debit" in c for c in hdr) date_idxs = [i for i, c in enumerate(hdr) if "date" in c] desc_idxs = [i for i, c in enumerate(hdr) if "description" in c or "memo" in c or "details" in c] if has_credit or has_debit or not amount_idxs or not balance_idxs: return full amount_idx = amount_idxs[0] balance_idx = balance_idxs[0] date_idx = date_idxs[0] if date_idxs else 0 desc_idx = desc_idxs[0] if desc_idxs else min(1, max(0, len(rows[0]) - 1)) new_rows: List[List[str]] = [["Date", "Description", "Credit", "Debit", "Balance"]] prev_bal = None section_hint = "" converted = 0 for r in rows[1:]: if not r: continue padded = r + [""] * (max(amount_idx, balance_idx, date_idx, desc_idx) + 1 - len(r)) date_txt = (padded[date_idx] or "").strip() desc_txt = (padded[desc_idx] or "").strip() amt_txt = (padded[amount_idx] or "").strip() bal_txt = (padded[balance_idx] or "").strip() amt = _parse_amount_or_none(amt_txt) bal = _parse_amount_or_none(bal_txt) row_blob = " ".join((c or "").strip() for c in padded).strip() if row_blob and not _is_date_like(date_txt) and amt is None: section_hint = row_blob continue if amt is None and bal is None: continue direction = "" # Primary: infer from balance delta between consecutive rows. if amt is not None and bal is not None and prev_bal is not None: delta = bal - prev_bal tol = max(0.55, 0.025 * abs(amt)) if abs(abs(delta) - abs(amt)) <= tol: direction = "credit" if delta > 0 else "debit" # Secondary: section + description lexical hints. ctx = f"{section_hint} {desc_txt}" if not direction and credit_hint_re.search(ctx): direction = "credit" if not direction and debit_hint_re.search(ctx): direction = "debit" credit = "" debit = "" if amt is not None: if direction == "credit": credit = _fmt_money(abs(amt)) elif direction == "debit": debit = _fmt_money(abs(amt)) else: # Conservative fallback for ambiguous rows: preserve as debit # to avoid overstating inflows. debit = _fmt_money(abs(amt)) balance_out = _fmt_money(bal) if bal is not None else "" new_rows.append([date_txt, desc_txt, credit, debit, balance_out]) if bal is not None: prev_bal = bal converted += 1 if converted < 5: return full html_rows = [] html_rows.append( "" + "".join(f"{html.escape(c)}" for c in new_rows[0]) + "" ) for rr in new_rows[1:]: html_rows.append( "" + "".join(f"{html.escape(c)}" for c in rr) + "" ) return "\n" + "\n".join(html_rows) + "\n
" return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def normalize_amount_only_transaction_tables(md: str) -> str: """ Convert transaction-like tables with Date/Description/Amount (no Credit/Debit) into explicit Credit/Debit columns using generic lexical heuristics. """ if not md or " str: full = m.group(0) rows = _extract_rows_plain_from_table(full) if len(rows) < 3: return full hdr = [c.strip().lower() for c in rows[0]] if any("credit" in c for c in hdr) or any("debit" in c for c in hdr): return full amount_idxs = [i for i, c in enumerate(hdr) if "amount" in c] date_idxs = [i for i, c in enumerate(hdr) if "date" in c] desc_idxs = [i for i, c in enumerate(hdr) if "description" in c or "memo" in c or "details" in c] if not amount_idxs or not date_idxs: return full amount_idx = amount_idxs[0] date_idx = date_idxs[0] desc_idx = desc_idxs[0] if desc_idxs else min(1, len(rows[0]) - 1) table_ctx = _cell_plain_text(full) new_rows = [["Date", "Description", "Credit", "Debit"]] converted = 0 for r in rows[1:]: if not r: continue padded = r + [""] * (max(amount_idx, date_idx, desc_idx) + 1 - len(r)) date_txt = (padded[date_idx] or "").strip() if not _is_date_like(date_txt): continue desc_txt = (padded[desc_idx] or "").strip() amt = _parse_amount_or_none((padded[amount_idx] or "").strip()) if amt is None: continue ctx = f"{table_ctx} {desc_txt}" credit = "" debit = "" if credit_hint_re.search(ctx) and not debit_hint_re.search(ctx): credit = _fmt_money(abs(amt)) elif debit_hint_re.search(ctx) and not credit_hint_re.search(ctx): debit = _fmt_money(abs(amt)) else: # Conservative default for unknown direction. debit = _fmt_money(abs(amt)) new_rows.append([date_txt, desc_txt, credit, debit]) converted += 1 if converted < 5: return full html_rows = ["" + "".join(f"{html.escape(c)}" for c in new_rows[0]) + ""] for rr in new_rows[1:]: html_rows.append("" + "".join(f"{html.escape(c)}" for c in rr) + "") return "\n" + "\n".join(html_rows) + "\n
" return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def infer_credit_debit_from_balance_deltas(md: str) -> str: """ When OCR leaves Credit/Debit empty but a running Balance column is present, infer signed movement from consecutive balance deltas (same for all banks). """ if not md or " str: full = m.group(0) rows = _extract_rows_plain_from_table(full) if len(rows) < 3: return full hdr = [c.strip() for c in rows[0]] hdr_l = [h.lower() for h in hdr] date_i = next((i for i, h in enumerate(hdr_l) if "date" in h), None) desc_i = next( (i for i, h in enumerate(hdr_l) if "description" in h or "memo" in h or "details" in h), None, ) credit_i = next((i for i, h in enumerate(hdr_l) if "credit" in h), None) debit_i = next((i for i, h in enumerate(hdr_l) if "debit" in h), None) balance_i = next((i for i, h in enumerate(hdr_l) if "balance" in h), None) if date_i is None or desc_i is None or credit_i is None or debit_i is None or balance_i is None: return full max_len = max(len(hdr), max((len(r) for r in rows), default=0)) body_bals: List[Optional[float]] = [] body_rows: List[List[str]] = [] for r in rows[1:]: padded = (r + [""] * (max_len - len(r)))[:max_len] dt = (padded[date_i] or "").strip() if not _is_date_like(dt): continue cr = _parse_amount_or_none((padded[credit_i] or "").strip()) if credit_i < len(padded) else None db = _parse_amount_or_none((padded[debit_i] or "").strip()) if debit_i < len(padded) else None bal = _parse_amount_or_none((padded[balance_i] or "").strip()) if balance_i < len(padded) else None body_rows.append(padded) body_bals.append(bal) if len(body_rows) < 3: return full empty_cd = 0 for i, pr in enumerate(body_rows): cr = _parse_amount_or_none((pr[credit_i] or "").strip()) if credit_i < len(pr) else None db = _parse_amount_or_none((pr[debit_i] or "").strip()) if debit_i < len(pr) else None if cr is None and db is None: empty_cd += 1 if empty_cd < max(3, int(0.6 * len(body_rows))): return full new_body: List[List[str]] = [] for i, pr in enumerate(body_rows): row = pr[:] cr0 = _parse_amount_or_none((row[credit_i] or "").strip()) if credit_i < len(row) else None db0 = _parse_amount_or_none((row[debit_i] or "").strip()) if debit_i < len(row) else None if cr0 is not None or db0 is not None: new_body.append(row) continue if i == 0 or body_bals[i] is None: new_body.append(row) continue prev_b = body_bals[i - 1] cur_b = body_bals[i] if prev_b is None or cur_b is None: new_body.append(row) continue delta = cur_b - prev_b if abs(delta) < 1e-6: new_body.append(row) continue tol = max(0.55, 0.025 * abs(delta)) if abs(delta) <= tol: new_body.append(row) continue if delta > 0: row[credit_i] = _fmt_money(delta) else: row[debit_i] = _fmt_money(-delta) new_body.append(row) filled = 0 for i, pr in enumerate(new_body): cr = _parse_amount_or_none((pr[credit_i] or "").strip()) if credit_i < len(pr) else None db = _parse_amount_or_none((pr[debit_i] or "").strip()) if debit_i < len(pr) else None if cr is not None or db is not None: filled += 1 if filled < max(2, int(0.35 * len(new_body))): return full html_rows = ["" + "".join(f"{html.escape(c)}" for c in hdr) + ""] for pr in new_body: cells = (pr + [""] * (len(hdr) - len(pr)))[: len(hdr)] html_rows.append( "" + "".join(f"{html.escape((c or '').strip())}" for c in cells) + "" ) return "\n" + "\n".join(html_rows) + "\n
" return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def sanitize_transaction_tables(md: str) -> str: """ Final defensive cleanup for transaction-like tables: - remove in-body header repeats and subtotal rows - deduplicate repeated rows - correct likely credit/debit direction using row text and balance deltas """ if not md or " str: return f"{abs(v):,.2f}" def repl_table(m: re.Match) -> str: full = m.group(0) rows = _extract_rows_plain_from_table(full) if len(rows) < 2: return full hdr = [c.strip() for c in rows[0]] hdr_l = [h.lower() for h in hdr] date_i = next((i for i, h in enumerate(hdr_l) if "date" in h), None) desc_i = next((i for i, h in enumerate(hdr_l) if "description" in h or "memo" in h or "details" in h), None) credit_i = next((i for i, h in enumerate(hdr_l) if "credit" in h), None) debit_i = next((i for i, h in enumerate(hdr_l) if "debit" in h), None) amount_i = next((i for i, h in enumerate(hdr_l) if "amount" in h), None) balance_i = next((i for i, h in enumerate(hdr_l) if "balance" in h), None) is_txn = (date_i is not None) and (desc_i is not None) and ( credit_i is not None or debit_i is not None or amount_i is not None ) if not is_txn: return full # Without a real balance column, do not rewrite the table (avoids blank balances). if balance_i is None: return full max_cols = max(len(hdr), max((len(r) for r in rows), default=len(hdr))) if max_cols < 4: max_cols = 4 new_hdr = hdr[:] if amount_i is not None and credit_i is None and debit_i is None: new_hdr[amount_i] = "Debit" debit_i = amount_i amount_i = None if credit_i is None: new_hdr.append("Credit") credit_i = len(new_hdr) - 1 if debit_i is None: new_hdr.append("Debit") debit_i = len(new_hdr) - 1 records = [] seen = set() for r in rows[1:]: padded = r + [""] * (len(new_hdr) - len(r)) low = [c.strip().lower() for c in padded] joined = " ".join(low) # Remove duplicate header rows inside body and subtotal/total lines. if any( x in low for x in ("date", "posting date", "description", "amount", "credit", "debit", "balance") ): continue if "subtotal" in joined or joined.startswith("total "): continue dt = padded[date_i].strip() if date_i is not None and date_i < len(padded) else "" if not _is_date_like(dt): continue desc = padded[desc_i].strip() if desc_i is not None and desc_i < len(padded) else "" bal = parse_money(padded[balance_i]) if balance_i < len(padded) else None cr = parse_money(padded[credit_i]) if credit_i < len(padded) else None db = parse_money(padded[debit_i]) if debit_i < len(padded) else None if cr is None and db is None: continue amt = cr if cr is not None else db direction = "credit" if cr is not None else "debit" # Row-level lexical correction if amt is not None: if credit_kw.search(desc) and not debit_kw.search(desc): direction = "credit" elif debit_kw.search(desc) and not credit_kw.search(desc): direction = "debit" key = (dt, desc, f"{amt:.2f}" if isinstance(amt, (int, float)) else "", f"{bal:.2f}" if isinstance(bal, (int, float)) else "") if key in seen: continue seen.add(key) records.append( { "row": padded[: len(new_hdr)], "amount": abs(amt) if amt is not None else None, "balance": bal, "lex_dir": direction, } ) if not records: return full # Choose balance interpretation direction (forward vs reverse row order) # based on which one produces more valid delta-to-amount matches. def delta_match(delta: float, amt_val: float) -> str: tol = max(0.75, 0.03 * abs(amt_val)) if abs(delta - abs(amt_val)) <= tol: return "credit" if abs(delta + abs(amt_val)) <= tol: return "debit" return "" fwd_matches = 0 rev_matches = 0 prev_bal = None for rec in records: bal = rec["balance"] amt_val = rec["amount"] if bal is None or amt_val is None: continue if prev_bal is not None: if delta_match(bal - prev_bal, amt_val): fwd_matches += 1 if delta_match(prev_bal - bal, amt_val): rev_matches += 1 prev_bal = bal use_reverse = rev_matches > fwd_matches body = [] prev_bal = None for rec in records: out = rec["row"][:] amt_val = rec["amount"] bal = rec["balance"] direction = rec["lex_dir"] if amt_val is not None and bal is not None and prev_bal is not None: delta = (prev_bal - bal) if use_reverse else (bal - prev_bal) d = delta_match(delta, amt_val) if d: direction = d if bal is not None: prev_bal = bal out[credit_i] = fmt_money(amt_val) if (amt_val is not None and direction == "credit") else "" out[debit_i] = fmt_money(amt_val) if (amt_val is not None and direction == "debit") else "" out[balance_i] = f"{bal:,.2f}" if bal is not None else out[balance_i] body.append(out) if not body: return full html_rows = ["" + "".join(f"{html.escape(c)}" for c in new_hdr) + ""] for rr in body: html_rows.append("" + "".join(f"{html.escape(c)}" for c in rr) + "") return "\n" + "\n".join(html_rows) + "\n
" return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def _extract_daily_balance_by_date(md: str): """ Build a date->balance map from daily-balance style tables. Supports compact statements with repeated Date/Balance column pairs. """ out = {} for tm in re.finditer(r"]*>.*?", md, flags=re.IGNORECASE | re.DOTALL): t = tm.group(0) rows = _extract_rows_plain_from_table(t) if len(rows) < 2: continue hdr = [c.strip().lower() for c in rows[0]] if not hdr: continue date_cols = [i for i, c in enumerate(hdr) if "date" in c] bal_cols = [i for i, c in enumerate(hdr) if "balance" in c] if not date_cols or not bal_cols: continue pairs = [] for di in date_cols: bi = next((b for b in bal_cols if b > di), None) if bi is not None: pairs.append((di, bi)) if not pairs: continue for r in rows[1:]: for di, bi in pairs: if di >= len(r) or bi >= len(r): continue d = (r[di] or "").strip() if not re.search(r"\b\d{1,2}/\d{1,2}(?:/\d{2,4})?\b", d): continue m = re.search(r"\d{1,2}/\d{1,2}(?:/\d{2,4})?", d) if not m: continue key = m.group(0) bal = _parse_amount_or_none(r[bi]) if bal is None: continue out[key] = bal short = "/".join(key.split("/")[:2]) out[short] = bal return out def enrich_transaction_tables_with_daily_balances(md: str) -> str: """ If transaction tables have Date+Amount but no Balance column, append a Balance column using date-matched daily ending balances from the same document. """ if not md or " str: table = m.group(0) if re.search(r"(?:colspan|rowspan)\s*=", table, flags=re.IGNORECASE): return table tr_blocks = list(re.finditer(r"]*>.*?", table, flags=re.IGNORECASE | re.DOTALL)) if len(tr_blocks) < 2: return table first = tr_blocks[0].group(0) h_inner_m = re.search(r"]*>(.*)", first, flags=re.IGNORECASE | re.DOTALL) if not h_inner_m: return table h_inner = h_inner_m.group(1) h_cells = list(_CELL.finditer(h_inner)) if not h_cells: return table hdr_txt = [_cell_plain_text(c.group(0)).strip().lower() for c in h_cells] if any("balance" in c for c in hdr_txt): return table if not any("date" in c for c in hdr_txt): return table if not any(("amount" in c) or ("debit" in c) or ("credit" in c) for c in hdr_txt): return table date_idx = next((i for i, c in enumerate(hdr_txt) if "date" in c), -1) if date_idx < 0: return table new_parts: List[str] = [] cursor = 0 added = 0 for i, trm in enumerate(tr_blocks): new_parts.append(table[cursor : trm.start()]) tr_seg = trm.group(0) inner_m = re.search(r"]*>(.*)", tr_seg, flags=re.IGNORECASE | re.DOTALL) if not inner_m: new_parts.append(tr_seg) cursor = trm.end() continue tr_open_m = re.search(r"]*>", tr_seg, flags=re.IGNORECASE) tr_open = tr_open_m.group(0) if tr_open_m else "" tr_close = "" inner = inner_m.group(1) cells = [c.group(0) for c in _CELL.finditer(inner)] if i == 0: new_inner = inner + "Balance" new_parts.append(tr_open + new_inner + tr_close) else: if date_idx >= len(cells): new_parts.append(tr_seg) cursor = trm.end() continue d_txt = _cell_plain_text(cells[date_idx]) dm = re.search(r"\d{1,2}/\d{1,2}(?:/\d{2,4})?", d_txt or "") bal = dmap.get(dm.group(0)) if dm else None if bal is None and dm: bal = dmap.get("/".join(dm.group(0).split("/")[:2])) if bal is None: new_parts.append(tr_seg) cursor = trm.end() continue new_inner = inner + f"{bal:,.2f}" new_parts.append(tr_open + new_inner + tr_close) added += 1 cursor = trm.end() new_parts.append(table[cursor:]) if added < 3: return table return "".join(new_parts) return re.sub(r"]*>.*?", repl, md, flags=re.IGNORECASE | re.DOTALL) def _parse_statement_edge_balances(md: str): start = None end = None m1 = re.search( r"(?:Beginning|Starting)\s+balance[^$\n]{0,80}\$?\s*(-?\d{1,3}(?:,\d{3})*(?:\.\d{2})?)", md, flags=re.IGNORECASE, ) if m1: start = _parse_amount_or_none(m1.group(1)) m2 = re.search( r"Ending\s+balance[^$\n]{0,80}\$?\s*(-?\d{1,3}(?:,\d{3})*(?:\.\d{2})?)", md, flags=re.IGNORECASE, ) if m2: end = _parse_amount_or_none(m2.group(1)) return start, end def synthesize_running_balance_columns(md: str) -> str: """ Fallback when no explicit balance column exists: infer running balances from statement beginning/ending balance and signed transaction amounts. """ if not md or "]*>.*?", re.IGNORECASE | re.DOTALL) tr_re = re.compile(r"]*>.*?", re.IGNORECASE | re.DOTALL) tables = list(table_re.finditer(md)) row_refs = [] amounts = [] headers = {} for ti, tm in enumerate(tables): t = tm.group(0) if re.search(r"(?:colspan|rowspan)\s*=", t, flags=re.IGNORECASE): continue trs = list(tr_re.finditer(t)) if len(trs) < 2: continue h_inner_m = re.search(r"]*>(.*)", trs[0].group(0), flags=re.IGNORECASE | re.DOTALL) if not h_inner_m: continue h_cells = [c.group(0) for c in _CELL.finditer(h_inner_m.group(1))] if not h_cells: continue htxt = [_cell_plain_text(c).strip().lower() for c in h_cells] if any("balance" in h for h in htxt): continue date_idx = next((i for i, h in enumerate(htxt) if "date" in h), -1) amt_idx = next((i for i, h in enumerate(htxt) if "amount" in h or "debit" in h or "credit" in h), -1) if date_idx < 0 or amt_idx < 0: continue headers[ti] = (date_idx, amt_idx) for ri, trm in enumerate(trs[1:], 1): inner_m = re.search(r"]*>(.*)", trm.group(0), flags=re.IGNORECASE | re.DOTALL) if not inner_m: continue cells = [c.group(0) for c in _CELL.finditer(inner_m.group(1))] if len(cells) <= max(date_idx, amt_idx): continue d_txt = _cell_plain_text(cells[date_idx]) if not re.search(r"\d{1,2}/\d{1,2}(?:/\d{2,4})?", d_txt or ""): continue amt = _parse_amount_or_none(_cell_plain_text(cells[amt_idx])) if amt is None: continue row_refs.append((ti, ri)) amounts.append(amt) if len(amounts) < 20: return md if start_bal is None: start_bal = (end_bal or 0.0) - sum(amounts) running = [] cur = float(start_bal) for a in amounts: cur += float(a) running.append(cur) bal_by_ref = {row_refs[i]: running[i] for i in range(len(row_refs))} pieces = [] last = 0 for ti, tm in enumerate(tables): pieces.append(md[last : tm.start()]) t = tm.group(0) if ti not in headers: pieces.append(t) last = tm.end() continue date_idx, amt_idx = headers[ti] trs = list(tr_re.finditer(t)) out_t = [] cur2 = 0 for idx, trm in enumerate(trs): out_t.append(t[cur2 : trm.start()]) tr_seg = trm.group(0) inner_m = re.search(r"]*>(.*)", tr_seg, flags=re.IGNORECASE | re.DOTALL) tr_open_m = re.search(r"]*>", tr_seg, flags=re.IGNORECASE) tr_open = tr_open_m.group(0) if tr_open_m else "" if not inner_m: out_t.append(tr_seg) cur2 = trm.end() continue inner = inner_m.group(1) if idx == 0: out_t.append(tr_open + inner + "Balance") else: b = bal_by_ref.get((ti, idx)) if b is None: out_t.append(tr_seg) else: out_t.append(tr_open + inner + f"{b:,.2f}") cur2 = trm.end() out_t.append(t[cur2:]) pieces.append("".join(out_t)) last = tm.end() pieces.append(md[last:]) return "".join(pieces) def _extract_txn_date_balance_pairs(md: str): pairs = [] for tm in re.finditer(r"]*>.*?", md, flags=re.IGNORECASE | re.DOTALL): t = tm.group(0) rows = _extract_rows_plain_from_table(t) if len(rows) < 2: continue hdr = [c.strip().lower() for c in rows[0]] if not hdr: continue d_i = next((i for i, h in enumerate(hdr) if "date" in h), -1) b_i = next((i for i, h in enumerate(hdr) if "balance" in h), -1) a_i = next((i for i, h in enumerate(hdr) if "amount" in h or "debit" in h or "credit" in h), -1) if d_i < 0 or b_i < 0 or a_i < 0: continue for r in rows[1:]: if len(r) <= max(d_i, b_i): continue dm = re.search(r"\d{1,2}/\d{1,2}(?:/\d{2,4})?", (r[d_i] or "")) if not dm: continue bal = _parse_amount_or_none(r[b_i]) if bal is None: continue pairs.append((dm.group(0), bal)) return pairs def augment_with_derived_balance_views(md: str) -> str: """ Add a compact derived daily balance table and inferred beginning/ending summary when the document lacks a robust explicit daily balance section. """ if not md: return md # If there is already a clear daily-balance section, keep output unchanged. if re.search(r"daily\s+(?:ledger\s+)?balances", md, flags=re.IGNORECASE): return md pairs = _extract_txn_date_balance_pairs(md) if len(pairs) < 12: return md by_day = {} for d, b in pairs: k = "/".join(d.split("/")[:2]) by_day[k] = b if len(by_day) < 8: return md days = sorted( by_day.keys(), key=lambda x: (int(x.split("/")[0]), int(x.split("/")[1])), ) begin = by_day[days[0]] end = by_day[days[-1]] has_begin = bool(re.search(r"(?:beginning|starting)\s+balance", md, flags=re.IGNORECASE)) has_end = bool(re.search(r"ending\s+balance", md, flags=re.IGNORECASE)) rows = ["DateBalance"] for d in days: rows.append(f"{html.escape(d)}{by_day[d]:,.2f}") daily_tbl = "\n" + "\n".join(rows) + "\n
" extra = ["", "
Derived daily balances
", daily_tbl] if not has_begin or not has_end: extra.insert( 0, ( f"Inferred beginning balance: {begin:,.2f}\n" f"Inferred ending balance: {end:,.2f}" ), ) return md.rstrip() + "\n\n" + "\n\n".join(extra) + "\n" def prepend_balance_snapshot(md: str) -> str: """ Add a compact, explicit balance snapshot near the top so downstream metadata extraction consistently captures beginning/ending balances. """ if not md: return md if "balance snapshot" in md.lower(): return md start_bal, end_bal = _parse_statement_edge_balances(md) if start_bal is None and end_bal is None: # Try account-summary table fallback for tm in re.finditer(r"]*>.*?", md, flags=re.IGNORECASE | re.DOTALL): rows = _extract_rows_plain_from_table(tm.group(0)) for r in rows: if len(r) < 2: continue key = (r[0] or "").strip().lower() val = _parse_amount_or_none(r[1] if len(r) > 1 else "") if val is None: continue if start_bal is None and ("beginning balance" in key or "starting balance" in key): start_bal = val if end_bal is None and "ending balance" in key: end_bal = val if start_bal is not None or end_bal is not None: break if start_bal is None and end_bal is None: return md lines = ["## Balance Snapshot"] if start_bal is not None: lines.append(f"Beginning balance: {start_bal:,.2f}") if end_bal is not None: lines.append(f"Ending balance: {end_bal:,.2f}") header = "\n".join(lines) return header + "\n\n" + md def render_pdf_pages_to_images(pdf_path: str) -> Tuple[List[str], List[int]]: import pymupdf as fitz from PIL import Image doc = fitz.open(pdf_path) page_images: List[str] = [] page_heights: List[int] = [] for i in range(len(doc)): page = doc[i] pix = page.get_pixmap(matrix=fitz.Matrix(RENDER_SCALE, RENDER_SCALE), alpha=False) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) img = _enhance_raster_for_ocr(img) w, h = img.size pad_l = int(w * PAD_LEFT_FRAC) pad_r = int(w * PAD_RIGHT_FRAC) pad_t = int(h * PAD_TOP_FRAC) pad_b = int(h * PAD_BOTTOM_FRAC) if any(p > 0 for p in (pad_l, pad_r, pad_t, pad_b)): canvas = Image.new("RGB", (w + pad_l + pad_r, h + pad_t + pad_b), (255, 255, 255)) canvas.paste(img, (pad_l, pad_t)) img = canvas # Deterministic guardrail: downscale oversized pages before upload. iw, ih = img.size long_side = max(iw, ih) total_px = iw * ih if long_side > MAX_PAGE_LONG_SIDE or total_px > MAX_PAGE_TOTAL_PIXELS: scale_long = MAX_PAGE_LONG_SIDE / float(long_side) scale_area = (MAX_PAGE_TOTAL_PIXELS / float(total_px)) ** 0.5 scale = min(scale_long, scale_area, 1.0) nw = max(900, int(iw * scale)) nh = max(900, int(ih * scale)) img = img.resize((nw, nh), Image.Resampling.LANCZOS) base_path = os.path.join(tempfile.gettempdir(), f"glmocr_page_{os.getpid()}_{i}") img_path = f"{base_path}.png" img.save(img_path, "PNG", compress_level=PAGE_PNG_COMPRESS_LEVEL) try: size_bytes = os.path.getsize(img_path) except OSError: size_bytes = 0 if size_bytes > MAX_PAGE_UPLOAD_BYTES: jpg_path = f"{base_path}.jpg" quality = PAGE_JPEG_QUALITY work = img for _ in range(5): work.save(jpg_path, "JPEG", quality=quality, optimize=True) try: jsize = os.path.getsize(jpg_path) except OSError: jsize = 0 if 0 < jsize <= MAX_PAGE_UPLOAD_BYTES: img_path = jpg_path break quality = max(68, quality - 6) nw = max(900, int(work.width * 0.93)) nh = max(900, int(work.height * 0.93)) work = work.resize((nw, nh), Image.Resampling.LANCZOS) else: img_path = jpg_path page_images.append(img_path) page_heights.append(img.height) doc.close() return page_images, page_heights def get_page_md_and_regions(page_result): md = "" if hasattr(page_result, "markdown_result") and page_result.markdown_result: md = (page_result.markdown_result or "").strip() regions = [] if hasattr(page_result, "json_result"): jr = page_result.json_result if isinstance(jr, dict) and "regions" in jr: regions = jr.get("regions") or [] elif isinstance(jr, list) and len(jr) > 0: r = jr[0] if isinstance(jr[0], list) else jr if isinstance(r, list): regions = r elif isinstance(r, dict) and "regions" in r: regions = r.get("regions") or [] return md, regions def run_ocr(uploaded_file): if uploaded_file is None: return "Please upload a file." page_images: List[str] = [] try: path = uploaded_file.name if hasattr(uploaded_file, "name") else str(uploaded_file) is_pdf = path.lower().endswith(".pdf") parser = get_parser() page_heights: List[int] = [] if is_pdf: page_images, page_heights = render_pdf_pages_to_images(path) results = parser.parse(page_images) else: page_images = [path] page_heights = [1000] results = parser.parse(path) if not isinstance(results, list): results = [results] all_pages = [] for page_num, page_result in enumerate(results): page_md, regions = get_page_md_and_regions(page_result) img_h = page_heights[page_num] if page_num < len(page_heights) else 1000 header_end_frac, footer_start_frac = get_header_footer_zones(regions, img_h) he = header_end_frac if header_end_frac is not None else DEFAULT_ZONE_FRAC fs = footer_start_frac if footer_start_frac is not None else (1.0 - DEFAULT_ZONE_FRAC) he = max(0.02, min(0.25, he)) fs = max(0.75, min(0.98, fs)) parts = [] hdr = "" if is_pdf: hdr = extract_zone_text_pdf(path, page_num, 0, he) if not (hdr and hdr.strip()): hdr = extract_pdf_text_in_band(path, page_num, 0, PDF_HEADER_BAND_FRAC) if not (hdr and hdr.strip()) and page_num < len(page_images): hdr = ocr_zone(page_images[page_num], 0, he) if hdr and hdr.strip(): parts.append(light_stabilize_markdown(fix_account_number(normalize_money_glyphs(hdr.strip())))) if page_md and page_md.strip(): parts.append(light_stabilize_markdown(page_md.strip())) if ENABLE_FOOTER_OCR and page_num < len(page_images): ftr = "" if is_pdf: ftr = extract_zone_text_pdf(path, page_num, fs, 1.0) if not (ftr and ftr.strip()): ftr = extract_pdf_text_in_band(path, page_num, PDF_FOOTER_BAND_FRAC, 1.0) if not (ftr and ftr.strip()): ftr = ocr_zone(page_images[page_num], fs, 1.0) if ftr and ftr.strip(): ftr_clean = normalize_money_glyphs(ftr.strip()) ftr_first_line = next( (ln.strip().lower() for ln in ftr_clean.splitlines() if ln.strip()), "", ) already_present = ftr_first_line and any( ftr_first_line in part.lower() for part in parts ) _footer_date_re = re.compile(r"\b\d{1,2}[-/]\d{2}\b") _footer_amt_re = re.compile(r"\b\d{1,3}(?:,\d{3})*\.\d{2}\b") _date_hits = len(_footer_date_re.findall(ftr_clean)) _amt_hits = len(_footer_amt_re.findall(ftr_clean)) is_txn_dump = _date_hits >= 3 and _amt_hits >= 3 if not already_present and not is_txn_dump: parts.append(ftr_clean) if parts: all_pages.append("\n\n".join(parts)) merged = "\n\n---page-separator---\n\n".join(all_pages) if all_pages else "(No content)" if merged and merged != "(No content)" and not merged.lstrip().startswith("Error:"): merged = strip_malformed_leading_table_rows(merged) merged = strip_subtotal_rows_from_transaction_tables(merged) merged = mask_non_monetary_long_numbers(merged) merged = normalize_ambiguous_amount_balance_tables(merged) merged = normalize_amount_only_transaction_tables(merged) merged = infer_credit_debit_from_balance_deltas(merged) merged = sanitize_transaction_tables(merged) merged = stabilize_table_markup(merged, rounds=4) merged = enrich_transaction_tables_with_daily_balances(merged) # Keep transaction amounts untouched; add explicit snapshot for metadata extraction. merged = prepend_balance_snapshot(merged) merged = stabilize_table_markup(merged, rounds=2) merged = repair_thead_cell_semantics(merged) merged = strip_degenerate_html_tables(merged) return merged except Exception as e: import traceback log.exception("run_ocr failed: %s", e) return f"Error: {e}\n\n{traceback.format_exc()}" finally: for p in page_images: try: if ( isinstance(p, str) and "glmocr_page_" in os.path.basename(p) and (p.endswith(".png") or p.endswith(".jpg") or p.endswith(".jpeg")) ): os.unlink(p) except Exception: pass def _create_gradio_demo(): import gradio as gr with gr.Blocks(title="GLM-OCR (simple)") as demo: gr.Markdown( "# GLM-OCR (simple)\n" "Upload a PDF or image. Header and footer bands are included; " "body OCR is passed through with only light markdown cleanup." ) file_in = gr.File( label="Upload PDF or image", file_types=[".pdf", ".png", ".jpg", ".jpeg", ".tiff", ".bmp"], ) run_btn = gr.Button("Run OCR", variant="primary") out = gr.Textbox(lines=40, label="Output (markdown / light HTML)") run_btn.click(fn=run_ocr, inputs=file_in, outputs=out) return demo if __name__ == "__main__": _create_gradio_demo().launch()