Spaces:
Sleeping
Sleeping
File size: 5,862 Bytes
2a9e166 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """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()
|