"""
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 = "e2b138b2005a41cb9d87dd18805838aa.lyd51L23rcDbsw0w"
# 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"", 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
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
# ---- Generic HTML table normalizer ----
class TableGridParser(HTMLParser):
"""Parse a 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 = [""]
for r, row in enumerate(grid):
lines.append("")
tag = "th" if r == 0 else "td"
for cell in row:
escaped = (
(cell or "")
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
)
lines.append(f"<{tag}>{escaped}")
lines.append("
")
lines.append("
")
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
data row instead of the | header row.
Two scenarios handled:
Scenario A — Simple misplaced header (e.g. TD Bank):
| ACCOUNT ACTIVITY | ← section title, not a real header
DATE DESCRIPTION | ... ← real column headers in a data row
01/05 ... | ← 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):
208479667 | Beginning Balance:$20.48 Ending Balance:$3.46 |
Date Description | Debits | Credits | Balance |
04/01 DEBIT CARD... 8.01 | ...
→ 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
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 "]*>.*?
", 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: NO transaction table exists on this page at all, but text
# layer has valid rows. Build a brand-new table and prepend it to
# the page content. This handles pages where OCR dropped the entire
# transaction section (e.g. East West Bank page 7 with only 2 rows).
#
# Guard: only fire if no table on this page already has DATE+DESCRIPTION
# columns (avoids duplicating non-transaction tables like DAILY BALANCES).
table_pattern2 = re.compile(r"]*>.*?", re.DOTALL | re.IGNORECASE)
has_txn_table = False
for tbl in table_pattern2.finditer(result):
p2 = TableGridParser()
p2.feed(tbl.group(0))
g2 = _build_grid(p2.rows)
if len(g2) < 2:
continue
hdr2 = [str(c or "").strip().upper() for c in g2[0]]
has_date = any(re.search(r"DATE", h) for h in hdr2)
has_desc = any(re.search(r"(DESCRIPTION|DETAILS?|NARRATION|PARTICULARS?)", h) for h in hdr2)
if has_date and has_desc:
has_txn_table = True
break
if not has_txn_table and tl_rows:
# Infer date format from text layer to pick column header label
_hy = re.compile(r"^\d{1,2}-\d{2}")
_sl = re.compile(r"^\d{1,2}/\d{2}")
_sample_dates = [_norm(r["date"]) for r in tl_rows[:3]]
if any(_hy.match(d) for d in _sample_dates):
date_label = "Date"
elif any(_sl.match(d) for d in _sample_dates):
date_label = "Date"
else:
date_label = "Trans Date"
has_post = any(r.get("post_date") for r in tl_rows)
if has_post:
header_row = [date_label, "Post Date", "Transaction Description", "Amount"]
else:
header_row = [date_label, "Transaction Description", "Amount"]
date_ci = 0
desc_ci = 2 if has_post else 1
amt_ci = 3 if has_post else 2
new_rows = []
for r in tl_rows:
row = [""] * len(header_row)
row[date_ci] = r["date"]
row[desc_ci] = r["desc"]
row[amt_ci] = r["amount"]
if has_post:
row[1] = r.get("post_date", "")
new_rows.append(row)
new_table_html = _grid_to_html([header_row] + new_rows)
result = new_table_html + "\n\n" + result
log.info("page %d: created new table from text layer (%d rows)", page_num, len(new_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 normalize_html_tables(text: str) -> str:
"""
For every :
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 data row instead of the | header row.
6. Fix 3 — split fused key-value rows in summary sections.
7. Emit normalised HTML.
"""
if not text or "]*>.*? ", 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)
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 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 = 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(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 page already has an OCR table — the table already
# captured this data correctly, so the footer text is redundant.
_footer_date_re = re.compile(r"\b\d{1,2}[-/]\d{2}\b")
is_txn_dump = (
len(_footer_date_re.findall(ftr_clean)) >= 3
and any(" |