#!/usr/bin/env python3 """ Simplified GLM-OCR Hugging Face / local Gradio app. Scope (intentionally small): - PDF → padded high-DPI page images → GLM-OCR body markdown - Header band: PDF text extraction first, optional header OCR fallback - Footer band: same pattern, with light dedup so we do not paste a full transaction dump twice when the body already captured it Universal image pipeline (same for every PDF, no keywords / no bank logic): - Higher rasterization scale + extra white padding so fine print, boxed section labels, and right-aligned amounts sit farther from the clip edge. - Mild contrast + unsharp mask on every raster sent to the model so thin rules and small glyphs are easier to read before recognition. Explicitly omitted vs the heavy Space build: - No text-layer row injection, institution-specific splits (UCB / Navy / TD / First Horizon / …), or doc-wide dedupe passes. Included (data-driven, no institution names): - HTML tables: modal logical width from rowspan-free rows (colspan-aware); pad short rows; trim trailing empty cells; if last cell is empty and the previous cell is only a currency token, move that token to the last column. Configure GLMOCR_API_KEY (environment variable). Optional: glmocr + gradio + pymupdf + pillow installed. When GLMOCR_PREFER_PDF_TEXT_BODY is 1 (default), searchable PDFs skip vision OCR for the body and return each page's embedded text so output matches the PDF text layer. Set GLMOCR_PREFER_PDF_TEXT_BODY=0 to force the image OCR path. Optional GLMOCR_MIN_PDF_BODY_CHARS (default 120) is the minimum characters per page required to use the text-layer path for the whole document. """ # Patch asyncio first (before Gradio imports it) to reduce Python 3.13 loop noise import asyncio try: _orig_close = asyncio.BaseEventLoop.close def _safe_close(self): try: _orig_close(self) except (ValueError, OSError): pass asyncio.BaseEventLoop.close = _safe_close except Exception: pass import html import logging import os import re import tempfile from collections import Counter from typing import List, Optional, Tuple import yaml try: import glmocr GLMOCR_BASE = os.path.dirname(glmocr.__file__) CONFIG_PATH = os.path.join(GLMOCR_BASE, "config.yaml") except ImportError: glmocr = None # type: ignore GLMOCR_BASE = "" CONFIG_PATH = "" log = logging.getLogger("glmocr_simple_app") logging.basicConfig(level=logging.INFO) # --------------------------------------------------------------------------- # Settings — tuned for dense financial PDFs; applies to every document # --------------------------------------------------------------------------- GLMOCR_API_KEY = "cee1d52dd91a4ab591b3f6e105f8ad89.LgbQTECuzX0zrito" if not GLMOCR_API_KEY: log.warning("GLMOCR_API_KEY is not set; GlmOcr() will fail until you export it.") # Rasterization: higher scale = more pixels per PDF point (helps small type, # boxed headers, and narrow columns). Same constant for all uploads. RENDER_SCALE = 3.05 # White margin as a fraction of page width/height after render. Extra right # margin helps right-aligned currency columns that hug the page edge. PAD_LEFT_FRAC = 0.035 PAD_RIGHT_FRAC = 0.11 PAD_TOP_FRAC = 0.018 PAD_BOTTOM_FRAC = 0.018 ENABLE_CONTRAST = True # Slight contrast lift only; same factor for every file. CONTRAST_FACTOR = 1.16 # Subtle edge enhancement after contrast (helps hairlines and small digits). ENABLE_UNSHARP = True UNSHARP_RADIUS = 0.78 UNSHARP_PERCENT = 72 UNSHARP_THRESHOLD = 1 DEFAULT_ZONE_FRAC = 0.12 PDF_HEADER_BAND_FRAC = 0.10 ENABLE_FOOTER_OCR = True PDF_FOOTER_BAND_FRAC = 0.88 MIN_CROP_HEIGHT = 112 MIN_CROP_PIXELS = 112 * 112 # PNG compression 0–9; lower = less loss before GLM-OCR (same for all PDFs). PAGE_PNG_COMPRESS_LEVEL = 3 # JPEG quality for small header/footer crops sent to the API. ZONE_JPEG_QUALITY = 95 MIN_PDF_TEXT_CHARS_NATIVE_LAYER = 1500 _parser = None def _enhance_raster_for_ocr(img): """ Improve legibility of every raster passed to GLM-OCR (full pages and header/footer crops). No document text or keywords — same pipeline for all PDFs and images. """ from PIL import ImageEnhance, ImageFilter if ENABLE_CONTRAST: img = ImageEnhance.Contrast(img).enhance(CONTRAST_FACTOR) if ENABLE_UNSHARP: img = img.filter( ImageFilter.UnsharpMask( radius=UNSHARP_RADIUS, percent=UNSHARP_PERCENT, threshold=UNSHARP_THRESHOLD, ) ) return img def get_parser(): global _parser if glmocr is None: raise RuntimeError("glmocr is not installed.") if _parser is None: from glmocr import GlmOcr _parser = GlmOcr(api_key=GLMOCR_API_KEY, mode="maas") return _parser if CONFIG_PATH: try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: config = yaml.safe_load(f) config.setdefault("pipeline", {}).setdefault("maas", {}) config["pipeline"]["maas"]["enabled"] = True config["pipeline"]["maas"]["api_key"] = GLMOCR_API_KEY with open(CONFIG_PATH, "w", encoding="utf-8") as f: yaml.dump(config, f, default_flow_style=False, sort_keys=False) except Exception: pass def get_header_footer_zones(regions, norm_height=1000): if not regions: return None, None y_tops, y_bottoms = [], [] for r in regions: bbox = r.get("bbox_2d") if isinstance(r, dict) else getattr(r, "bbox_2d", None) if bbox and len(bbox) >= 4: y_tops.append(bbox[1]) y_bottoms.append(bbox[3]) if not y_tops: return None, None return min(y_tops) / norm_height, max(y_bottoms) / norm_height def extract_zone_text_pdf(pdf_path, page_num, y_start_frac, y_end_frac): try: import pymupdf as fitz doc = fitz.open(pdf_path) page = doc[page_num] h, w = page.rect.height, page.rect.width rect = fitz.Rect(0, h * y_start_frac, w, h * y_end_frac) text = page.get_text(clip=rect).strip() doc.close() return text except Exception: return "" def _prefer_pdf_text_body() -> bool: v = os.environ.get("GLMOCR_PREFER_PDF_TEXT_BODY", "1").strip().lower() return v in ("1", "true", "yes", "") def extract_pdf_body_text_all_pages_if_suitable(pdf_path: str) -> Optional[List[str]]: try: import pymupdf as fitz except ImportError: return None try: min_c = int(os.environ.get("GLMOCR_MIN_PDF_BODY_CHARS", "120")) except ValueError: min_c = 120 try: doc = fitz.open(pdf_path) except Exception: return None try: if len(doc) < 1: return None out: List[str] = [] for i in range(len(doc)): t = (doc[i].get_text("text") or "").strip() if len(t) < min_c: return None out.append(t) return out finally: try: doc.close() except Exception: pass def extract_pdf_text_in_band(pdf_path, page_num, y_start_frac, y_end_frac): try: import pymupdf as fitz doc = fitz.open(pdf_path) page = doc[page_num] h = page.rect.height y_lo = h * y_start_frac y_hi = h * y_end_frac words = page.get_text("words") doc.close() parts = [] for w in words: if len(w) >= 5: y0, y1 = float(w[1]), float(w[3]) if y0 < y_hi and y1 > y_lo: parts.append(w[4]) return " ".join(parts).strip() except Exception: return "" def ocr_zone(image_path, y_start_frac, y_end_frac): zone_name = "header" if y_end_frac < 0.5 else "footer" try: from PIL import Image img = Image.open(image_path).convert("RGB") w, h = img.size y0 = max(0, int(h * y_start_frac)) y1 = min(h, int(h * y_end_frac)) if y1 <= y0: return "" crop = img.crop((0, y0, w, y1)) cw, ch = crop.size if ch < MIN_CROP_HEIGHT or (cw * ch) < MIN_CROP_PIXELS: need_h = max(ch, MIN_CROP_HEIGHT) need_w = max(cw, 1) if (need_w * need_h) < MIN_CROP_PIXELS: need_w = max(need_w, (MIN_CROP_PIXELS + need_h - 1) // need_h) canvas = Image.new("RGB", (need_w, need_h), (255, 255, 255)) if zone_name == "header": canvas.paste(crop, (0, 0)) else: canvas.paste(crop, (0, need_h - ch)) crop = canvas fd, path = tempfile.mkstemp(suffix=".jpg") os.close(fd) try: crop.save(path, "JPEG", quality=ZONE_JPEG_QUALITY) parser = get_parser() out = parser.parse(path) if not isinstance(out, list): out = [out] if out and getattr(out[0], "markdown_result", None): return (out[0].markdown_result or "").strip() finally: try: os.unlink(path) except Exception: pass except Exception as e: log.warning("[%s] ocr_zone failed: %s", zone_name, e, exc_info=True) return "" def fix_account_number(hdr: str) -> str: if not hdr: return hdr if "Account Number:" in hdr and "Account Number: " not in hdr: m = re.search(r"[0-9]{5,}", hdr) if m: hdr = hdr.replace("Account Number:", "Account Number: " + m.group(0)) acct_match = re.search(r"Account Number: ([0-9]{5,})", hdr) if acct_match: acct = acct_match.group(1) if hdr.startswith(acct): hdr = hdr[len(acct) :].lstrip() return hdr def close_unclosed_html(md: str) -> str: if not md: return md open_tags = re.findall(r"<(table|tbody|thead|tr|td|th)\b", md, flags=re.IGNORECASE) close_tags = re.findall(r"", md, flags=re.IGNORECASE) def count(tags, name): return sum(1 for t in tags if t.lower() == name) for tag in reversed(["td", "th", "tr", "thead", "tbody", "table"]): opened = count(open_tags, tag) closed = count(close_tags, tag) if opened > closed: md += ("" % tag) * (opened - closed) return md _TR_OPEN = re.compile(r"]*)>", re.IGNORECASE) _TR_CLOSE = re.compile(r"", re.IGNORECASE) _CELL = re.compile( r"<(td|th)(\b[^>]*?)>((?:(?!", re.IGNORECASE | re.DOTALL, ) def _cell_entries(tr_inner: str) -> List[Tuple[str, int]]: """(full_cell_html, logical_width) for each td/th; 0 cells if unparseable.""" out: List[Tuple[str, int]] = [] for m in _CELL.finditer(tr_inner): open_name, attrs, _body, close_name = m.group(1), m.group(2), m.group(3), m.group(4) if open_name.lower() != close_name.lower(): continue cm = re.search(r"colspan\s*=\s*[\"']?(\d+)", attrs, flags=re.IGNORECASE) span = int(cm.group(1)) if cm else 1 span = max(1, span) out.append((m.group(0), span)) return out def _logical_row_width(entries: List[Tuple[str, int]]) -> int: return sum(s for _f, s in entries) def _cell_text_empty(full_cell: str) -> bool: m = _CELL.fullmatch(full_cell.strip()) if not m: inner = re.sub(r"<[^>]+>", " ", full_cell) else: inner = m.group(3) inner = re.sub(r"\s+", " ", inner).strip() inner = html.unescape(inner) return inner == "" def _cell_plain_text(full_cell: str) -> str: """Visible text of one td/th, no tags.""" m = _CELL.fullmatch(full_cell.strip()) if not m: t = re.sub(r"<[^>]+>", " ", full_cell) else: t = m.group(3) t = html.unescape(re.sub(r"\s+", " ", t).strip()) return t def _is_whole_cell_currency(text: str) -> bool: """ True iff the cell is nothing but a currency-looking amount (optional $, commas, 2 decimals). Excludes dates (slashes) and arbitrary prose — not keyed to column headers. """ if not text or "/" in text: return False return bool( re.fullmatch( r"-?(?:\$|€|£)?\s*\d{1,3}(?:,\d{3})*\.\d{2}\s*", text, ) or re.fullmatch(r"-?(?:\$|€|£)?\s*\d+\.\d{2}\s*", text) ) def _realign_money_if_last_cell_empty(cells: List[str]) -> List[str]: """ … | X | empty -> … | empty | X when X is currency-only (fixes amount parked one column left of an empty trailing cell, including after width padding). """ if len(cells) < 2: return cells if not _cell_text_empty(cells[-1]): return cells if not _is_whole_cell_currency(_cell_plain_text(cells[-2])): return cells return cells[:-2] + ["", cells[-2]] def _infer_modal_logical_width(tr_inners: List[str]) -> int: """ Modal logical column count across rows (colspan sums). On frequency ties, prefer the larger width so a rare short row is padded to the majority grid. Returns -1 if the table uses rowspan (skip) or has no measurable rows. """ widths: List[int] = [] for inner in tr_inners: if re.search(r"rowspan\s*=", inner, flags=re.IGNORECASE): return -1 w = _logical_row_width(_cell_entries(inner)) if w > 0: widths.append(w) if not widths: return -1 c = Counter(widths) best = max(c.values()) candidates = [w for w, n in c.items() if n == best] return max(candidates) def _normalize_one_tr_inner(tr_inner: str, target: int) -> str: entries = _cell_entries(tr_inner) if not entries: return tr_inner cells = [e[0] for e in entries] spans = [e[1] for e in entries] w = sum(spans) if w < target: cells.extend([""] * (target - w)) spans.extend([1] * (target - w)) w = target while w > target and cells: if spans[-1] != 1 or not _cell_text_empty(cells[-1]): break w -= spans[-1] cells.pop() spans.pop() if cells and all(s == 1 for s in spans): cells = _realign_money_if_last_cell_empty(cells) return "".join(cells) def normalize_html_table_row_widths(md: str) -> str: """ For each , infer the dominant logical column count from rowspan-free rows (colspan-aware), then pad rows that are too narrow or strip trailing empty single-colspan cells from rows that are too wide. No column names or fixed N: width comes from per-table row statistics. If the last cell is empty and the previous cell is only a currency token, that amount is moved into the last column (whole-cell shape, not headers). Tables with rowspan are skipped. Same-width wrong text that is not currency-shaped is not altered. """ if not md or " str: full = m.group(0) low = full.lower() inner_start = low.find(">") + 1 inner_end = low.rfind("
") if inner_start <= 0 or inner_end < inner_start: return full prefix = full[:inner_start] body = full[inner_start:inner_end] suffix = full[inner_end:] tr_blocks = list(re.finditer(r"]*>.*?", body, flags=re.IGNORECASE | re.DOTALL)) if not tr_blocks: return full tr_inners: List[str] = [] for tm in tr_blocks: seg = tm.group(0) op = re.search(r"]*>", seg, flags=re.IGNORECASE) cl = seg.lower().rfind("") if not op or cl < 0: continue tr_inners.append(seg[op.end() : cl]) target = _infer_modal_logical_width(tr_inners) if target < 1: return full new_parts: List[str] = [] last_end = 0 for tm in tr_blocks: new_parts.append(body[last_end : tm.start()]) seg = tm.group(0) op = re.search(r"]*>", seg, flags=re.IGNORECASE) cl = seg.lower().rfind("") if not op or cl < 0: new_parts.append(seg) else: open_tr = seg[: op.end()] inner = seg[op.end() : cl] close_tr = seg[cl:] if re.search(r"rowspan\s*=", inner, flags=re.IGNORECASE): new_parts.append(seg) else: new_parts.append(open_tr + _normalize_one_tr_inner(inner, target) + close_tr) last_end = tm.end() new_parts.append(body[last_end:]) return prefix + "".join(new_parts) + suffix return re.sub( r"]*>.*?", repl_table, md, flags=re.IGNORECASE | re.DOTALL, ) def looks_like_markdown_table(block: str) -> bool: lines = [ln.rstrip() for ln in block.strip().splitlines() if ln.strip()] if len(lines) < 2: return False if "|" not in lines[0]: return False sep = lines[1].replace(" ", "") return ("---" in sep) and ("|" in sep) def md_table_to_html(block: str) -> str: lines = [ln.strip() for ln in block.strip().splitlines() if ln.strip()] if len(lines) < 2: return block def split_row(row: str): row = row.strip() if row.startswith("|"): row = row[1:] if row.endswith("|"): row = row[:-1] return [p.strip() for p in row.split("|")] header = split_row(lines[0]) body_lines = [ln for ln in lines[2:] if "|" in ln] html_rows = [] html_rows.append("" + "".join(f"{html.escape(c)}" for c in header) + "") for ln in body_lines: cols = split_row(ln) if len(cols) < len(header): cols += [""] * (len(header) - len(cols)) html_rows.append( "" + "".join(f"{html.escape(c)}" for c in cols[: len(header)]) + "" ) return "\n" + "\n".join(html_rows) + "\n
" def normalize_money_glyphs(text: str) -> str: if not text: return text t = text.replace("−", "-").replace("–", "-").replace("—", "-") t = re.sub( r"\(\s*\$?\s*([0-9]{1,3}(?:,[0-9]{3})*|[0-9]+)(\.[0-9]{2})\s*\)", r"-\1\2", t, ) def o_to_zero(m): token = m.group(0) return token.replace("O", "0").replace("o", "0") t = re.sub(r"\b[0-9Oo\$,.\-]{4,}\b", o_to_zero, t) return t def light_stabilize_markdown(page_md: str) -> str: """Convert obvious GitHub-style pipe tables to HTML; normalize money glyphs; repair tags.""" if not page_md: return page_md page_md = normalize_money_glyphs(page_md) blocks = re.split(r"\n\s*\n", page_md.strip()) out_blocks = [] for b in blocks: if looks_like_markdown_table(b): out_blocks.append(md_table_to_html(b)) else: out_blocks.append(b) merged = close_unclosed_html("\n\n".join(out_blocks)) return normalize_html_table_row_widths(merged) def render_pdf_pages_to_images(pdf_path: str) -> Tuple[List[str], List[int]]: import pymupdf as fitz from PIL import Image doc = fitz.open(pdf_path) page_images: List[str] = [] page_heights: List[int] = [] for i in range(len(doc)): page = doc[i] pix = page.get_pixmap(matrix=fitz.Matrix(RENDER_SCALE, RENDER_SCALE), alpha=False) img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) img = _enhance_raster_for_ocr(img) w, h = img.size pad_l = int(w * PAD_LEFT_FRAC) pad_r = int(w * PAD_RIGHT_FRAC) pad_t = int(h * PAD_TOP_FRAC) pad_b = int(h * PAD_BOTTOM_FRAC) if any(p > 0 for p in (pad_l, pad_r, pad_t, pad_b)): canvas = Image.new("RGB", (w + pad_l + pad_r, h + pad_t + pad_b), (255, 255, 255)) canvas.paste(img, (pad_l, pad_t)) img = canvas img_path = os.path.join(tempfile.gettempdir(), f"glmocr_page_{os.getpid()}_{i}.png") img.save(img_path, "PNG", compress_level=PAGE_PNG_COMPRESS_LEVEL) page_images.append(img_path) page_heights.append(img.height) doc.close() return page_images, page_heights def get_page_md_and_regions(page_result): md = "" if hasattr(page_result, "markdown_result") and page_result.markdown_result: md = (page_result.markdown_result or "").strip() regions = [] if hasattr(page_result, "json_result"): jr = page_result.json_result if isinstance(jr, dict) and "regions" in jr: regions = jr.get("regions") or [] elif isinstance(jr, list) and len(jr) > 0: r = jr[0] if isinstance(jr[0], list) else jr if isinstance(r, list): regions = r elif isinstance(r, dict) and "regions" in r: regions = r.get("regions") or [] return md, regions def run_ocr(uploaded_file): if uploaded_file is None: return "Please upload a file." page_images: List[str] = [] try: path = uploaded_file.name if hasattr(uploaded_file, "name") else str(uploaded_file) is_pdf = path.lower().endswith(".pdf") if is_pdf and _prefer_pdf_text_body(): pdf_pages = extract_pdf_body_text_all_pages_if_suitable(path) if pdf_pages is not None: return "\n\n---page-separator---\n\n".join(pdf_pages) parser = get_parser() page_heights: List[int] = [] if is_pdf: page_images, page_heights = render_pdf_pages_to_images(path) results = parser.parse(page_images) else: page_images = [path] page_heights = [1000] results = parser.parse(path) if not isinstance(results, list): results = [results] all_pages = [] for page_num, page_result in enumerate(results): page_md, regions = get_page_md_and_regions(page_result) img_h = page_heights[page_num] if page_num < len(page_heights) else 1000 header_end_frac, footer_start_frac = get_header_footer_zones(regions, img_h) he = header_end_frac if header_end_frac is not None else DEFAULT_ZONE_FRAC fs = footer_start_frac if footer_start_frac is not None else (1.0 - DEFAULT_ZONE_FRAC) he = max(0.02, min(0.25, he)) fs = max(0.75, min(0.98, fs)) parts = [] hdr = "" if is_pdf: hdr = extract_zone_text_pdf(path, page_num, 0, he) if not (hdr and hdr.strip()): hdr = extract_pdf_text_in_band(path, page_num, 0, PDF_HEADER_BAND_FRAC) if not (hdr and hdr.strip()) and page_num < len(page_images): hdr = ocr_zone(page_images[page_num], 0, he) if hdr and hdr.strip(): parts.append(light_stabilize_markdown(fix_account_number(normalize_money_glyphs(hdr.strip())))) if page_md and page_md.strip(): parts.append(light_stabilize_markdown(page_md.strip())) if ENABLE_FOOTER_OCR and page_num < len(page_images): ftr = "" if is_pdf: ftr = extract_zone_text_pdf(path, page_num, fs, 1.0) if not (ftr and ftr.strip()): ftr = extract_pdf_text_in_band(path, page_num, PDF_FOOTER_BAND_FRAC, 1.0) if not (ftr and ftr.strip()): ftr = ocr_zone(page_images[page_num], fs, 1.0) if ftr and ftr.strip(): ftr_clean = normalize_money_glyphs(ftr.strip()) ftr_first_line = next( (ln.strip().lower() for ln in ftr_clean.splitlines() if ln.strip()), "", ) already_present = ftr_first_line and any( ftr_first_line in part.lower() for part in parts ) _footer_date_re = re.compile(r"\b\d{1,2}[-/]\d{2}\b") _footer_amt_re = re.compile(r"\b\d{1,3}(?:,\d{3})*\.\d{2}\b") _date_hits = len(_footer_date_re.findall(ftr_clean)) _amt_hits = len(_footer_amt_re.findall(ftr_clean)) is_txn_dump = _date_hits >= 3 and _amt_hits >= 3 if not already_present and not is_txn_dump: parts.append(ftr_clean) if parts: all_pages.append("\n\n".join(parts)) merged = "\n\n---page-separator---\n\n".join(all_pages) if all_pages else "(No content)" return merged except Exception as e: import traceback log.exception("run_ocr failed: %s", e) return f"Error: {e}\n\n{traceback.format_exc()}" finally: for p in page_images: try: if isinstance(p, str) and p.endswith(".png") and "glmocr_page_" in os.path.basename(p): os.unlink(p) except Exception: pass def _create_gradio_demo(): import gradio as gr with gr.Blocks(title="GLM-OCR (simple)") as demo: gr.Markdown( "# GLM-OCR (simple)\n" "Searchable PDFs use embedded page text by default (matches the PDF text layer). " "Otherwise the vision model reads rasterized pages. Images always use vision OCR." ) file_in = gr.File( label="Upload PDF or image", file_types=[".pdf", ".png", ".jpg", ".jpeg", ".tiff", ".bmp"], ) run_btn = gr.Button("Run OCR", variant="primary") out = gr.Textbox(lines=40, label="Output (markdown / light HTML)") run_btn.click(fn=run_ocr, inputs=file_in, outputs=out) return demo if __name__ == "__main__": _create_gradio_demo().launch(share=True)