app.py

#!/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 HTML table rewriting, text-layer row injection, institution-specific
    splits (UCB / Navy / TD / First Horizon / …), or doc-wide dedupe passes

Configure GLMOCR_API_KEY (environment variable). Optional: glmocr + gradio +
pymupdf + pillow installed.
"""

# Patch asyncio first (before Gradio imports it) to reduce Python 3.13 loop noise
import asyncio

try:
    _orig_close = asyncio.BaseEventLoop.close

    def _safe_close(self):
        try:
            _orig_close(self)
        except (ValueError, OSError):
            pass

    asyncio.BaseEventLoop.close = _safe_close
except Exception:
    pass

import html
import logging
import os
import re
import tempfile
from 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 = os.environ.get("GLMOCR_API_KEY", "").strip()
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 = 2.85

# 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.03
PAD_RIGHT_FRAC = 0.10
PAD_TOP_FRAC = 0.015
PAD_BOTTOM_FRAC = 0.015

ENABLE_CONTRAST = True
# Slight contrast lift only; same factor for every file.
CONTRAST_FACTOR = 1.12

# Subtle edge enhancement after contrast (helps hairlines and small digits).
ENABLE_UNSHARP = True
UNSHARP_RADIUS = 0.85
UNSHARP_PERCENT = 65
UNSHARP_THRESHOLD = 2

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

_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 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


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)
    return close_unclosed_html("\n\n".join(out_blocks))


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")
        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"
            "Upload a PDF or image. Header and footer bands are included; "
            "body OCR is passed through with only light markdown cleanup."
        )
        file_in = gr.File(
            label="Upload PDF or image",
            file_types=[".pdf", ".png", ".jpg", ".jpeg", ".tiff", ".bmp"],
        )
        run_btn = gr.Button("Run OCR", variant="primary")
        out = gr.Textbox(lines=40, label="Output (markdown / light HTML)")
        run_btn.click(fn=run_ocr, inputs=file_in, outputs=out)
    return demo


if __name__ == "__main__":
    _create_gradio_demo().launch()