Add OCR notification/data extractor Gradio app for Hugging Face Spaces

Takes PDF or image uploads, OCRs pages (native PDF text or Tesseract for
scans), flags domain notification keywords (construction, compliance,
finance), extracts structured fields (PO/SAP/invoice numbers, dates,
amounts, cert codes) via regex, and exports results to CSV.

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+.venv/
+*.csv
+!sample_data/*.csv
+.DS_Store

README.md

@@ -0,0 +1,53 @@
+---
+title: OCR Notification & Data Extractor
+emoji: 🧾
+colorFrom: blue
+colorTo: indigo
+sdk: gradio
+sdk_version: 4.44.0
+app_file: app.py
+pinned: false
+license: mit
+---
+
+# OCR Notification & Data Extractor
+
+Upload scanned PDFs or images and turn them into structured, exportable
+data. Built for teams that need to pull actionable "notification" events
+and key fields out of paper-heavy workflows:
+
+- **Construction / SAP data support** — change orders, delay notices,
+  punch list items, RFIs, PO numbers, SAP document numbers, quantities.
+- **Compliance** — non-conformances, violations, expirations, corrective
+  actions, certification codes (ISO, OSHA, ASTM, ANSI).
+- **Business finance** — past-due invoices, payment rejections, credit
+  holds, invoice numbers, dollar amounts, due dates.
+
+## How it works
+
+1. Upload one or more PDFs or images.
+2. Each page's text is pulled directly from the PDF when it has a text
+   layer; scanned/image-only pages are rendered and run through
+   Tesseract OCR.
+3. The text is scanned against keyword sets for the categories you pick
+   (or your own custom keywords), and structured fields (PO/SAP/invoice
+   numbers, dates, dollar amounts, percentages, certification codes) are
+   pulled out with regex.
+4. Results are shown in a table and exported as a CSV, one row per page
+   that has a hit (or every page, if you turn that filter off).
+
+## Local development
+
+```bash
+pip install -r requirements.txt
+# Tesseract must also be installed on the system, e.g.:
+#   apt-get install tesseract-ocr
+python app.py
+```
+
+## Files
+
+- `app.py` — Gradio UI and orchestration.
+- `ocr_engine.py` — PDF/image loading and OCR (PyMuPDF + Tesseract).
+- `extractors.py` — keyword notification scanning and structured field
+  extraction (regex-based, easy to extend per domain).

app.py

@@ -0,0 +1,155 @@
+"""Gradio app: OCR documents into structured notification/data records.
+
+Upload scanned PDFs or images (construction reports, compliance
+certificates, invoices, etc.), scan the extracted text for domain
+"notification" keywords, pull out structured fields (PO/SAP/invoice
+numbers, dates, amounts, cert codes), and export everything to CSV.
+"""
+import os
+import tempfile
+from datetime import datetime
+
+import gradio as gr
+import pandas as pd
+
+import extractors
+import ocr_engine
+
+CATEGORY_CHOICES = ["construction", "compliance", "finance", "general"]
+
+COLUMNS = [
+    "file_name", "page", "text_source", "ocr_confidence",
+    "notification_flags", "notification_context",
+    "po_numbers", "sap_doc_numbers", "invoice_numbers", "cert_codes",
+    "dates", "amounts", "percentages",
+]
+
+
+def _file_path(f):
+    return f if isinstance(f, str) else f.name
+
+
+def process_files(files, categories, custom_keywords_str, only_matches):
+    if not files:
+        return "Upload at least one PDF or image file.", pd.DataFrame(columns=COLUMNS), None, ""
+
+    custom_keywords = None
+    if custom_keywords_str and custom_keywords_str.strip():
+        custom_keywords = [k.strip() for k in custom_keywords_str.split(",") if k.strip()]
+
+    rows = []
+    full_text_log = []
+    warnings = []
+    total_pages = 0
+
+    for f in files:
+        path = _file_path(f)
+        file_name = os.path.basename(path)
+        try:
+            pages = ocr_engine.extract_pages(path)
+        except Exception as exc:
+            warnings.append(f"{file_name}: {exc}")
+            continue
+
+        for page_data in pages:
+            total_pages += 1
+            text = page_data["text"]
+            notifications = extractors.find_notifications(text, categories, custom_keywords)
+            fields = extractors.extract_fields(text)
+            has_content = bool(notifications) or any(fields.values())
+
+            full_text_log.append(
+                f"--- {file_name} | page {page_data['page']} | {page_data['source']} ---\n{text}\n"
+            )
+
+            if only_matches and not has_content:
+                continue
+
+            contexts = []
+            for n in notifications:
+                if n["context"] not in contexts:
+                    contexts.append(n["context"])
+
+            rows.append({
+                "file_name": file_name,
+                "page": page_data["page"],
+                "text_source": page_data["source"],
+                "ocr_confidence": page_data["ocr_confidence"],
+                "notification_flags": "; ".join(
+                    sorted({f"[{n['category']}] {n['keyword']}" for n in notifications})
+                ),
+                "notification_context": " | ".join(contexts[:5]),
+                "po_numbers": "; ".join(fields["po_numbers"]),
+                "sap_doc_numbers": "; ".join(fields["sap_doc_numbers"]),
+                "invoice_numbers": "; ".join(fields["invoice_numbers"]),
+                "cert_codes": "; ".join(fields["cert_codes"]),
+                "dates": "; ".join(fields["dates"]),
+                "amounts": "; ".join(fields["amounts"]),
+                "percentages": "; ".join(fields["percentages"]),
+            })
+
+    df = pd.DataFrame(rows, columns=COLUMNS)
+
+    csv_path = None
+    if not df.empty:
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        csv_path = os.path.join(tempfile.gettempdir(), f"ocr_extract_{timestamp}.csv")
+        df.to_csv(csv_path, index=False)
+
+    status_lines = [
+        f"Processed {len(files)} file(s), {total_pages} page(s) total.",
+        f"{len(df)} row(s) exported" + (" (matches only)." if only_matches else "."),
+    ]
+    if warnings:
+        status_lines.append("Warnings: " + "; ".join(warnings))
+    status = "\n".join(status_lines)
+
+    return status, df, csv_path, "\n".join(full_text_log)
+
+
+with gr.Blocks(title="OCR Notification & Data Extractor") as demo:
+    gr.Markdown(
+        "# OCR Notification & Data Extractor\n"
+        "Upload scanned PDFs or images (construction reports, compliance "
+        "certificates, invoices, SAP printouts, etc.). The app OCRs each "
+        "page, flags notification-worthy keywords per domain, extracts "
+        "structured fields (PO/SAP/invoice numbers, dates, amounts, cert "
+        "codes), and exports everything to CSV."
+    )
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            files_input = gr.Files(
+                label="PDF or image files",
+                file_types=[".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp"],
+            )
+            categories_input = gr.CheckboxGroup(
+                choices=CATEGORY_CHOICES,
+                value=CATEGORY_CHOICES,
+                label="Notification categories",
+            )
+            custom_keywords_input = gr.Textbox(
+                label="Custom keywords (comma-separated, optional)",
+                placeholder="e.g. re-inspection, warranty claim, hold payment",
+            )
+            only_matches_input = gr.Checkbox(
+                value=True,
+                label="Only export pages with a notification hit or extracted field",
+            )
+            run_button = gr.Button("Run OCR & Extract", variant="primary")
+
+        with gr.Column(scale=2):
+            status_output = gr.Textbox(label="Status", lines=3, interactive=False)
+            table_output = gr.Dataframe(label="Extracted records", headers=COLUMNS, wrap=True)
+            csv_output = gr.File(label="Download CSV")
+            with gr.Accordion("Full OCR text (debug)", open=False):
+                text_output = gr.Textbox(label="Raw extracted text", lines=20)
+
+    run_button.click(
+        fn=process_files,
+        inputs=[files_input, categories_input, custom_keywords_input, only_matches_input],
+        outputs=[status_output, table_output, csv_output, text_output],
+    )
+
+if __name__ == "__main__":
+    demo.launch()

extractors.py

@@ -0,0 +1,123 @@
+"""Keyword and structured-field extraction for OCR'd document text.
+
+Domain keyword sets flag "notification" events (things a reviewer would want
+alerted on), and field patterns pull structured values (IDs, dates, amounts)
+out of the surrounding text so results can be dropped straight into a CSV.
+"""
+import re
+
+KEYWORD_SETS = {
+    "construction": [
+        "change order", "delay notice", "punch list", "non-conformance",
+        "rfi", "submittal rejected", "back charge", "backcharge",
+        "stop work", "safety violation", "material shortage",
+        "schedule slip", "cost overrun", "field order", "notice to proceed",
+    ],
+    "compliance": [
+        "non-compliant", "noncompliance", "violation", "deficiency",
+        "corrective action", "audit finding", "expired", "expiration",
+        "recall", "citation", "penalty", "out of compliance",
+        "failed inspection", "not approved",
+    ],
+    "finance": [
+        "past due", "overdue", "payment rejected", "invoice disputed",
+        "credit hold", "chargeback", "insufficient funds", "write-off",
+        "collections", "declined", "late fee", "short paid",
+    ],
+    "general": [
+        "urgent", "immediate action required", "cancelled", "rejected",
+        "approved", "pending approval", "revised", "on hold",
+    ],
+}
+
+FIELD_PATTERNS = {
+    "dates": re.compile(
+        r"\b(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{4}-\d{2}-\d{2}|"
+        r"(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*\.?\s+"
+        r"\d{1,2},?\s+\d{4})\b",
+        re.IGNORECASE,
+    ),
+    "amounts": re.compile(
+        r"\$\s?\d[\d,]*\.?\d{0,2}\b|\bUSD\s?\d[\d,]*\.?\d{0,2}\b",
+        re.IGNORECASE,
+    ),
+    "percentages": re.compile(r"\b\d{1,3}(?:\.\d+)?\s?%"),
+    "po_numbers": re.compile(
+        r"\bP\.?\s?O\.?[ \t]*#?[ \t]*[:\-]?[ \t]*((?=[A-Z0-9-]*\d)[A-Z0-9\-]{3,15})\b",
+        re.IGNORECASE,
+    ),
+    "sap_doc_numbers": re.compile(
+        r"\bSAP[ \t]*(?:Doc(?:ument)?\.?|#)?[ \t]*[:\-]?[ \t]*(\d{8,10})\b|\b(4[5-8]\d{8})\b",
+        re.IGNORECASE,
+    ),
+    "invoice_numbers": re.compile(
+        r"\bINV(?:OICE)?\.?[ \t]*#?[ \t]*[:\-]?[ \t]*((?=[A-Z0-9-]*\d)[A-Z0-9\-]{3,15})\b",
+        re.IGNORECASE,
+    ),
+    "cert_codes": re.compile(
+        r"\b(ISO\s?\d{3,5}(?:-\d+)?|OSHA\s?\d{3,4}(?:\.\d+)?|"
+        r"ASTM\s?[A-Z]?\d{2,4}(?:-\d{2})?|ANSI\s?[A-Z]?\d+(?:\.\d+)?)\b",
+        re.IGNORECASE,
+    ),
+}
+
+CONTEXT_WINDOW = 60
+
+
+def _dedupe(values):
+    seen = []
+    for v in values:
+        v = v.strip()
+        if v and v not in seen:
+            seen.append(v)
+    return seen
+
+
+def extract_fields(text):
+    """Pull structured fields out of a page/block of text."""
+    fields = {}
+    for name, pattern in FIELD_PATTERNS.items():
+        matches = pattern.findall(text)
+        flat = []
+        for m in matches:
+            if isinstance(m, tuple):
+                flat.extend(g for g in m if g)
+            else:
+                flat.append(m)
+        fields[name] = _dedupe(flat)
+    return fields
+
+
+def find_notifications(text, categories, custom_keywords=None):
+    """Scan text for keyword hits across the given categories.
+
+    Returns a list of dicts: {category, keyword, context}.
+    """
+    hits = []
+    keywords_by_category = {}
+    for cat in categories:
+        keywords_by_category[cat] = list(KEYWORD_SETS.get(cat, []))
+    if custom_keywords:
+        keywords_by_category["custom"] = list(custom_keywords)
+
+    lower_text = text.lower()
+    for category, keywords in keywords_by_category.items():
+        for kw in keywords:
+            kw_clean = kw.strip().lower()
+            if not kw_clean:
+                continue
+            start = 0
+            while True:
+                idx = lower_text.find(kw_clean, start)
+                if idx == -1:
+                    break
+                ctx_start = max(0, idx - CONTEXT_WINDOW)
+                ctx_end = min(len(text), idx + len(kw_clean) + CONTEXT_WINDOW)
+                context = text[ctx_start:ctx_end].replace("\n", " ").strip()
+                hits.append({
+                    "category": category,
+                    "keyword": kw.strip(),
+                    "context": context,
+                })
+                start = idx + len(kw_clean)
+    return hits

ocr_engine.py

@@ -0,0 +1,73 @@
+"""PDF/image loading and OCR.
+
+Native (embedded) text is pulled straight out of the PDF when present;
+pages without a usable text layer (scanned documents) are rendered to an
+image and run through Tesseract instead.
+"""
+import os
+
+import fitz  # PyMuPDF
+import pytesseract
+from PIL import Image
+
+MIN_EMBEDDED_TEXT_LEN = 20
+RENDER_DPI = 300
+IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp"}
+
+
+def _ocr_image(img):
+    data = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT)
+    confidences = [int(c) for c in data.get("conf", []) if str(c).isdigit() and int(c) >= 0]
+    mean_conf = round(sum(confidences) / len(confidences), 1) if confidences else None
+    text = pytesseract.image_to_string(img)
+    return text, mean_conf
+
+
+def _process_pdf(file_path):
+    pages = []
+    doc = fitz.open(file_path)
+    try:
+        for page_num, page in enumerate(doc, start=1):
+            embedded_text = page.get_text().strip()
+            if len(embedded_text) >= MIN_EMBEDDED_TEXT_LEN:
+                pages.append({
+                    "page": page_num,
+                    "text": embedded_text,
+                    "source": "embedded",
+                    "ocr_confidence": None,
+                })
+                continue
+
+            pix = page.get_pixmap(dpi=RENDER_DPI)
+            img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+            ocr_text, mean_conf = _ocr_image(img)
+            pages.append({
+                "page": page_num,
+                "text": ocr_text.strip(),
+                "source": "ocr",
+                "ocr_confidence": mean_conf,
+            })
+    finally:
+        doc.close()
+    return pages
+
+
+def _process_image(file_path):
+    img = Image.open(file_path).convert("RGB")
+    text, mean_conf = _ocr_image(img)
+    return [{
+        "page": 1,
+        "text": text.strip(),
+        "source": "ocr",
+        "ocr_confidence": mean_conf,
+    }]
+
+
+def extract_pages(file_path):
+    """Return a list of {page, text, source, ocr_confidence} dicts for a file."""
+    ext = os.path.splitext(file_path)[1].lower()
+    if ext == ".pdf":
+        return _process_pdf(file_path)
+    if ext in IMAGE_EXTENSIONS:
+        return _process_image(file_path)
+    raise ValueError(f"Unsupported file type: {ext}")

packages.txt

@@ -0,0 +1 @@
+tesseract-ocr

requirements.txt

@@ -0,0 +1,5 @@
+gradio>=4.44.0
+pymupdf>=1.24.0
+pytesseract>=0.3.10
+pillow>=10.0.0
+pandas>=2.0.0