"""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 labeled data fields (standard IDs/dates/ amounts plus any custom labels the user names), and export everything to CSV as tidy (one-fact-per-row) tables, alongside an audit record of the settings that produced the run. """ import os import tempfile import uuid from datetime import datetime import gradio as gr import pandas as pd import chat_config import extractors import ocr_engine CATEGORY_CHOICES = chat_config.CATEGORY_CHOICES DEFAULT_SETTINGS = { "categories": CATEGORY_CHOICES, "custom_keywords": [], "custom_fields": [], "only_matches": True, } FIELDS_COLUMNS = [ "run_id", "file_name", "page", "field_type", "field_name", "field_value", "text_source", "ocr_confidence", ] NOTIFICATIONS_COLUMNS = [ "run_id", "file_name", "page", "category", "keyword", "context", "text_source", "ocr_confidence", ] METADATA_COLUMNS = [ "run_id", "timestamp", "files", "categories", "custom_keywords", "custom_fields", "only_matches_filter", "chat_model", "chat_transcript", ] QUICK_PROMPTS = { "🏗️ Construction / SAP": ( "Flag change orders, delay notices, and safety violations. " "Pull PO numbers, SAP document numbers, and reference numbers." ), "✅ Compliance": ( "Flag non-compliance, violations, and expired certifications. " "Pull certification codes and expiration dates." ), "💵 Finance / AR": ( "Flag overdue and past-due invoices, credit holds, and disputes. " "Pull invoice numbers, amounts, and due dates." ), } CUSTOM_CSS = """ #hero { text-align: center; padding: 8px 8px 4px 8px; } #hero h1 { font-size: 1.9rem; margin-bottom: 2px; } #hero p { color: var(--body-text-color-subdued); max-width: 720px; margin: 0 auto; } .step-card { border-radius: 16px !important; padding: 22px !important; margin-bottom: 18px; } .step-kicker { font-weight: 600; font-size: 1.05rem; margin-bottom: 2px; } .step-subtitle { color: var(--body-text-color-subdued); margin-bottom: 14px; font-size: 0.92rem; } #run-button button { font-size: 1.05rem !important; padding: 14px !important; } #quick-prompts button { font-size: 0.85rem !important; } """ def _file_path(f): return f if isinstance(f, str) else f.name def _parse_comma_list(s): return [k.strip() for k in s.split(",") if k.strip()] if s and s.strip() else [] def _format_transcript(history): if not history: return "" lines = [] for turn in history: role = turn.get("role", "?") content = turn.get("content", "") lines.append(f"{role}: {content}") return "\n".join(lines) def process_files(files, categories, custom_keywords_str, custom_fields_str, only_matches, chat_history): empty = (pd.DataFrame(columns=FIELDS_COLUMNS), pd.DataFrame(columns=NOTIFICATIONS_COLUMNS)) if not files: return "Upload at least one PDF or image file.", empty[0], empty[1], None, "", {} custom_keywords = _parse_comma_list(custom_keywords_str) custom_field_labels = _parse_comma_list(custom_fields_str) run_id = uuid.uuid4().hex[:8] field_rows = [] notification_rows = [] full_text_log = [] warnings = [] total_pages = 0 file_names = [] for f in files: path = _file_path(f) file_name = os.path.basename(path) file_names.append(file_name) 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"] source = page_data["source"] confidence = page_data["ocr_confidence"] notifications = extractors.find_notifications(text, categories, custom_keywords) std_fields = extractors.extract_fields(text, categories) custom_field_values = extractors.extract_custom_fields(text, custom_field_labels) full_text_log.append(f"--- {file_name} | page {page_data['page']} | {source} ---\n{text}\n") has_content = bool(notifications) or any(std_fields.values()) or any(custom_field_values.values()) if only_matches and not has_content: continue for field_name, values in std_fields.items(): for value in values: field_rows.append({ "run_id": run_id, "file_name": file_name, "page": page_data["page"], "field_type": "standard", "field_name": field_name, "field_value": value, "text_source": source, "ocr_confidence": confidence, }) for field_name, values in custom_field_values.items(): for value in values: field_rows.append({ "run_id": run_id, "file_name": file_name, "page": page_data["page"], "field_type": "custom", "field_name": field_name, "field_value": value, "text_source": source, "ocr_confidence": confidence, }) for n in notifications: notification_rows.append({ "run_id": run_id, "file_name": file_name, "page": page_data["page"], "category": n["category"], "keyword": n["keyword"], "context": n["context"], "text_source": source, "ocr_confidence": confidence, }) fields_df = pd.DataFrame(field_rows, columns=FIELDS_COLUMNS) notifications_df = pd.DataFrame(notification_rows, columns=NOTIFICATIONS_COLUMNS) metadata_row = { "run_id": run_id, "timestamp": datetime.now().isoformat(timespec="seconds"), "files": "; ".join(file_names), "categories": "; ".join(categories), "custom_keywords": "; ".join(custom_keywords), "custom_fields": "; ".join(custom_field_labels), "only_matches_filter": only_matches, "chat_model": chat_config.DEFAULT_MODEL if chat_history else "", "chat_transcript": _format_transcript(chat_history), } metadata_df = pd.DataFrame([metadata_row], columns=METADATA_COLUMNS) csv_paths = [] if not fields_df.empty or not notifications_df.empty: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") tmp_dir = tempfile.gettempdir() fields_path = os.path.join(tmp_dir, f"ocr_fields_{timestamp}_{run_id}.csv") notifications_path = os.path.join(tmp_dir, f"ocr_notifications_{timestamp}_{run_id}.csv") metadata_path = os.path.join(tmp_dir, f"ocr_run_metadata_{timestamp}_{run_id}.csv") fields_df.to_csv(fields_path, index=False) notifications_df.to_csv(notifications_path, index=False) metadata_df.to_csv(metadata_path, index=False) csv_paths = [fields_path, notifications_path, metadata_path] status_lines = [ f"Processed {len(files)} file(s), {total_pages} page(s) total.", f"{len(fields_df)} field value(s), {len(notifications_df)} notification hit(s)" + (" (matches-only pages)." if only_matches else " (all pages)."), ] if warnings: status_lines.append("Warnings: " + "; ".join(warnings)) status = "\n".join(status_lines) return status, fields_df, notifications_df, csv_paths, "\n".join(full_text_log), metadata_row def handle_chat(history, message, settings, hf_token): history = history or [] if not message or not message.strip(): return ( history, "", settings, settings, settings["categories"], ", ".join(settings["custom_keywords"]), ", ".join(settings["custom_fields"]), settings["only_matches"], ) reply, new_settings = chat_config.chat_update(history, message.strip(), settings, hf_token) history = history + [ {"role": "user", "content": message.strip()}, {"role": "assistant", "content": reply}, ] return ( history, "", new_settings, new_settings, new_settings["categories"], ", ".join(new_settings["custom_keywords"]), ", ".join(new_settings["custom_fields"]), new_settings["only_matches"], ) theme = gr.themes.Soft( primary_hue="indigo", secondary_hue="slate", neutral_hue="slate", font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"], ) with gr.Blocks(title="OCR Notification & Data Extractor", theme=theme, css=CUSTOM_CSS) as demo: gr.Markdown( "# 🧾 OCR Notification & Data Extractor\n" "Tell it what you need, upload your documents, and get a clean, " "audit-ready CSV back — built for construction/SAP, compliance, " "and finance teams.", elem_id="hero", ) settings_state = gr.State(dict(DEFAULT_SETTINGS)) with gr.Group(elem_classes=["step-card"]): gr.Markdown('
Step 1 · Tell us what you need
') gr.Markdown( '
Describe what to flag and what to pull as data, ' "in your own words — or tap a starting point below.
" ) with gr.Row(elem_id="quick-prompts"): quick_buttons = {label: gr.Button(label, size="sm") for label in QUICK_PROMPTS} with gr.Accordion("Use your own Hugging Face token (optional)", open=False): gr.Markdown("Only needed if this Space's `HF_TOKEN` secret isn't set, or you want to use your own.") hf_token_input = gr.Textbox(label="Hugging Face token", type="password", placeholder="hf_...", show_label=False) chatbot = gr.Chatbot(height=320, label="Extraction assistant") with gr.Row(): chat_input = gr.Textbox( label="Message", scale=4, show_label=False, container=False, placeholder="e.g. Flag overdue invoices; pull reference number, gross weight, delivery date", ) chat_send = gr.Button("Send", scale=1, variant="primary") settings_display = gr.JSON(label="What we'll extract", value=DEFAULT_SETTINGS) with gr.Group(elem_classes=["step-card"]): gr.Markdown('
Step 2 · Upload your documents
') gr.Markdown('
Upload as many PDFs or images as you need reviewed at once.
') files_input = gr.Files( label="PDF or image files", file_types=[".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp"], file_count="multiple", ) with gr.Accordion("Advanced: edit settings manually", open=False): categories_input = gr.CheckboxGroup( choices=CATEGORY_CHOICES, value=CATEGORY_CHOICES, label="Notification categories", info="Also scopes which standard fields (PO/SAP/invoice/cert) get extracted.", ) custom_keywords_input = gr.Textbox( label="Custom notification keywords (comma-separated)", placeholder="e.g. re-inspection, warranty claim, hold payment", info="Phrases that just raise a flag - not extracted as values.", ) custom_fields_input = gr.Textbox( label="Custom fields to extract as data (comma-separated labels)", placeholder="e.g. reference number, gross weight, delivery date", info="Labels looked up in the document; the value after each label is captured.", ) only_matches_input = gr.Checkbox( value=True, label="Only export pages with a notification hit or extracted field", ) with gr.Group(elem_classes=["step-card"]): gr.Markdown('
Step 3 · Run & review
') run_button = gr.Button("Run OCR & Extract", variant="primary", elem_id="run-button") status_output = gr.Textbox(label="Status", lines=2, interactive=False) csv_output = gr.Files(label="Download CSVs (fields, notifications, run metadata)") with gr.Tabs(): with gr.Tab("Extracted fields"): fields_table_output = gr.Dataframe(headers=FIELDS_COLUMNS, wrap=True) with gr.Tab("Notification hits"): notifications_table_output = gr.Dataframe(headers=NOTIFICATIONS_COLUMNS, wrap=True) with gr.Tab("Run audit"): run_metadata_output = gr.JSON(label="Exactly what settings produced this run") with gr.Tab("Raw OCR text (debug)"): text_output = gr.Textbox(label="Raw extracted text", lines=20, show_label=False) chat_outputs = [ chatbot, chat_input, settings_state, settings_display, categories_input, custom_keywords_input, custom_fields_input, only_matches_input, ] chat_send.click(fn=handle_chat, inputs=[chatbot, chat_input, settings_state, hf_token_input], outputs=chat_outputs) chat_input.submit(fn=handle_chat, inputs=[chatbot, chat_input, settings_state, hf_token_input], outputs=chat_outputs) for label, prompt in QUICK_PROMPTS.items(): quick_buttons[label].click(fn=lambda p=prompt: p, outputs=chat_input).then( fn=handle_chat, inputs=[chatbot, chat_input, settings_state, hf_token_input], outputs=chat_outputs ) run_button.click( fn=process_files, inputs=[files_input, categories_input, custom_keywords_input, custom_fields_input, only_matches_input, chatbot], outputs=[status_output, fields_table_output, notifications_table_output, csv_output, text_output, run_metadata_output], ) if __name__ == "__main__": demo.launch()