Spaces:
Sleeping
Sleeping
| #!/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"</(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 | |
| _TR_OPEN = re.compile(r"<tr\b([^>]*)>", re.IGNORECASE) | |
| _TR_CLOSE = re.compile(r"</tr>", re.IGNORECASE) | |
| _CELL = re.compile( | |
| r"<(td|th)(\b[^>]*?)>((?:(?!</?(?:td|th)\b).)*?)</(td|th)\s*>", | |
| 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] + ["<td></td>", 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(["<td></td>"] * (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 <table>, 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 "<table" not in md.lower(): | |
| return md | |
| def repl_table(m: re.Match) -> str: | |
| full = m.group(0) | |
| low = full.lower() | |
| inner_start = low.find(">") + 1 | |
| inner_end = low.rfind("</table>") | |
| 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"<tr\b[^>]*>.*?</tr>", 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"<tr\b[^>]*>", seg, flags=re.IGNORECASE) | |
| cl = seg.lower().rfind("</tr>") | |
| 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"<tr\b[^>]*>", seg, flags=re.IGNORECASE) | |
| cl = seg.lower().rfind("</tr>") | |
| 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"<table\b[^>]*>.*?</table>", | |
| 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("<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 | |
| 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) | |