app.py

"""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()