Spaces:
Sleeping
Sleeping
| """ | |
| GLM-OCR Hugging Face Space app for PDF/image OCR with header inclusion | |
| and table-structure stabilization for downstream bank-statement pipelines. | |
| Hard-coded knobs (no environment variables required). | |
| Primary goals for reconcile rate: | |
| - preserve right-most columns (often "Balance") by higher DPI render + right padding | |
| - keep tables as tables (convert markdown pipe tables -> HTML table) | |
| - return ---page-separator--- between pages | |
| - normalize HTML tables generically so downstream parsing/classification is stable: | |
| 1) expand colspan/rowspan into a rectangular grid | |
| 2) drop truly-empty columns (common in summary tables) | |
| 3) merge "blank header" columns that contain text into the left column (common when DESCRIPTION is split) | |
| 4) clean common header artifacts (e.g. "DESCRIPTIONBeginning Balance", "BALANCE$3,447.10") | |
| 5) recover fused first data row generically (two-signal guard: text fusion + money fusion must both fire) | |
| 6) promote misplaced header row: when real column headers land in a data row, restructure the grid | |
| 7) fix fused key-value rows in summary sections (e.g. Interest Summary) appended to transaction tables | |
| 8) reconstruct mid-table separator rows split across columns by OCR (e.g. 'Card account # XXXX 2 | | 889') | |
| - footer extraction enabled on all pages with deduplication guard (no double-print if body OCR already captured it) | |
| - Fix 4: cross-validate OCR table row counts against PDF text layer; inject missing duplicate rows | |
| (safe no-op for scanned PDFs and non-transaction pages) | |
| """ | |
| # Patch asyncio first (before Gradio imports it) to suppress Python 3.13 cleanup 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 logging | |
| import os | |
| import re | |
| import html | |
| import tempfile | |
| from typing import List, Tuple | |
| from collections import defaultdict | |
| from html.parser import HTMLParser | |
| import yaml | |
| import gradio as gr | |
| import glmocr | |
| log = logging.getLogger("glmocr_app") | |
| logging.basicConfig(level=logging.INFO) | |
| GLMOCR_BASE = os.path.dirname(glmocr.__file__) | |
| CONFIG_PATH = os.path.join(GLMOCR_BASE, "config.yaml") | |
| FORMATTER_PATH = os.path.join(GLMOCR_BASE, "postprocess", "result_formatter.py") | |
| # ============================================================ | |
| # HARD-CODED SETTINGS (edit these numbers to tune quality/speed) | |
| # ============================================================ | |
| GLMOCR_API_KEY = "1960cf47d08547d2b4d03544143cce98.AAxia6FudQwLdUDb" | |
| # Higher = better OCR for small/right-aligned digits; slower | |
| RENDER_SCALE = 2.2 # try 2.5 if right-side numbers are missed | |
| # Padding to protect columns near edges (Balance is often right-most) | |
| PAD_LEFT_FRAC = 0.02 | |
| PAD_RIGHT_FRAC = 0.06 # try 0.10 if right-most balances are missing | |
| PAD_TOP_FRAC = 0.01 | |
| PAD_BOTTOM_FRAC = 0.01 | |
| ENABLE_CONTRAST = True | |
| DEFAULT_ZONE_FRAC = 0.12 | |
| PDF_HEADER_BAND_FRAC = 0.10 | |
| ENABLE_FOOTER_OCR = True # enabled — 3-step fallback with dedup guard | |
| PDF_FOOTER_BAND_FRAC = 0.88 | |
| MIN_CROP_HEIGHT = 112 | |
| MIN_CROP_PIXELS = 112 * 112 | |
| # ============================================================ | |
| _parser = None | |
| def get_parser(): | |
| global _parser | |
| if _parser is None: | |
| from glmocr import GlmOcr | |
| _parser = GlmOcr(api_key=GLMOCR_API_KEY, mode="maas") | |
| return _parser | |
| # --------------------------------------------------------------------------- | |
| # Best-effort config tweaks (safe to fail on read-only HF env) | |
| # --------------------------------------------------------------------------- | |
| try: | |
| with open(CONFIG_PATH, "r") as f: | |
| config = yaml.safe_load(f) | |
| config["pipeline"]["maas"]["enabled"] = True | |
| config["pipeline"]["maas"]["api_key"] = GLMOCR_API_KEY | |
| with open(CONFIG_PATH, "w") as f: | |
| yaml.dump(config, f, default_flow_style=False, sort_keys=False) | |
| except Exception: | |
| pass | |
| # Best-effort formatter tweak: avoid stripping header/footer labels | |
| try: | |
| with open(FORMATTER_PATH, "r") as f: | |
| source = f.read() | |
| for label in ( | |
| '"header"', "'header'", | |
| '"footer"', "'footer'", | |
| '"doc_header"', "'doc_header'", | |
| '"doc_footer"', "'doc_footer'", | |
| ): | |
| source = re.sub(r",\s*" + re.escape(label), "", source) | |
| source = re.sub(re.escape(label) + r"\s*,", "", source) | |
| source = re.sub(re.escape(label), "", source) | |
| with open(FORMATTER_PATH, "w") as f: | |
| f.write(source) | |
| except Exception: | |
| pass | |
| # -------------------------- | |
| # Header/footer helpers | |
| # -------------------------- | |
| 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=92) | |
| 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 | |
| # -------------------------- | |
| # Table stabilization helpers | |
| # -------------------------- | |
| 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"</(table|tbody|thead|tr|td|th)>", 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 += ("</%s>" % tag) * (opened - closed) | |
| return md | |
| 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("<tr>" + "".join(f"<th>{html.escape(c)}</th>" for c in header) + "</tr>") | |
| for ln in body_lines: | |
| cols = split_row(ln) | |
| if len(cols) < len(header): | |
| cols += [""] * (len(header) - len(cols)) | |
| html_rows.append("<tr>" + "".join(f"<td>{html.escape(c)}</td>" for c in cols[: len(header)]) + "</tr>") | |
| return "<table>\n" + "\n".join(html_rows) + "\n</table>" | |
| 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 | |
| # ---- Generic HTML table normalizer ---- | |
| class TableGridParser(HTMLParser): | |
| """Parse a <table> into rows of (text, colspan, rowspan).""" | |
| def __init__(self): | |
| super().__init__() | |
| self.rows = [] | |
| self._current_row = [] | |
| self._cell_text = [] | |
| self._colspan = 1 | |
| self._rowspan = 1 | |
| self._in_cell = False | |
| def handle_starttag(self, tag, attrs): | |
| if tag == "tr": | |
| self._current_row = [] | |
| elif tag in ("td", "th"): | |
| attrs_d = dict(attrs) | |
| self._colspan = max(1, int(attrs_d.get("colspan", 1))) | |
| self._rowspan = max(1, int(attrs_d.get("rowspan", 1))) | |
| self._cell_text = [] | |
| self._in_cell = True | |
| def handle_endtag(self, tag): | |
| if tag in ("td", "th"): | |
| text = "".join(self._cell_text).strip().replace("\n", " ") | |
| self._current_row.append((text, self._colspan, self._rowspan)) | |
| self._in_cell = False | |
| elif tag == "tr": | |
| self.rows.append(self._current_row) | |
| def handle_data(self, data): | |
| if self._in_cell: | |
| self._cell_text.append(data) | |
| def _build_grid(rows_data): | |
| if not rows_data: | |
| return [] | |
| blocked = defaultdict(set) | |
| grid = [] | |
| for r, row_cells in enumerate(rows_data): | |
| grid.append([]) | |
| col = 0 | |
| for content, C, R in row_cells: | |
| while col in blocked[r]: | |
| grid[r].append("") | |
| col += 1 | |
| for k in range(C): | |
| grid[r].append(content if k == 0 else "") | |
| for k in range(1, R): | |
| blocked[r + k].add(col) | |
| col += C | |
| max_cols = max(len(row) for row in grid) if grid else 0 | |
| for row in grid: | |
| while len(row) < max_cols: | |
| row.append("") | |
| return grid | |
| def _grid_to_html(grid): | |
| if not grid: | |
| return "" | |
| lines = ["<table>"] | |
| for r, row in enumerate(grid): | |
| lines.append("<tr>") | |
| tag = "th" if r == 0 else "td" | |
| for cell in row: | |
| escaped = ( | |
| (cell or "") | |
| .replace("&", "&") | |
| .replace("<", "<") | |
| .replace(">", ">") | |
| .replace('"', """) | |
| ) | |
| lines.append(f"<{tag}>{escaped}</{tag}>") | |
| lines.append("</tr>") | |
| lines.append("</table>") | |
| return "\n".join(lines) | |
| def _is_amount_like(s: str) -> bool: | |
| if not s: | |
| return False | |
| return re.fullmatch(r"\$?-?\d{1,3}(?:,\d{3})*(?:\.\d{2})?", s.strip()) is not None | |
| def _drop_truly_empty_columns(grid): | |
| """ | |
| Drop columns that are empty in header AND almost always empty in body. | |
| This fixes summary tables that have an extra blank trailing column. | |
| """ | |
| if not grid or len(grid) < 1: | |
| return grid | |
| header = [str(c or "").strip() for c in grid[0]] | |
| ncols = len(header) | |
| if ncols <= 1: | |
| return grid | |
| body = grid[1:] | |
| keep = [True] * ncols | |
| for i in range(ncols): | |
| if header[i] != "": | |
| continue | |
| total = 0 | |
| non_empty = 0 | |
| for row in body: | |
| if i >= len(row): | |
| continue | |
| total += 1 | |
| if str(row[i] or "").strip(): | |
| non_empty += 1 | |
| if total > 0 and (non_empty / total) <= 0.05: | |
| keep[i] = False | |
| if all(keep): | |
| return grid | |
| new_grid = [] | |
| for row in grid: | |
| new_grid.append([cell for idx, cell in enumerate(row) if idx < len(keep) and keep[idx]]) | |
| return new_grid | |
| def _merge_blank_header_text_columns(grid): | |
| """ | |
| If the header row has blank columns, and most body rows have non-empty *text* in that | |
| blank column, merge that column into the nearest non-empty header to the left (usually | |
| DESCRIPTION), then remove the blank column. | |
| This fixes transaction tables where DESCRIPTION is split across two columns. | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| header = [str(c or "").strip() for c in grid[0]] | |
| ncols = len(header) | |
| body = grid[1:] | |
| blank_cols = [i for i, h in enumerate(header) if h == ""] | |
| if not blank_cols: | |
| return grid | |
| keep = [True] * ncols | |
| for i in blank_cols: | |
| j = i - 1 | |
| while j >= 0 and header[j] == "": | |
| j -= 1 | |
| if j < 0: | |
| continue | |
| total = 0 | |
| non_empty = 0 | |
| texty = 0 | |
| for row in body: | |
| if i >= len(row) or j >= len(row): | |
| continue | |
| v = str(row[i] or "").strip() | |
| total += 1 | |
| if v: | |
| non_empty += 1 | |
| if not _is_amount_like(v): | |
| texty += 1 | |
| if total == 0: | |
| continue | |
| non_empty_ratio = non_empty / total | |
| texty_ratio = (texty / non_empty) if non_empty else 0.0 | |
| if non_empty_ratio >= 0.55 and texty_ratio >= 0.70: | |
| for r in range(1, len(grid)): | |
| row = grid[r] | |
| if i >= len(row) or j >= len(row): | |
| continue | |
| left = str(row[j] or "").strip() | |
| right = str(row[i] or "").strip() | |
| if right: | |
| row[j] = (left + " " + right).strip() if left else right | |
| keep[i] = False | |
| if all(keep): | |
| return grid | |
| new_grid = [] | |
| for row in grid: | |
| new_grid.append([cell for idx, cell in enumerate(row) if idx < len(keep) and keep[idx]]) | |
| return new_grid | |
| # -------------------------- | |
| # Shared keyword definitions | |
| # -------------------------- | |
| # Matches a standalone money value across currencies. | |
| # Covers: $3,447.10 -1,234.56 £500.00 Rs1000 etc. | |
| _MONEY_RE = re.compile( | |
| r"^(?:[\$£€¥]|Rs\.?|INR|PKR)?\s*-?\s?\d{1,3}(?:,\d{3})*(?:\.\d{1,4})?$", | |
| re.IGNORECASE, | |
| ) | |
| # All column-header keywords recognised across any bank statement format. | |
| _HEADER_KW_PATTERNS = [ | |
| # Date variants | |
| r"DATE", | |
| r"POSTING\s+DATE", | |
| r"VALUE\s+DATE", | |
| r"TXN\s+DATE", | |
| r"TRANSACTION\s+DATE", | |
| r"ENTRY\s+DATE", | |
| r"EFFECTIVE\s+DATE", | |
| # Transaction ID / reference | |
| r"TRANSACTION(?:\s+(?:ID|TYPE|NO|NUMBER))?", | |
| r"TXN(?:\s+(?:ID|NO|TYPE))?", | |
| r"REF(?:ERENCE)?(?:\s*(?:NO|NUM|NUMBER))?", | |
| r"CHEQUE(?:\s*(?:NO|NUMBER))?", | |
| r"CHQ(?:\s*(?:NO|NUMBER))?", | |
| r"VOUCHER(?:\s*(?:NO|NUMBER))?", | |
| r"SR\.?\s*NO\.?", | |
| r"SERIAL(?:\s*(?:NO|NUMBER))?", | |
| # Description variants | |
| r"DESCRIPTION", | |
| r"DETAILS?", | |
| r"PARTICULARS?(?:\s+OF\s+TRANSACTION)?", | |
| r"NARRATION", | |
| r"REMARKS?", | |
| r"NOTES?", | |
| # Debit variants | |
| r"DEBIT", | |
| r"DEBITS?", | |
| r"DR\.?", | |
| r"WITHDRAWALS?", | |
| r"PAID\s+OUT", | |
| r"MONEY\s+OUT", | |
| # Credit variants | |
| r"CREDIT", | |
| r"CREDITS?", | |
| r"CR\.?", | |
| r"DEPOSITS?", | |
| r"PAID\s+IN", | |
| r"MONEY\s+IN", | |
| # Amount | |
| r"AMOUNT", | |
| # Balance variants | |
| r"BALANCE", | |
| r"BAL\.?", | |
| r"RUNNING\s+BALANCE", | |
| r"RUNNING\s+BAL\.?", | |
| r"AVAILABLE\s+BALANCE", | |
| r"AVAILABLE\s+BAL\.?", | |
| r"AVAIL\.?\s+BAL\.?", | |
| r"LEDGER\s+BALANCE", | |
| r"LEDGER\s+BAL\.?", | |
| r"CLOSING\s+BALANCE", | |
| r"CLOSING\s+BAL\.?", | |
| r"OPENING\s+BALANCE", | |
| r"OPENING\s+BAL\.?", | |
| ] | |
| # Pre-compiled: each pattern anchored at start, case-insensitive | |
| _HEADER_KW_RES = [ | |
| re.compile(r"^(" + p + r")(.*)", re.IGNORECASE | re.DOTALL) | |
| for p in _HEADER_KW_PATTERNS | |
| ] | |
| # Quick exact-match check: "is this entire string a known keyword?" | |
| _HEADER_KW_EXACT_RE = re.compile( | |
| r"^(?:" + r"|".join(_HEADER_KW_PATTERNS) + r")$", | |
| re.IGNORECASE, | |
| ) | |
| # Minimum fraction of cells in a row that must be pure keywords for that row | |
| # to be considered a misplaced header row. | |
| _HEADER_ROW_KEYWORD_THRESHOLD = 0.5 | |
| # -------------------------- | |
| # Fix 1: Fused-header recovery | |
| # -------------------------- | |
| def _split_keyword_remainder(cell_text: str): | |
| """ | |
| If cell_text starts with a known header keyword followed by extra content, | |
| return (keyword, remainder). Otherwise return (cell_text, ""). | |
| """ | |
| s = cell_text.strip() | |
| for pattern in _HEADER_KW_RES: | |
| m = pattern.match(s) | |
| if m: | |
| keyword = m.group(1).strip() | |
| remainder = m.group(2).strip() | |
| return keyword, remainder | |
| return s, "" | |
| def _extract_fused_header_artifacts(header_row): | |
| """ | |
| Generic detector for the OCR artifact where the first data row of a table | |
| gets fused into the header cells during OCR. | |
| The artifact pattern (BOTH signals must fire simultaneously): | |
| - Signal 1 — text-fused: a header cell contains KEYWORD + free descriptive text | |
| e.g. "DESCRIPTIONBeginning Balance" "NARRATIONOpening Balance" | |
| - Signal 2 — money-fused: a DIFFERENT header cell contains KEYWORD + money amount | |
| e.g. "BALANCE$3,447.10" "AMOUNT 5,000.00" | |
| Two-signal guard prevents false positives on clean PDFs from any bank. | |
| Returns: | |
| (cleaned_header : list[str], recovered_row : list[str] | None) | |
| """ | |
| if not header_row: | |
| return list(header_row), None | |
| ncols = len(header_row) | |
| cleaned = list(header_row) | |
| recovered = [""] * ncols | |
| text_fused = [] | |
| money_fused = [] | |
| for idx, cell in enumerate(header_row): | |
| cell_s = str(cell or "").strip() | |
| if not cell_s: | |
| continue | |
| keyword, remainder = _split_keyword_remainder(cell_s) | |
| if not remainder: | |
| continue | |
| remainder_no_space = remainder.replace(" ", "") | |
| if _MONEY_RE.match(remainder_no_space): | |
| cleaned[idx] = keyword | |
| recovered[idx] = remainder | |
| money_fused.append(idx) | |
| elif not _HEADER_KW_EXACT_RE.match(remainder.split()[0] if remainder.split() else ""): | |
| cleaned[idx] = keyword | |
| recovered[idx] = remainder | |
| text_fused.append(idx) | |
| if text_fused and money_fused: | |
| return cleaned, recovered | |
| return list(header_row), None | |
| def _clean_header_artifacts(grid): | |
| """ | |
| Entry point for Fix 1 called from normalize_html_tables. | |
| Returns (grid, recovered_row | None). | |
| """ | |
| if not grid or not grid[0]: | |
| return grid, None | |
| cleaned_header, recovered_row = _extract_fused_header_artifacts(grid[0]) | |
| grid[0] = cleaned_header | |
| return grid, recovered_row | |
| # -------------------------- | |
| # Fix 2: Misplaced header row promotion | |
| # -------------------------- | |
| def _row_keyword_score(row): | |
| """ | |
| Return the fraction of non-empty cells in this row that are pure header | |
| keywords (e.g. DATE, DESCRIPTION, DEBIT, CREDIT, BALANCE). | |
| Also handles cells like "DATE DESCRIPTION" where two keywords are | |
| space-joined into one cell — these count as a keyword cell too. | |
| """ | |
| non_empty = [str(c or "").strip() for c in row if str(c or "").strip()] | |
| if not non_empty: | |
| return 0.0 | |
| keyword_hits = 0 | |
| for cell in non_empty: | |
| if _HEADER_KW_EXACT_RE.match(cell): | |
| keyword_hits += 1 | |
| continue | |
| parts = cell.split() | |
| if all(_HEADER_KW_EXACT_RE.match(p) for p in parts) and len(parts) > 1: | |
| keyword_hits += 1 | |
| continue | |
| return keyword_hits / len(non_empty) | |
| # Pattern: header cell contains balance/account info — do NOT promote away from it. | |
| # Matches things like "Beginning Balance:", "Ending Balance:", account numbers, "$20.48$3.46" | |
| _BALANCE_INFO_RE = re.compile( | |
| r"(beginning|ending|opening|closing)\s+balance" | |
| r"|account\s*#?\s*\d{5,}" | |
| r"|\b\d{7,}\b" | |
| r"|\$\d{1,3}(?:,\d{3})*\.\d{2}", | |
| re.IGNORECASE, | |
| ) | |
| def _header_contains_balance_info(header_row): | |
| """ | |
| Return True if any cell in the header row contains balance or account | |
| information — indicating this IS a legitimate header row even if its | |
| keyword score is low (e.g. Citi's "208479667 | Beginning Balance:$20.48 | ..."). | |
| """ | |
| for cell in header_row: | |
| if _BALANCE_INFO_RE.search(str(cell or "")): | |
| return True | |
| return False | |
| def _promote_misplaced_header_row(grid): | |
| """ | |
| Detects and fixes the OCR artifact where real column headers land in a | |
| <td> data row instead of the <th> header row. | |
| Two scenarios handled: | |
| Scenario A — Simple misplaced header (e.g. TD Bank): | |
| <th>ACCOUNT ACTIVITY</th> ← section title, not a real header | |
| <td>DATE DESCRIPTION</td> ... ← real column headers in a data row | |
| <td>01/05 ...</td> ← data | |
| → Promote the keyword row to header, discard the section title rows above. | |
| Scenario B — Metadata header + misplaced column headers (e.g. Citi page 1): | |
| <th>208479667</th> <th>Beginning Balance:$20.48 Ending Balance:$3.46</th> | |
| <td>Date Description</td> <td>Debits</td> <td>Credits</td> <td>Balance</td> | |
| <td>04/01 DEBIT CARD... 8.01</td> ... | |
| → The existing th row has balance/account metadata (NOT a real column header). | |
| → Promote the keyword data row to header. | |
| → Preserve metadata row cells as a plain-text prefix OUTSIDE the table | |
| by embedding them as a leading data row with colspan (kept for reference). | |
| → Split any "Date Description" fused cells in data rows. | |
| Guard conditions (no-op if not met — safe for all other PDFs): | |
| - Current header keyword score < threshold. | |
| - A data row within the first 5 rows has keyword score >= threshold. | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| current_header_score = _row_keyword_score(grid[0]) | |
| if current_header_score >= _HEADER_ROW_KEYWORD_THRESHOLD: | |
| return grid | |
| # Search the first few data rows for a candidate keyword header row. | |
| candidate_idx = None | |
| candidate_score = 0.0 | |
| for i in range(1, min(len(grid), 6)): | |
| score = _row_keyword_score(grid[i]) | |
| if score >= _HEADER_ROW_KEYWORD_THRESHOLD and score > candidate_score: | |
| candidate_score = score | |
| candidate_idx = i | |
| if candidate_idx is None: | |
| return grid | |
| candidate_row = grid[candidate_idx] | |
| # Expand any merged keyword cells (e.g. "DATE DESCRIPTION" → ["DATE", "DESCRIPTION"]) | |
| expanded_header = [] | |
| for cell in candidate_row: | |
| cell_s = str(cell or "").strip() | |
| parts = cell_s.split() | |
| if len(parts) > 1 and all(_HEADER_KW_EXACT_RE.match(p) for p in parts): | |
| expanded_header.extend(parts) | |
| else: | |
| expanded_header.append(cell_s) | |
| new_ncols = len(expanded_header) | |
| col_expansion = new_ncols - len(candidate_row) | |
| # If the existing header row contains balance/account metadata (Scenario B), | |
| # preserve it as the first data row so the information is not lost. | |
| # This row will have its content in col 0 (joined) and blanks elsewhere. | |
| metadata_row = None | |
| if _header_contains_balance_info(grid[0]): | |
| meta_cells = [str(c or "").strip() for c in grid[0]] | |
| meta_text = " ".join(c for c in meta_cells if c).strip() | |
| if meta_text: | |
| metadata_row = [meta_text] + [""] * (new_ncols - 1) | |
| # Re-align data rows below the candidate header | |
| new_data_rows = [] | |
| for row in grid[candidate_idx + 1:]: | |
| if col_expansion > 0 and row: | |
| first_cell = str(row[0] or "").strip() | |
| date_match = re.match(r"^(\d{1,2}/\d{1,2})\s+(.*)", first_cell, re.DOTALL) | |
| if date_match and col_expansion == 1: | |
| new_row = [date_match.group(1), date_match.group(2).strip()] + list(row[1:]) | |
| else: | |
| new_row = list(row) + [""] * col_expansion | |
| new_data_rows.append(new_row) | |
| else: | |
| new_data_rows.append(list(row)) | |
| def pad(row, n): | |
| r = list(row) | |
| while len(r) < n: | |
| r.append("") | |
| return r[:n] | |
| new_grid = [pad(expanded_header, new_ncols)] | |
| # Insert metadata row first if present (preserves Beginning/Ending Balance) | |
| if metadata_row is not None: | |
| new_grid.append(pad(metadata_row, new_ncols)) | |
| for row in new_data_rows: | |
| new_grid.append(pad(row, new_ncols)) | |
| return new_grid | |
| # -------------------------- | |
| # Fix 3: Fused key-value rows in summary sections | |
| # -------------------------- | |
| # Matches a value suffix fused onto a label with no space. | |
| # e.g. "Beginning Interest Rate0.00%" → label + "0.00%" | |
| # "Number of days in this Period31" → label + "31" | |
| # Strict: no \s* between groups so label trailing-space detects space-separated values | |
| _FUSED_KV_RE = re.compile( | |
| r"^(.+?)(-?\d{1,3}(?:,\d{3})*(?:\.\d+)?%?)$", | |
| re.DOTALL, | |
| ) | |
| def _fix_fused_keyvalue_rows(grid): | |
| """ | |
| Fix rows where label+value are fused into the first cell with all other | |
| cells empty — common in Interest Summary / Fee Summary sections appended | |
| to a transaction table by OCR. | |
| e.g. "Beginning Interest Rate0.00%" → label="Beginning Interest Rate" value="0.00%" | |
| "Number of days in this Period31" → label="..." value="31" | |
| Guard conditions (no-op if not met — safe for all other tables): | |
| - All cells except first must be empty. | |
| - A non-space character must immediately precede the digit run (fused). | |
| - Both label and value parts must be non-empty after splitting. | |
| - The cell must NOT contain an account number (7+ digit run) — prevents | |
| incorrectly splitting "STREAMLINED CHECKING #208479667" or | |
| "Charges debited from account #208479667". | |
| - The cell must NOT contain a proper dollar amount with decimal — prevents | |
| incorrectly splitting metadata like "Beginning Balance:$20.48$3.46". | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| # Matches a proper dollar/currency amount with decimal point — these appear | |
| # in legitimate label cells and must not trigger the fused-KV split. | |
| _DOLLAR_AMOUNT_RE = re.compile(r"\$\s*\d{1,3}(?:,\d{3})*\.\d{2}") | |
| # Matches a long account/reference number (7+ consecutive digits) | |
| _LONG_NUMBER_RE = re.compile(r"\d{7,}") | |
| ncols = len(grid[0]) | |
| new_grid = [grid[0]] | |
| for row in grid[1:]: | |
| cells = [str(c or "").strip() for c in row] | |
| if not cells: | |
| new_grid.append(row) | |
| continue | |
| first = cells[0] | |
| rest_empty = all(c == "" for c in cells[1:]) | |
| if not rest_empty or not first: | |
| new_grid.append(row) | |
| continue | |
| # Guard: skip rows containing a proper dollar amount with decimal | |
| # (these are metadata/balance rows, not fused key-value summary rows) | |
| if _DOLLAR_AMOUNT_RE.search(first): | |
| new_grid.append(row) | |
| continue | |
| # Guard: skip rows containing a long account/reference number (7+ digits) | |
| # (e.g. "STREAMLINED CHECKING #208479667", "account #208479667") | |
| if _LONG_NUMBER_RE.search(first): | |
| new_grid.append(row) | |
| continue | |
| # Guard: skip rows containing a # followed by digits or X-patterns | |
| # (card/account number labels like "Card account # XXXX XXXX XXXX 2889") | |
| # These are separator rows reconstructed by Fix5, not fused KV rows. | |
| if re.search(r"#\s*[\dX]", first): | |
| new_grid.append(row) | |
| continue | |
| m = _FUSED_KV_RE.match(first) | |
| if not m: | |
| new_grid.append(row) | |
| continue | |
| label_raw = m.group(1) # trailing space means value was space-separated, not fused | |
| value = m.group(2).strip() | |
| # Space before value means NOT fused — "Interest Rate 0.00%" is clean | |
| if label_raw != label_raw.rstrip(): | |
| new_grid.append(row) | |
| continue | |
| label = label_raw.strip() | |
| if not label or not value: | |
| new_grid.append(row) | |
| continue | |
| new_row = [""] * ncols | |
| new_row[0] = label | |
| new_row[ncols - 1] = value | |
| new_grid.append(new_row) | |
| return new_grid | |
| # -------------------------- | |
| # Fix 4: PDF text-layer row-count patch | |
| # -------------------------- | |
| def _extract_textlayer_rows(pdf_path: str, page_num: int): | |
| """ | |
| Extract structured transaction rows from the PDF text layer using pymupdf. | |
| Returns list of {"date": str, "desc": str, "amount": str} dicts, or [] | |
| if the PDF has no text layer, pymupdf is unavailable, or fewer than 2 | |
| transaction rows are found (guards against non-transaction pages). | |
| Supports multiple date formats: | |
| - Numeric: MM/DD, MM/DD/YY, MM/DD/YYYY (Chase, TD, BoA, Citi) | |
| - Month-name: Jan 16, Feb 7, Mar 05 (Capital One, Amex) | |
| - ISO: YYYY-MM-DD | |
| """ | |
| try: | |
| import pymupdf as fitz | |
| doc = fitz.open(pdf_path) | |
| page = doc[page_num] | |
| words = page.get_text("words") | |
| doc.close() | |
| if not words: | |
| return [] | |
| # Group words by y-bucket (5pt tolerance) | |
| lines_by_y = {} | |
| for w in words: | |
| x0, y0, word = float(w[0]), float(w[1]), w[4] | |
| bucket = None | |
| for existing_y in lines_by_y: | |
| if abs(existing_y - y0) <= 5: | |
| bucket = existing_y | |
| break | |
| if bucket is None: | |
| bucket = y0 | |
| lines_by_y.setdefault(bucket, []).append((x0, word)) | |
| sorted_lines = [] | |
| for y in sorted(lines_by_y): | |
| line_words = sorted(lines_by_y[y], key=lambda t: t[0]) | |
| sorted_lines.append([w for _, w in line_words]) | |
| # Date patterns — numeric (MM/DD, MM-DD variants) or month-name (Jan 16) | |
| _MONTHS = r"(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)" | |
| date_re = re.compile( | |
| r"^\d{1,2}/\d{2}(?:/\d{2,4})?$" # MM/DD, MM/DD/YY, MM/DD/YYYY | |
| r"|^\d{1,2}-\d{2}(?:-\d{2,4})?$" # MM-DD, MM-DD-YY (East West Bank) | |
| r"|^\d{4}-\d{2}-\d{2}$" # YYYY-MM-DD | |
| r"|^" + _MONTHS + r"$", # "Jan", "Feb" etc (month-name date part 1) | |
| re.IGNORECASE, | |
| ) | |
| # When a month-name date appears, the next token is the day number | |
| month_re = re.compile(r"^" + _MONTHS + r"$", re.IGNORECASE) | |
| day_re = re.compile(r"^\d{1,2}$") | |
| # Amount: -1,234.56 $193.63 -$240.46 - $500.00 (with space after -) | |
| amount_re = re.compile( | |
| r"^-?\$?\d{1,3}(?:,\d{3})*(?:\.\d{1,4})?$" | |
| r"|^\$-?\d{1,3}(?:,\d{3})*(?:\.\d{1,4})?$" | |
| r"|^-\s+\$\d{1,3}(?:,\d{3})*(?:\.\d{1,4})?$" # "- $500.00" | |
| ) | |
| rows = [] | |
| for line in sorted_lines: | |
| if len(line) < 2: | |
| continue | |
| # Detect date and find where description starts | |
| date_str = None | |
| desc_start = None | |
| if date_re.match(line[0]): | |
| if month_re.match(line[0]) and len(line) > 1 and day_re.match(line[1]): | |
| # Month-name date: "Jan 16" → two tokens | |
| date_str = line[0] + " " + line[1] | |
| desc_start = 2 | |
| else: | |
| # Single-token numeric date | |
| date_str = line[0] | |
| desc_start = 1 | |
| else: | |
| continue | |
| if desc_start is None or desc_start >= len(line): | |
| continue | |
| # For Capital One: Trans Date and Post Date are both present | |
| # Line looks like: Jan 16 Jan 16 CAPITAL ONE MOBILE PYMT - $500.00 | |
| # After consuming first date, check if next tokens are also a date | |
| remaining = line[desc_start:] | |
| # Capture post_date if next tokens are also a date | |
| post_date_str = "" | |
| if remaining and month_re.match(remaining[0]): | |
| if len(remaining) > 1 and day_re.match(remaining[1]): | |
| post_date_str = remaining[0] + " " + remaining[1] | |
| remaining = remaining[2:] | |
| elif len(remaining) > 0: | |
| remaining = remaining[1:] | |
| elif remaining and date_re.match(remaining[0]) and not month_re.match(remaining[0]): | |
| post_date_str = remaining[0] | |
| remaining = remaining[1:] | |
| if len(remaining) < 2: | |
| continue | |
| # Handle "- $500.00" split as two tokens at end | |
| if (len(remaining) >= 2 | |
| and remaining[-2] == "-" | |
| and remaining[-1].startswith("$")): | |
| amount_candidate = remaining[-2] + " " + remaining[-1] | |
| desc_tokens = remaining[:-2] | |
| else: | |
| amount_candidate = remaining[-1] | |
| desc_tokens = remaining[:-1] | |
| if not amount_re.match(amount_candidate): | |
| continue | |
| desc = " ".join(desc_tokens).strip() | |
| if not desc: | |
| continue | |
| # Guard: description must contain at least one letter. | |
| # Rows where description is purely digits/dates/amounts are | |
| # DAILY BALANCE rows (e.g. "170,198.04 01-13 45,442.31 01-24"), | |
| # not real transactions — skip them. | |
| if not re.search(r"[A-Za-z]", desc): | |
| continue | |
| rows.append({"date": date_str, "post_date": post_date_str, "desc": desc, "amount": amount_candidate}) | |
| # Guard: require at least 2 rows to avoid false positives on non-transaction pages | |
| return rows if len(rows) >= 2 else [] | |
| except Exception as e: | |
| log.debug("_extract_textlayer_rows failed (page %d): %s", page_num, e) | |
| return [] | |
| def _patch_ocr_with_textlayer(page_md: str, pdf_path: str, page_num: int) -> str: | |
| """ | |
| Compare OCR table rows against PDF text-layer rows and inject any rows | |
| the OCR model silently dropped (typically identical consecutive rows). | |
| Guard conditions (all must pass for any injection): | |
| 1. PDF text layer must exist and yield >= 2 transaction rows. | |
| 2. The OCR table must have DATE + DESCRIPTION + AMOUNT columns. | |
| 3. At least one (date, desc, amount) key must appear in both OCR and | |
| text layer — ensures we are patching the right table. | |
| 4. Only injects copies of rows ALREADY present in OCR (ocr_count > 0) | |
| — never invents new row content. | |
| 5. Entire function wrapped in try/except — any error returns page_md unchanged. | |
| """ | |
| if not page_md or "<table" not in page_md.lower(): | |
| return page_md | |
| try: | |
| tl_rows = _extract_textlayer_rows(pdf_path, page_num) | |
| if not tl_rows: | |
| return page_md # scanned PDF or non-transaction page — safe no-op | |
| def _norm(s): | |
| return re.sub(r"\s+", " ", str(s or "").strip().lower()) | |
| # Build text-layer frequency map | |
| tl_freq = {} | |
| for r in tl_rows: | |
| key = (_norm(r["date"]), _norm(r["desc"]), _norm(r["amount"])) | |
| tl_freq[key] = tl_freq.get(key, 0) + 1 | |
| table_pattern = re.compile(r"<table[^>]*>.*?</table>", re.DOTALL | re.IGNORECASE) | |
| result = page_md | |
| for tbl_match in table_pattern.finditer(page_md): | |
| table_html = tbl_match.group(0) | |
| p = TableGridParser() | |
| p.feed(table_html) | |
| grid = _build_grid(p.rows) | |
| if len(grid) < 2: | |
| continue | |
| # Identify DATE, DESCRIPTION, AMOUNT column indices | |
| header = [str(c or "").strip().upper() for c in grid[0]] | |
| date_col = desc_col = amt_col = None | |
| for ci, h in enumerate(header): | |
| if re.search(r"\bDATE\b", h) and date_col is None: | |
| date_col = ci | |
| elif re.search(r"\b(DESCRIPTION|DETAILS?|NARRATION|PARTICULARS?)\b", h) and desc_col is None: | |
| desc_col = ci | |
| elif re.search(r"\b(AMOUNT|DEBIT|CREDIT|DR|CR)\b", h) and amt_col is None: | |
| amt_col = ci | |
| if date_col is None or desc_col is None or amt_col is None: | |
| continue # not a transaction table | |
| # Build OCR frequency map | |
| ocr_freq = {} | |
| ocr_rows_indexed = [] | |
| for ri, row in enumerate(grid[1:], start=1): | |
| cells = [str(c or "").strip() for c in row] | |
| if len(cells) <= max(date_col, desc_col, amt_col): | |
| continue | |
| d = _norm(cells[date_col]) | |
| desc = _norm(cells[desc_col]) | |
| amt = _norm(cells[amt_col]) | |
| if not d or not desc: | |
| continue | |
| key = (d, desc, amt) | |
| ocr_freq[key] = ocr_freq.get(key, 0) + 1 | |
| ocr_rows_indexed.append((ri, key)) | |
| # Guard: overlap check for Case A (duplicate restore). | |
| # For Case A we require at least one row in both OCR and text layer | |
| # to confirm we are patching the right table. | |
| # For Case B (entirely missing rows) we use a lighter structural check: | |
| # if the OCR table has DATE+DESCRIPTION+AMOUNT columns (already verified) | |
| # and the text layer date format matches the OCR date format, | |
| # we can safely inject — even when zero rows overlap. | |
| overlap = set(ocr_freq.keys()) & set(tl_freq.keys()) | |
| # Detect date format used in OCR table (MM/DD vs MM-DD vs Mon DD) | |
| _ocr_dates = [ | |
| _norm(str(row[date_col] or "")) | |
| for row in grid[1:] | |
| if len(row) > date_col and str(row[date_col] or "").strip() | |
| ] | |
| _tl_dates = [_norm(r["date"]) for r in tl_rows] | |
| _slash_re = re.compile(r"^\d{1,2}/\d{2}") | |
| _hyphen_re = re.compile(r"^\d{1,2}-\d{2}") | |
| def _date_fmt(dates): | |
| if any(_slash_re.match(d) for d in dates): return "slash" | |
| if any(_hyphen_re.match(d) for d in dates): return "hyphen" | |
| return "other" | |
| ocr_fmt = _date_fmt(_ocr_dates) | |
| tl_fmt = _date_fmt(_tl_dates) | |
| date_fmt_match = (ocr_fmt == tl_fmt) or "other" in (ocr_fmt, tl_fmt) | |
| # Determine missing rows. | |
| # Case A: row exists in OCR but count is too low (duplicate dropped) | |
| # Case B: row exists in text layer but completely absent from OCR | |
| to_inject = {} | |
| to_inject_new = {} | |
| for key in tl_freq: | |
| ocr_count = ocr_freq.get(key, 0) | |
| tl_count = tl_freq[key] | |
| if tl_count > ocr_count: | |
| missing = tl_count - ocr_count | |
| if ocr_count > 0: | |
| # Case A: restore duplicates — requires overlap confirmation | |
| if overlap: | |
| to_inject[key] = missing | |
| else: | |
| # Case B: entirely new row — requires date format match | |
| # (lighter guard: no overlap needed, just structural match) | |
| if date_fmt_match: | |
| tl_row_data = next( | |
| (r for r in tl_rows | |
| if (_norm(r["date"]), _norm(r["desc"]), _norm(r["amount"])) == key), | |
| None | |
| ) | |
| if tl_row_data: | |
| to_inject_new[key] = (missing, tl_row_data) | |
| if not to_inject and not to_inject_new: | |
| continue | |
| # Build patched grid — inject missing duplicate rows (Case A) | |
| new_grid = [grid[0]] | |
| for ri, row in enumerate(grid[1:], start=1): | |
| new_grid.append(row) | |
| cells = [str(c or "").strip() for c in row] | |
| if len(cells) <= max(date_col, desc_col, amt_col): | |
| continue | |
| d = _norm(cells[date_col]) | |
| desc = _norm(cells[desc_col]) | |
| amt = _norm(cells[amt_col]) | |
| key = (d, desc, amt) | |
| if key in to_inject and to_inject[key] > 0: | |
| last_occ = max(idx for idx, k in ocr_rows_indexed if k == key) | |
| if ri == last_occ: | |
| for _ in range(to_inject[key]): | |
| new_grid.append(list(row)) | |
| to_inject[key] = 0 | |
| new_table_html = _grid_to_html(new_grid) | |
| result = result.replace(table_html, new_table_html, 1) | |
| # Case B: completely missing rows → build a SEPARATE new table | |
| # appended after the patched OCR table. Using a separate table | |
| # preserves the original PDF structure (e.g. Capital One has | |
| # separate "Payments" and "Transactions" sections). | |
| if to_inject_new: | |
| ncols = len(grid[0]) | |
| header_row = grid[0] | |
| # Find post_date column index if it exists in the header | |
| post_date_col = None | |
| for ci, h in enumerate(header_row): | |
| hh = str(h or "").strip().upper() | |
| if "POST" in hh and "DATE" in hh: | |
| post_date_col = ci | |
| break | |
| new_rows = [] | |
| for key, (count, tl_row_data) in to_inject_new.items(): | |
| for _ in range(count): | |
| new_row = [""] * ncols | |
| new_row[date_col] = tl_row_data["date"] | |
| new_row[desc_col] = tl_row_data["desc"] | |
| new_row[amt_col] = tl_row_data["amount"] | |
| if post_date_col is not None: | |
| new_row[post_date_col] = tl_row_data.get("post_date", "") | |
| new_rows.append(new_row) | |
| log.info( | |
| "page %d: new table row from text layer: %s %s", | |
| page_num, tl_row_data["date"], tl_row_data["desc"][:40] | |
| ) | |
| if new_rows: | |
| extra_grid = [header_row] + new_rows | |
| extra_html = "\n\n" + _grid_to_html(extra_grid) | |
| # Insert immediately after the patched table | |
| insert_pos = result.find(new_table_html) + len(new_table_html) | |
| result = result[:insert_pos] + extra_html + result[insert_pos:] | |
| # Case C: page has text-layer rows but NO transaction table in OCR output. | |
| # Build a new table from the text layer and prepend it. | |
| # Guard: only fires when the page has NO table with both DATE + DESCRIPTION cols. | |
| # This is placed AFTER the for-loop so it only runs once per page, not per table. | |
| if tl_rows: | |
| _has_txn_table = False | |
| for _tbl in table_pattern.finditer(result): | |
| _p2 = TableGridParser() | |
| _p2.feed(_tbl.group(0)) | |
| _g2 = _build_grid(_p2.rows) | |
| if len(_g2) < 2: | |
| continue | |
| _h2 = [str(c or "").strip().upper() for c in _g2[0]] | |
| if (any(re.search(r"\bDATE\b", x) for x in _h2) and | |
| any(re.search(r"\b(DESCRIPTION|DETAILS?|NARRATION|PARTICULARS?)\b", x) for x in _h2)): | |
| _has_txn_table = True | |
| break | |
| if not _has_txn_table: | |
| _has_post = any(r.get("post_date") for r in tl_rows) | |
| if _has_post: | |
| _chdr = ["Date", "Post Date", "Transaction Description", "Amount"] | |
| _di2, _pi2, _xi2, _ai2 = 0, 1, 2, 3 | |
| else: | |
| _chdr = ["Date", "Transaction Description", "Amount"] | |
| _di2, _xi2, _ai2 = 0, 1, 2 | |
| _cnew_rows = [] | |
| for r in tl_rows: | |
| _crow = [""] * len(_chdr) | |
| _crow[_di2] = r["date"] | |
| _crow[_xi2] = r["desc"] | |
| _crow[_ai2] = r["amount"] | |
| if _has_post: | |
| _crow[_pi2] = r.get("post_date", "") | |
| _cnew_rows.append(_crow) | |
| result = _grid_to_html([_chdr] + _cnew_rows) + "\n\n" + result | |
| log.info("page %d: Case C — new table (%d rows) from text layer", | |
| page_num, len(_cnew_rows)) | |
| return result | |
| except Exception as e: | |
| log.warning("_patch_ocr_with_textlayer failed (page %d): %s", page_num, e) | |
| return page_md # always safe — return original on any error | |
| # -------------------------- | |
| # Fix 5: Mid-table separator row reconstruction | |
| # -------------------------- | |
| # Date pattern for transaction rows: MM/DD, MM/DD/YY, MM/DD/YYYY | |
| _DATE_RE = re.compile(r"^\d{1,2}/\d{2}(?:/\d{2,4})?$") | |
| # Standalone money amount (may be split off a separator label by OCR) | |
| _SPLIT_AMOUNT_RE = re.compile( | |
| r"^-?\$?\d{1,3}(?:,\d{3})*(?:\.\d{1,4})?$|^\$-?\d{1,3}(?:,\d{3})*(?:\.\d{1,4})?$" | |
| ) | |
| def _reconstruct_separator_rows(grid): | |
| """ | |
| Detect and reconstruct mid-table separator / label rows that OCR has | |
| incorrectly split across columns. | |
| The artifact (Bank of America and similar): | |
| A section label like "Card account # XXXX XXXX XXXX 2889" sits between | |
| transaction rows as a full-width label. OCR splits the trailing digits | |
| across columns because they are right-aligned, producing variations like: | |
| 2-cell split: ["Card account # XXXX XXXX XXXX 2", "", "889"] | |
| 3-cell split: ["Card account # XXXX XXXX XXXX", "2", "889"] | |
| clean 1-cell: ["Card account # XXXX XXXX XXXX 2889", "", ""] | |
| Detection criteria (ALL must hold — guards safe for all other PDFs): | |
| 1. First cell is NOT a date token (transaction rows always start with date). | |
| 2. First cell contains letters (it is a label, not a bare number). | |
| 3. All cells except the first either: | |
| a. are empty, OR | |
| b. are a short pure-digit fragment (1-4 digits, no decimal, no sign) | |
| — these are the split-off tails of the account/card number. | |
| 4. There must be at least one non-empty cell after the first (to detect | |
| the split; single-cell rows are also handled as clean label rows). | |
| The fix: | |
| Concatenate all non-empty cells in order, place the result in col 0, | |
| blank all other cells. | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| ncols = len(grid[0]) | |
| if ncols < 2: | |
| return grid | |
| new_grid = [grid[0]] # keep header unchanged | |
| for row in grid[1:]: | |
| cells = [str(c or "").strip() for c in row] | |
| while len(cells) < ncols: | |
| cells.append("") | |
| non_empty = [(i, c) for i, c in enumerate(cells) if c] | |
| # Completely empty row — leave as-is | |
| if len(non_empty) == 0: | |
| new_grid.append(row) | |
| continue | |
| first_idx, first_val = non_empty[0] | |
| # Guard 1: first cell must not be a date token | |
| if _DATE_RE.match(first_val): | |
| new_grid.append(row) | |
| continue | |
| # Guard 2: first cell must contain letters (label, not a bare number) | |
| if not re.search(r"[A-Za-z]", first_val): | |
| new_grid.append(row) | |
| continue | |
| # Single non-empty cell — already a clean label row | |
| if len(non_empty) == 1: | |
| new_row = [""] * ncols | |
| new_row[0] = first_val | |
| new_grid.append(new_row) | |
| continue | |
| # Multiple non-empty cells: check that every cell AFTER the first | |
| # is a short pure-digit fragment (1-4 digits, no decimal, no sign). | |
| # This is the key generic guard — it allows 2, 3, or more cells | |
| # as long as all the trailing cells are digit-only fragments. | |
| # Real transaction rows always have amounts with decimals (-14.19) | |
| # or descriptions with letters, so they never pass this check. | |
| trailing = [val for _, val in non_empty[1:]] | |
| all_trailing_are_digit_fragments = all( | |
| re.fullmatch(r"\d{1,4}", v) for v in trailing | |
| ) | |
| if not all_trailing_are_digit_fragments: | |
| new_grid.append(row) | |
| continue | |
| # Reconstruct: concatenate all non-empty cells in order | |
| reconstructed = "".join(val for _, val in non_empty).strip() | |
| new_row = [""] * ncols | |
| new_row[0] = reconstructed | |
| new_grid.append(new_row) | |
| return new_grid | |
| # -------------------------- | |
| # Normalizer entry point | |
| # -------------------------- | |
| def _merge_split_rows(grid): | |
| """ | |
| Fix OCR artifact where a transaction row is split across two <tr> rows | |
| because the description wrapped to a second line in the source PDF. | |
| Two patterns detected (Hardin County Bank and similar monospace statements): | |
| Pattern A — description + empty continuation: | |
| Row N: [desc, '', '', '', ''] ← description, no date/money | |
| Row N+1: ['', debit, credit, date, bal] ← money/date, no description | |
| → Merge into: [desc, debit, credit, date, bal] | |
| Pattern B — description + text continuation + money: | |
| Row N: [desc, '', '', '', ''] ← first line of description | |
| Row N+1: [desc_cont, debit, credit, date, bal] ← continuation + money | |
| → Merge into: [desc + ' ' + desc_cont, debit, credit, date, bal] | |
| Guard conditions (no-op unless both rows match the pattern): | |
| - Row N must have non-empty col 0 (description). | |
| - Row N must have empty date column (col 3 or wherever DATE is). | |
| - Row N must have empty debit AND credit columns. | |
| - Row N+1 must have non-empty date column. | |
| - Row N+1 must have non-empty debit OR credit column. | |
| - Row N balance (if present) must be a short fragment without a decimal | |
| (1-6 chars, no '.') — confirms it's a split-off ref number, not a balance. | |
| """ | |
| if not grid or len(grid) < 3: | |
| return grid | |
| ncols = len(grid[0]) | |
| if ncols < 3: | |
| return grid | |
| # Identify column indices from header | |
| header = [str(c or "").strip().upper() for c in grid[0]] | |
| date_col = desc_col = bal_col = None | |
| debit_cols = [] | |
| credit_cols = [] | |
| for ci, h in enumerate(header): | |
| if re.search(r"\bDATE\b", h) and date_col is None: | |
| date_col = ci | |
| if re.search(r"\b(DESCRIPTION|DETAILS?|NARRATION|PARTICULARS?|TRANSACTION)\b", h) and desc_col is None: | |
| desc_col = ci | |
| if re.search(r"\bBALANCE\b", h) and bal_col is None: | |
| bal_col = ci | |
| if re.search(r"\b(DEBIT|DEBITS|DR|WITHDRAWAL|SUBTRACTIONS?)\b", h): | |
| debit_cols.append(ci) | |
| if re.search(r"\b(CREDIT|CREDITS|CR|DEPOSIT|ADDITIONS?)\b", h): | |
| credit_cols.append(ci) | |
| # Need at least date + description + one money column | |
| if date_col is None or desc_col is None: | |
| return grid | |
| if not debit_cols and not credit_cols: | |
| return grid | |
| money_cols = debit_cols + credit_cols | |
| _MONEY_RE2 = re.compile(r"^-?\$?\d{1,3}(?:,\d{3})*\.\d{1,4}$") | |
| _DATE_RE2 = re.compile(r"^\d{1,2}/\d{2}(?:/\d{2,4})?$|^\d{1,2}-\d{2}(?:-\d{2,4})?$") | |
| # Short fragment guard: 1-6 non-space chars, no decimal point → split-off ref tail | |
| _FRAGMENT_RE = re.compile(r"^[^\s\.]{1,6}$") | |
| new_grid = [grid[0]] | |
| i = 1 | |
| while i < len(grid): | |
| row = [str(c or "").strip() for c in grid[i]] | |
| while len(row) < ncols: | |
| row.append("") | |
| # Check if this row is a "description-only" split row candidate | |
| has_desc = bool(row[desc_col]) | |
| no_date = not row[date_col] | |
| no_money = all(not row[c] for c in money_cols) | |
| bal_is_frag = (bal_col is not None and | |
| (_FRAGMENT_RE.match(row[bal_col]) or not row[bal_col])) | |
| if has_desc and no_date and no_money and bal_is_frag and (i + 1) < len(grid): | |
| next_row = [str(c or "").strip() for c in grid[i + 1]] | |
| while len(next_row) < ncols: | |
| next_row.append("") | |
| next_has_date = bool(next_row[date_col]) and _DATE_RE2.match(next_row[date_col]) | |
| next_has_money = any(bool(next_row[c]) for c in money_cols) | |
| if next_has_date and next_has_money: | |
| # Pattern A or B — merge | |
| merged = list(next_row) | |
| # Append description (and possible continuation from next row) | |
| if next_row[desc_col]: | |
| # Pattern B: next row has desc continuation | |
| merged[desc_col] = row[desc_col] + " " + next_row[desc_col] | |
| else: | |
| # Pattern A: next row has no description | |
| merged[desc_col] = row[desc_col] | |
| new_grid.append(merged) | |
| i += 2 # skip both rows, emit merged | |
| continue | |
| new_grid.append(row) | |
| i += 1 | |
| return new_grid | |
| def _extract_fused_desc_amount(grid): | |
| """ | |
| Fix OCR artifact where a trailing money amount gets fused into the | |
| description cell instead of its own debit/credit column. | |
| Example (Hardin County Bank): | |
| OCR: ['CHASE CREDIT CRD EPAY 8319249882 500.00', '', '', '04/11/25', '16,699.04'] | |
| Fix: ['CHASE CREDIT CRD EPAY 8319249882', '500.00', '', '04/11/25', '16,699.04'] | |
| Guard conditions (all must hold): | |
| - Description ends with a space + money amount (NNN.NN or N,NNN.NN). | |
| - ALL debit AND credit columns are empty for this row. | |
| - Date column is non-empty (confirms it is a complete data row). | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| ncols = len(grid[0]) | |
| header = [str(c or "").strip().upper() for c in grid[0]] | |
| date_col = desc_col = None | |
| debit_cols = [] | |
| credit_cols = [] | |
| for ci, h in enumerate(header): | |
| if re.search(r"\bDATE\b", h) and date_col is None: | |
| date_col = ci | |
| if re.search(r"\b(DESCRIPTION|DETAILS?|NARRATION|PARTICULARS?|TRANSACTION)\b", h) and desc_col is None: | |
| desc_col = ci | |
| if re.search(r"\b(DEBIT|DEBITS|DR|WITHDRAWAL|SUBTRACTIONS?)\b", h): | |
| debit_cols.append(ci) | |
| if re.search(r"\b(CREDIT|CREDITS|CR|DEPOSIT|ADDITIONS?)\b", h): | |
| credit_cols.append(ci) | |
| if date_col is None or desc_col is None or not debit_cols: | |
| return grid | |
| money_cols = debit_cols + credit_cols | |
| _TRAILING_AMT = re.compile(r"^(.+?)\s+(-?\d{1,3}(?:,\d{3})*\.\d{2})$") | |
| debit_col = debit_cols[0] | |
| new_grid = [grid[0]] | |
| for row in grid[1:]: | |
| cells = [str(c or "").strip() for c in row] | |
| while len(cells) < ncols: | |
| cells.append("") | |
| desc = cells[desc_col] | |
| no_money = all(not cells[c] for c in money_cols) | |
| has_date = bool(cells[date_col]) | |
| if desc and no_money and has_date: | |
| m = _TRAILING_AMT.match(desc) | |
| if m: | |
| cells = list(cells) | |
| cells[desc_col] = m.group(1).strip() | |
| cells[debit_col] = m.group(2) | |
| new_grid.append(cells) | |
| continue | |
| new_grid.append(cells) | |
| return new_grid | |
| def _is_junk_table(grid): | |
| """ | |
| Detect and suppress known junk tables that are OCR artifacts from | |
| page headers/footers, not real transaction data. | |
| Current patterns: | |
| 1. CUSTOMER INFORMATION fragment tables (First Horizon Bank): | |
| Single-column table with header "CUSTOMER INFORMATION" and | |
| data rows that are short numeric fragments (e.g. "254", "24") | |
| from the partial account number / year printed in the header band. | |
| """ | |
| if not grid or len(grid) < 2: | |
| return False | |
| ncols = len(grid[0]) | |
| if ncols != 1: | |
| return False | |
| header_text = str(grid[0][0] or "").strip().upper() | |
| if "CUSTOMER INFORMATION" in header_text: | |
| # All data rows should be short (< 10 chars) numeric/alphanumeric fragments | |
| data_rows = grid[1:] | |
| if all(len(str(r[0] or "").strip()) <= 10 for r in data_rows): | |
| return True | |
| return False | |
| def _split_fused_multicolumn_header(grid): | |
| """ | |
| Fix OCR artifact where multiple column headers are fused into one <th> cell. | |
| Example (First Horizon Bank alternating pages): | |
| Fused: ['DATE DESCRIPTION CARD #', 'DEPOSIT', 'WITHDRAWAL'] | |
| Correct: ['DATE', 'DESCRIPTION', 'DEPOSIT', 'WITHDRAWAL', 'CARD #'] | |
| The data rows also have date + description + card# all in col 0: | |
| Fused: ['01/16 PURCHASE - UBER TRIP... 4919', '', '$21.02'] | |
| Correct: ['01/16', 'PURCHASE - UBER TRIP...', '', '$21.02', '4919'] | |
| Detection: first header cell contains 2+ keyword terms separated by spaces, | |
| AND remaining header cells are valid money column keywords. | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| ncols = len(grid[0]) | |
| if ncols < 2: | |
| return grid | |
| first_header = str(grid[0][0] or "").strip() | |
| # Split first header cell into parts and check if multiple are keywords | |
| parts = first_header.split() | |
| kw_hits = [p for p in parts if _HEADER_KW_EXACT_RE.match(p)] | |
| if len(kw_hits) < 2: | |
| return grid # not a fused multi-keyword header | |
| # The remaining header cells must all be keyword columns | |
| rest_headers = [str(c or "").strip() for c in grid[0][1:]] | |
| if not all(_HEADER_KW_EXACT_RE.match(h) for h in rest_headers if h): | |
| return grid | |
| # Build the new expanded header: | |
| # Split the fused first cell into individual keyword columns | |
| # Keep non-keyword trailing parts (e.g. "CARD #") as the last new col | |
| new_header_cols = [] | |
| non_kw_parts = [] | |
| for p in parts: | |
| if _HEADER_KW_EXACT_RE.match(p): | |
| if non_kw_parts: | |
| new_header_cols.append(" ".join(non_kw_parts)) | |
| non_kw_parts = [] | |
| new_header_cols.append(p) | |
| else: | |
| non_kw_parts.append(p) | |
| if non_kw_parts: | |
| new_header_cols.append(" ".join(non_kw_parts)) | |
| # Full new header = expanded first cell + remaining cells | |
| new_header = new_header_cols + rest_headers | |
| new_ncols = len(new_header) | |
| extra_cols = new_ncols - ncols # how many new columns were added | |
| # Identify column roles in the new header | |
| _DATE_COL_RE = re.compile(r"\bDATE\b", re.IGNORECASE) | |
| _CARD_COL_RE = re.compile(r"\bCARD\b", re.IGNORECASE) | |
| _DESC_COL_RE = re.compile(r"\b(DESCRIPTION|DETAILS?|NARRATION|PARTICULARS?)\b", re.IGNORECASE) | |
| _MONEY_COL_RE = re.compile(r"\b(DEPOSIT|WITHDRAWAL|DEBIT|CREDIT|AMOUNT|ADDITION|SUBTRACTION)S?\b", re.IGNORECASE) | |
| new_date_col = next((i for i,h in enumerate(new_header) if _DATE_COL_RE.search(h)), None) | |
| new_desc_col = next((i for i,h in enumerate(new_header) if _DESC_COL_RE.search(h)), None) | |
| new_card_col = next((i for i,h in enumerate(new_header) if _CARD_COL_RE.search(h)), None) | |
| if new_date_col is None or new_desc_col is None: | |
| return grid # can't map columns | |
| # Re-split each data row: col 0 had "date description... card#" fused together | |
| _DATE_PREFIX = re.compile(r"^(\d{1,2}/\d{2}(?:/\d{2,4})?)\s+(.*)", re.DOTALL) | |
| _CARD_SUFFIX = re.compile(r"^(.*?)\s+(\d{4})$", re.DOTALL) # 4-digit card number at end | |
| def pad(row, n): | |
| r = list(row) | |
| while len(r) < n: r.append("") | |
| return r[:n] | |
| new_grid = [pad(new_header, new_ncols)] | |
| for row in grid[1:]: | |
| cells = [str(c or "").strip() for c in row] | |
| while len(cells) < ncols: cells.append("") | |
| fused_cell = cells[0] | |
| rest_cells = cells[1:] | |
| # Extract date from front | |
| date_val = "" | |
| desc_val = fused_cell | |
| card_val = "" | |
| dm = _DATE_PREFIX.match(fused_cell) | |
| if dm: | |
| date_val = dm.group(1) | |
| desc_val = dm.group(2).strip() | |
| # Extract card# from end (4-digit number) | |
| if new_card_col is not None: | |
| cm = _CARD_SUFFIX.match(desc_val) | |
| if cm: | |
| desc_val = cm.group(1).strip() | |
| card_val = cm.group(2) | |
| # Build new row | |
| new_row = [""] * new_ncols | |
| new_row[new_date_col] = date_val | |
| new_row[new_desc_col] = desc_val | |
| if new_card_col is not None: | |
| new_row[new_card_col] = card_val | |
| # Fill remaining money columns from rest_cells | |
| money_col_indices = [i for i,h in enumerate(new_header) if _MONEY_COL_RE.search(h)] | |
| for mi, ri in enumerate(range(len(rest_cells))): | |
| if mi < len(money_col_indices): | |
| new_row[money_col_indices[mi]] = rest_cells[ri] | |
| new_grid.append(new_row) | |
| return new_grid | |
| def _normalize_daily_balance_table(grid): | |
| """ | |
| Fix OCR artifact in DAILY BALANCE SUMMARY tables where: | |
| - OCR produces N DATE columns followed by N BALANCE columns | |
| instead of interleaved DATE|BALANCE|DATE|BALANCE pairs | |
| - Multiple dates are stacked in one cell "01/02\n01/03\n01/04\n01/05" | |
| Reorders columns and expands stacked date cells into proper rows. | |
| Only fires when ALL of these hold: | |
| - Header has exactly 2N columns where N >= 2 | |
| - First N columns are all DATE, last N columns are all BALANCE | |
| - At least one data cell contains a newline (stacked dates) | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| header = [str(c or "").strip().upper() for c in grid[0]] | |
| ncols = len(header) | |
| if ncols < 4 or ncols % 2 != 0: | |
| return grid | |
| half = ncols // 2 | |
| date_half = header[:half] | |
| bal_half = header[half:] | |
| if not all(h == "DATE" for h in date_half): | |
| return grid | |
| if not all(h == "BALANCE" for h in bal_half): | |
| return grid | |
| # Build new interleaved header: DATE BALANCE DATE BALANCE ... | |
| new_header = [] | |
| for i in range(half): | |
| new_header.append("DATE") | |
| new_header.append("BALANCE") | |
| # Expand each data row | |
| new_rows = [] | |
| _DATE_RE3 = re.compile(r"^\d{1,2}/\d{2}(?:/\d{2,4})?$") | |
| for row in grid[1:]: | |
| cells = [str(c or "").strip() for c in row] | |
| while len(cells) < ncols: | |
| cells.append("") | |
| # Dates are all stacked in cells[0]; balances follow immediately in cells[1..N] | |
| # (OCR layout: stacked_dates | bal1 | bal2 | bal3 | bal4 | empty...) | |
| # NOT: date | date | date | date | bal | bal | bal | bal (despite header order) | |
| date_cells = [cells[0]] | |
| bal_cells = cells[1:] | |
| # Check if dates are stacked in first cell (newline OR space-separated) | |
| # e.g. "01/02 01/03 01/04 01/05" or "01/02\n01/03\n01/04\n01/05" | |
| first = date_cells[0] | |
| tokens = re.split(r"[\n\r\s]+", first) | |
| date_tokens = [t.strip() for t in tokens if t.strip() and _DATE_RE3.match(t.strip())] | |
| if len(date_tokens) >= 2: | |
| stacked = date_tokens | |
| else: | |
| stacked = date_cells # already one date per cell | |
| # Build ONE interleaved row for this group of stacked dates | |
| new_row = [""] * ncols | |
| for i, date_val in enumerate(stacked): | |
| if not _DATE_RE3.match(date_val): | |
| continue | |
| bal_val = bal_cells[i] if i < len(bal_cells) else "" | |
| new_row[i * 2] = date_val | |
| new_row[i * 2 + 1] = bal_val | |
| new_rows.append(new_row) | |
| if not new_rows: | |
| return grid | |
| return [new_header] + new_rows | |
| def _normalize_checks_paid_table(grid): | |
| """ | |
| Fix OCR artifact in CHECKS PAID SUMMARY tables where: | |
| - Header col0 = "DATE CHECK # DATE CHECK # DATE CHECK #" (fused repeating groups) | |
| - Remaining headers = "AMOUNT AMOUNT AMOUNT" | |
| - Data col0 = "01/24 4701 01/18 4704 * 01/18 916831 *" (all groups fused) | |
| - Remaining data cells = amount values | |
| Expands to proper: DATE | CHECK # | AMOUNT | DATE | CHECK # | AMOUNT | ... | |
| Detection: first header cell matches pattern (DATE CHECK #)+ | |
| and remaining headers are all AMOUNT. | |
| """ | |
| if not grid or len(grid) < 2: | |
| return grid | |
| ncols = len(grid[0]) | |
| if ncols < 2: | |
| return grid | |
| first_h = str(grid[0][0] or "").strip() | |
| _CHECKS_HDR = re.compile(r"^(DATE\s+CHECK\s+#\s*)+$", re.IGNORECASE) | |
| if not _CHECKS_HDR.match(first_h): | |
| return grid | |
| # Remaining headers must be AMOUNT (or empty) | |
| rest_h = [str(c or "").strip().upper() for c in grid[0][1:]] | |
| if not all(h in ("AMOUNT", "") for h in rest_h if h): | |
| return grid | |
| # Count groups from how many times DATE appears in the fused header | |
| n_groups = len(re.findall(r"\bDATE\b", first_h, re.IGNORECASE)) | |
| if n_groups < 1: | |
| return grid | |
| # Build new header: DATE | CHECK # | AMOUNT repeated n_groups times | |
| new_header = [] | |
| for _ in range(n_groups): | |
| new_header.extend(["DATE", "CHECK #", "AMOUNT"]) | |
| _DATE_RE = re.compile(r"^\d{1,2}/\d{2}$") | |
| new_rows = [new_header] | |
| for row in grid[1:]: | |
| cells = [str(c or "").strip() for c in row] | |
| while len(cells) < ncols: | |
| cells.append("") | |
| fused_data = cells[0] | |
| amounts = cells[1:] | |
| # Split fused data into groups by date boundary | |
| tokens = fused_data.split() | |
| groups = [] | |
| current = [] | |
| for tok in tokens: | |
| if _DATE_RE.match(tok) and current: | |
| groups.append(current) | |
| current = [tok] | |
| else: | |
| current.append(tok) | |
| if current and current[0]: | |
| groups.append(current) | |
| # Build one output row | |
| new_row = [""] * len(new_header) | |
| for i, grp in enumerate(groups): | |
| if i >= n_groups: | |
| break | |
| new_row[i * 3] = grp[0] # DATE | |
| new_row[i * 3 + 1] = " ".join(grp[1:]).strip() # CHECK # | |
| new_row[i * 3 + 2] = amounts[i] if i < len(amounts) else "" # AMOUNT | |
| new_rows.append(new_row) | |
| return new_rows | |
| # ── Johnson Bank plain-text section converter ───────────────────────────── | |
| # Johnson Bank formats Deposits, Withdrawals, Checks, and Daily Account Balance | |
| # as plain text columns (no HTML tables). This converter detects those sections | |
| # and emits proper <table> HTML. | |
| # | |
| # SAFETY GUARDS — all four must hold before any conversion fires: | |
| # 1. Line matches exact section-header pattern | |
| # 2. Next non-blank line matches exact column-header pattern | |
| # 3. First data row uses MM-DD date format (Johnson Bank's unique date separator) | |
| # 4. The candidate section contains NO existing <table> tags | |
| # Guard 3 is the key discriminator: all other supported banks use MM/DD. | |
| _JB_SECTION_HDR = re.compile( | |
| r"^(Deposits?|Withdrawals?|Checks?(?:\s+Paid)?|Daily\s+Account\s+Balance)" | |
| r"\s*(?:\(cont\.?\))?\s*$", | |
| re.IGNORECASE, | |
| ) | |
| _JB_COL_HDR = re.compile( | |
| r"^Date\s+(Description|Number)\s+Amount\s*$" | |
| r"|^Date\s+(?:Number\s+)?(?:Balance|Amount)\s*$", | |
| re.IGNORECASE, | |
| ) | |
| # MM-DD date (dash separator) — Johnson Bank's unique format | |
| _JB_DATE_DD = re.compile(r"^\d{2}-\d{2}\s+") | |
| _JB_TXN_ROW = re.compile(r"^(\d{2}-\d{2})\s+(.+?)\s+(-?\d{1,3}(?:,\d{3})*\.\d{2})\s*$") | |
| _JB_BAL_ROW = re.compile(r"^(\d{2}-\d{2})\s+(\d{1,3}(?:,\d{3})*\.\d{2})\s*$") | |
| def _jb_rows_to_html(header_cols, rows): | |
| th = "".join("<th>" + h + "</th>" for h in header_cols) | |
| parts = ["<table>", "<tr>" + th + "</tr>"] | |
| for row in rows: | |
| td = "".join("<td>" + str(row.get(h, "")) + "</td>" for h in header_cols) | |
| parts.append("<tr>" + td + "</tr>") | |
| parts.append("</table>") | |
| return "\n".join(parts) | |
| def _jb_section_has_table(lines, start, end): | |
| """Return True if any line in lines[start:end] contains a <table tag.""" | |
| return any("<table" in l.lower() for l in lines[start:end]) | |
| def _find_first_data_row(lines, start): | |
| """Return index of first non-blank line after start, or None.""" | |
| for i in range(start, min(start + 20, len(lines))): | |
| if lines[i].strip(): | |
| return i | |
| return None | |
| def convert_plaintext_bank_sections(text: str) -> str: | |
| """ | |
| Convert Johnson Bank plain-text transaction sections to HTML tables. | |
| All four safety guards must pass before conversion fires. | |
| Safe no-op for all other banks. | |
| """ | |
| # Quick exit: if the page already has lots of tables, skip entirely | |
| # (handles pages that are already properly formatted) | |
| lines = text.splitlines() | |
| out = [] | |
| i = 0 | |
| while i < len(lines): | |
| line = lines[i].strip() | |
| # Guard 1: section header | |
| if not _JB_SECTION_HDR.match(line): | |
| out.append(lines[i]) | |
| i += 1 | |
| continue | |
| section_title = line | |
| j = i + 1 | |
| # Skip blank lines to find column header | |
| while j < len(lines) and not lines[j].strip(): | |
| j += 1 | |
| # Guard 2: column header | |
| if j >= len(lines) or not _JB_COL_HDR.match(lines[j].strip()): | |
| out.append(lines[i]) | |
| i += 1 | |
| continue | |
| col_hdr_line = lines[j].strip() | |
| data_start = j + 1 | |
| # Find first non-blank data line | |
| first_data_idx = _find_first_data_row(lines, data_start) | |
| # Guard 3: first data row must use MM-DD date format | |
| if first_data_idx is None or not _JB_DATE_DD.match(lines[first_data_idx].strip()): | |
| out.append(lines[i]) | |
| i += 1 | |
| continue | |
| # Find section end (next section header, page separator, or end of text) | |
| section_end = len(lines) | |
| for k in range(data_start, len(lines)): | |
| l = lines[k].strip() | |
| if (l != section_title and _JB_SECTION_HDR.match(l) | |
| or l.startswith("---page-separator") | |
| or l.startswith("Page:")): | |
| section_end = k | |
| break | |
| # Guard 4: no existing <table> tags in this section | |
| if _jb_section_has_table(lines, i, section_end): | |
| out.append(lines[i]) | |
| i += 1 | |
| continue | |
| # All guards passed — parse and convert | |
| parts_upper = col_hdr_line.upper().split() | |
| if len(parts_upper) == 2 and parts_upper[1] == "BALANCE": | |
| header_cols = ["Date", "Balance"] | |
| elif "NUMBER" in parts_upper: | |
| header_cols = ["Date", "Number", "Amount"] | |
| else: | |
| header_cols = ["Date", "Description", "Amount"] | |
| out.append(section_title) | |
| rows = [] | |
| current = None | |
| k = data_start | |
| while k < section_end: | |
| raw = lines[k] | |
| row_line = raw.strip() | |
| if not row_line: | |
| k += 1 | |
| continue | |
| if (_JB_SECTION_HDR.match(row_line) | |
| or row_line.startswith("---page-separator") | |
| or row_line.startswith("Page:")): | |
| break | |
| # Balance-only row | |
| mb = _JB_BAL_ROW.match(row_line) | |
| if mb and header_cols == ["Date", "Balance"]: | |
| if current: | |
| rows.append(current) | |
| current = {"Date": mb.group(1), "Balance": mb.group(2)} | |
| k += 1 | |
| continue | |
| # Full transaction row | |
| mt = _JB_TXN_ROW.match(row_line) | |
| if mt: | |
| if current: | |
| rows.append(current) | |
| desc_key = "Description" if "Description" in header_cols else "Number" | |
| current = { | |
| "Date": mt.group(1), | |
| desc_key: mt.group(2).strip(), | |
| "Amount": mt.group(3), | |
| } | |
| k += 1 | |
| continue | |
| # Continuation line | |
| if current and row_line: | |
| desc_key = "Description" if "Description" in header_cols else "Number" | |
| if desc_key in current: | |
| current[desc_key] += " " + row_line | |
| k += 1 | |
| continue | |
| k += 1 | |
| if current: | |
| rows.append(current) | |
| if rows: | |
| out.append(_jb_rows_to_html(header_cols, rows)) | |
| i = section_end # jump past the whole section | |
| continue | |
| return "\n".join(out) | |
| def normalize_html_tables(text: str) -> str: | |
| """ | |
| For every <table>...</table>: | |
| 1. Parse to grid (expand colspan/rowspan). | |
| 2. Drop truly-empty columns. | |
| 3. Merge blank-header text columns into the left column. | |
| 4. Fix 1 — clean fused header artifacts and recover the lost first data row. | |
| 5. Fix 2 — promote a misplaced header row when real column headers landed | |
| in a <td> data row instead of the <th> header row. | |
| 6. Fix 3 — split fused key-value rows in summary sections. | |
| 7. Emit normalised <table> HTML. | |
| """ | |
| if not text or "<table" not in text.lower(): | |
| return text | |
| pattern = re.compile(r"<table[^>]*>.*?</table>", re.DOTALL | re.IGNORECASE) | |
| out = [] | |
| last = 0 | |
| for m in pattern.finditer(text): | |
| out.append(text[last : m.start()]) | |
| table_html = m.group(0) | |
| try: | |
| p = TableGridParser() | |
| p.feed(table_html) | |
| grid = _build_grid(p.rows) | |
| # Suppress known junk tables (OCR artifacts from page headers) | |
| if _is_junk_table(grid): | |
| # Append nothing (suppress the table), update last, skip rest of pipeline | |
| out.append("") | |
| last = m.end() # must set here before continue since line below is outside try | |
| continue | |
| grid = _drop_truly_empty_columns(grid) | |
| grid = _merge_blank_header_text_columns(grid) | |
| # Fix 5: reconstruct mid-table separator rows split across columns by OCR | |
| grid = _reconstruct_separator_rows(grid) | |
| # Fix 1: fused-header artifact (two-signal guard — safe on clean PDFs) | |
| grid, recovered_row = _clean_header_artifacts(grid) | |
| if recovered_row is not None: | |
| grid.insert(1, recovered_row) | |
| # Fix 2: misplaced header row promotion (keyword-score guard — safe on clean PDFs) | |
| grid = _promote_misplaced_header_row(grid) | |
| # Fix 8: split fused multi-keyword header cell (First Horizon Bank) | |
| grid = _split_fused_multicolumn_header(grid) | |
| # Fix 9: normalize DAILY BALANCE SUMMARY tables with stacked dates | |
| grid = _normalize_daily_balance_table(grid) | |
| # Fix 10: normalize CHECKS PAID SUMMARY repeating-group tables | |
| grid = _normalize_checks_paid_table(grid) | |
| # Fix 6: merge split rows where description wraps to next line | |
| # (Hardin County Bank and similar monospace statement formats) | |
| grid = _merge_split_rows(grid) | |
| # Fix 7: extract trailing amount fused into description cell | |
| # (Hardin County Bank: "CHASE CREDIT CRD EPAY 8319249882 500.00") | |
| grid = _extract_fused_desc_amount(grid) | |
| # Fix 3: fused key-value rows in summary sections (no-space guard — safe on all PDFs) | |
| grid = _fix_fused_keyvalue_rows(grid) | |
| out.append(_grid_to_html(grid) if grid else table_html) | |
| except Exception: | |
| out.append(table_html) | |
| last = m.end() | |
| out.append(text[last:]) | |
| return "".join(out) | |
| def stabilize_tables_and_text(page_md: str) -> str: | |
| 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(b) | |
| stabilized = "\n\n".join(out_blocks) | |
| stabilized = convert_plaintext_bank_sections(stabilized) | |
| stabilized = normalize_html_tables(stabilized) | |
| return close_unclosed_html(stabilized) | |
| # -------------------------- | |
| # PDF rendering with padding | |
| # -------------------------- | |
| def render_pdf_pages_to_images(pdf_path: str) -> Tuple[List[str], List[int]]: | |
| import pymupdf as fitz | |
| from PIL import Image, ImageEnhance | |
| 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) | |
| if ENABLE_CONTRAST: | |
| img = ImageEnhance.Contrast(img).enhance(1.12) | |
| 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 | |
| img_path = os.path.join(tempfile.gettempdir(), f"glmocr_page_{os.getpid()}_{i}.png") | |
| img.save(img_path, "PNG", compress_level=6) | |
| page_images.append(img_path) | |
| page_heights.append(img.height) | |
| doc.close() | |
| return page_images, page_heights | |
| # -------------------------- | |
| # GLM-OCR result extraction | |
| # -------------------------- | |
| 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 | |
| # -------------------------- | |
| # Main entry | |
| # -------------------------- | |
| def run_ocr(uploaded_file): | |
| if uploaded_file is None: | |
| return "Please upload a file." | |
| page_images = [] | |
| 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 = [] | |
| 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 = [] | |
| # Header inclusion: PDF text -> band words -> OCR band | |
| 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(normalize_html_tables(fix_account_number(normalize_money_glyphs(hdr.strip())))) | |
| # Main OCR body: stabilize then patch missing duplicate rows from text layer | |
| if page_md and page_md.strip(): | |
| stabilized = stabilize_tables_and_text(page_md.strip()) | |
| # Fix 4: restore any rows OCR dropped by cross-checking the PDF text layer. | |
| # Safe no-op for scanned PDFs (no text layer) and non-transaction pages. | |
| if is_pdf: | |
| stabilized = _patch_ocr_with_textlayer(stabilized, path, page_num) | |
| parts.append(stabilized) | |
| # Footer extraction: PDF text layer -> band words -> OCR crop | |
| # Deduplication guard prevents double-printing when body OCR already captured it. | |
| 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()) | |
| # Guard 1: first-line dedup (original check) | |
| 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 | |
| ) | |
| # Guard 2: suppress footer that is a raw text-layer dump of | |
| # transaction rows (e.g. East West Bank two-column layout). | |
| # Detected when the footer contains 3+ date tokens (MM/DD or MM-DD) | |
| # AND the footer itself contains amounts — meaning it is transaction | |
| # data, not a legitimate page footer like an address or disclaimer. | |
| _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)) | |
| return "\n\n---page-separator---\n\n".join(all_pages) if all_pages else "(No content)" | |
| 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 p.endswith(".png") and "glmocr_page_" in os.path.basename(p): | |
| os.unlink(p) | |
| except Exception: | |
| pass | |
| with gr.Blocks(title="GLM-OCR") as demo: | |
| gr.Markdown("# GLM-OCR\nUpload a PDF or image. Headers included; tables stabilized.") | |
| 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)") | |
| run_btn.click(fn=run_ocr, inputs=file_in, outputs=out) | |
| if __name__ == "__main__": | |
| demo.launch() | |