Spaces:
Runtime error
Runtime error
| """Hugging Face Spaces / local Gradio entrypoint.""" | |
| from __future__ import annotations | |
| import sys | |
| import tempfile | |
| from decimal import Decimal | |
| from io import BytesIO | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| import gradio as gr | |
| import pandas as pd | |
| ROOT = Path(__file__).resolve().parent | |
| SRC = ROOT / "src" | |
| if str(SRC) not in sys.path: | |
| sys.path.insert(0, str(SRC)) | |
| from invoice_reconciler.mapping import FieldMapping, records_from_table, suggest_mapping | |
| from invoice_reconciler.models import MatchStatus, ReconciliationResult | |
| from invoice_reconciler.parsers import parse_file | |
| from invoice_reconciler.parsers.base import ParsedTable | |
| from invoice_reconciler.reconciliation import reconcile | |
| from invoice_reconciler.reporting import export_excel_report | |
| def _save_upload(file_obj) -> tuple[Path, str]: | |
| """Persist a Gradio upload to a temp path; return path and original name.""" | |
| if file_obj is None: | |
| raise ValueError("No file uploaded") | |
| # gradio File: .name is temp path; .orig_name may exist on NamedString | |
| if isinstance(file_obj, str): | |
| path = Path(file_obj) | |
| return path, path.name | |
| name = getattr(file_obj, "orig_name", None) or Path(str(file_obj)).name | |
| src = Path(str(file_obj)) | |
| suffix = Path(name).suffix or src.suffix | |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) | |
| tmp.write(src.read_bytes()) | |
| tmp.close() | |
| return Path(tmp.name), name | |
| def _fmt_money(v: Decimal | None) -> str: | |
| if v is None: | |
| return "" | |
| return f"{v:.2f}" | |
| def _preview_df(table: ParsedTable | None, limit: int = 15) -> pd.DataFrame: | |
| if table is None or not table.rows: | |
| return pd.DataFrame() | |
| return pd.DataFrame(table.preview(limit)) | |
| def _exceptions_df(result: ReconciliationResult) -> pd.DataFrame: | |
| rows = [] | |
| for e in result.exceptions: | |
| rows.append( | |
| { | |
| "Invoice Number": e.invoice_number, | |
| "Status": e.status.value, | |
| "Source Amount": _fmt_money(e.source_amount), | |
| "QuickBooks Amount": _fmt_money(e.quickbooks_amount), | |
| "Difference": _fmt_money(e.difference), | |
| "Source Page/Row": e.source_location, | |
| "QuickBooks Page/Row": e.quickbooks_location, | |
| "Reason": e.reason, | |
| } | |
| ) | |
| return pd.DataFrame(rows) | |
| def _summary_md(result: ReconciliationResult) -> str: | |
| s = result.summary | |
| return f"""### Summary Dashboard | |
| | Metric | Count | | |
| |---|---:| | |
| | Total Source Invoices | {s.total_source} | | |
| | Total QuickBooks Invoices | {s.total_quickbooks} | | |
| | Matched | {s.matched} | | |
| | Amount Mismatch | {s.amount_mismatch} | | |
| | Missing from QuickBooks | {s.missing_from_quickbooks} | | |
| | Extra in QuickBooks | {s.extra_in_quickbooks} | | |
| | Duplicates | {s.duplicates} | | |
| | Needs Review | {s.needs_review} | | |
| """ | |
| def parse_uploads( | |
| source_file, qb_file | |
| ) -> tuple[ | |
| Optional[ParsedTable], | |
| Optional[ParsedTable], | |
| gr.Dropdown, | |
| gr.Dropdown, | |
| gr.Dropdown, | |
| gr.Dropdown, | |
| pd.DataFrame, | |
| pd.DataFrame, | |
| str, | |
| ]: | |
| if source_file is None or qb_file is None: | |
| empty = gr.Dropdown(choices=[], value=None) | |
| return ( | |
| None, | |
| None, | |
| empty, | |
| empty, | |
| empty, | |
| empty, | |
| pd.DataFrame(), | |
| pd.DataFrame(), | |
| "Upload **both** Source of Truth and QuickBooks files, then click **Parse & detect fields**.", | |
| ) | |
| src_path, src_name = _save_upload(source_file) | |
| qb_path, qb_name = _save_upload(qb_file) | |
| source_table = parse_file(src_path, original_name=src_name) | |
| qb_table = parse_file(qb_path, original_name=qb_name) | |
| src_sug = suggest_mapping(source_table.columns) | |
| qb_sug = suggest_mapping(qb_table.columns) | |
| notes = [ | |
| f"Parsed **{src_name}**: {len(source_table.rows)} rows, columns={source_table.columns}", | |
| f"Parsed **{qb_name}**: {len(qb_table.rows)} rows, columns={qb_table.columns}", | |
| ] | |
| if source_table.ocr_required or source_table.ocr_used: | |
| notes.append("Source PDF may need / used OCR — confirm fields carefully.") | |
| if qb_table.ocr_required or qb_table.ocr_used: | |
| notes.append("QuickBooks PDF may need / used OCR — confirm fields carefully.") | |
| notes.append("Confirm Invoice Number and Gross Amount — do **not** assume the first money column.") | |
| return ( | |
| source_table, | |
| qb_table, | |
| gr.Dropdown(choices=source_table.columns, value=src_sug.invoice_column, interactive=True), | |
| gr.Dropdown(choices=source_table.columns, value=src_sug.amount_column, interactive=True), | |
| gr.Dropdown(choices=qb_table.columns, value=qb_sug.invoice_column, interactive=True), | |
| gr.Dropdown(choices=qb_table.columns, value=qb_sug.amount_column, interactive=True), | |
| _preview_df(source_table), | |
| _preview_df(qb_table), | |
| "\n\n".join(notes), | |
| ) | |
| def run_reconciliation( | |
| source_table: ParsedTable | None, | |
| qb_table: ParsedTable | None, | |
| src_inv: str | None, | |
| src_amt: str | None, | |
| qb_inv: str | None, | |
| qb_amt: str | None, | |
| ) -> tuple[str, pd.DataFrame, Optional[str], Optional[ReconciliationResult]]: | |
| if source_table is None or qb_table is None: | |
| return "Parse both files first.", pd.DataFrame(), None, None | |
| if not all([src_inv, src_amt, qb_inv, qb_amt]): | |
| return "Select Invoice Number and Gross Amount columns for both files.", pd.DataFrame(), None, None | |
| source_records = records_from_table( | |
| source_table, FieldMapping(invoice_column=src_inv, amount_column=src_amt) | |
| ) | |
| qb_records = records_from_table( | |
| qb_table, FieldMapping(invoice_column=qb_inv, amount_column=qb_amt) | |
| ) | |
| result = reconcile(source_records, qb_records) | |
| buf = BytesIO() | |
| export_excel_report(result, buf) | |
| out_path = Path(tempfile.gettempdir()) / "reconciliation_report.xlsx" | |
| out_path.write_bytes(buf.getvalue()) | |
| return _summary_md(result), _exceptions_df(result), str(out_path), result | |
| def build_demo() -> gr.Blocks: | |
| with gr.Blocks(title="Invoice Reconciler") as demo: | |
| gr.Markdown( | |
| """ | |
| # Invoice Reconciler | |
| Source of truth ↔ QuickBooks · **parsers · engine · UI** stay separate | |
| """ | |
| ) | |
| source_state = gr.State(None) | |
| qb_state = gr.State(None) | |
| result_state = gr.State(None) | |
| with gr.Row(): | |
| source_file = gr.File( | |
| label="1. Upload Source of Truth (PDF / XLSX / XLS / CSV)", | |
| file_types=[".pdf", ".xlsx", ".xls", ".csv"], | |
| ) | |
| qb_file = gr.File( | |
| label="2. Upload QuickBooks (PDF / XLSX / XLS / CSV)", | |
| file_types=[".pdf", ".xlsx", ".xls", ".csv"], | |
| ) | |
| parse_btn = gr.Button("Parse & detect fields", variant="secondary") | |
| status = gr.Markdown() | |
| gr.Markdown("### 3. Review detected fields") | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("**Source of Truth mapping**") | |
| src_inv = gr.Dropdown(label="Invoice Number →", choices=[], interactive=True) | |
| src_amt = gr.Dropdown( | |
| label="Gross Amount →", | |
| choices=[], | |
| interactive=True, | |
| info="Do not assume the first monetary column is the amount.", | |
| ) | |
| src_preview = gr.Dataframe(label="Source preview", interactive=False) | |
| with gr.Column(): | |
| gr.Markdown("**QuickBooks mapping**") | |
| qb_inv = gr.Dropdown(label="Invoice Number →", choices=[], interactive=True) | |
| qb_amt = gr.Dropdown(label="Gross Amount →", choices=[], interactive=True) | |
| qb_preview = gr.Dataframe(label="QuickBooks preview", interactive=False) | |
| run_btn = gr.Button("4. Run Reconciliation", variant="primary") | |
| summary = gr.Markdown("### 5. Summary Dashboard\n_Run reconciliation to see counts._") | |
| exceptions = gr.Dataframe(label="6. Detailed Exceptions", interactive=False) | |
| report = gr.File(label="7. Export Report") | |
| parse_btn.click( | |
| fn=parse_uploads, | |
| inputs=[source_file, qb_file], | |
| outputs=[ | |
| source_state, | |
| qb_state, | |
| src_inv, | |
| src_amt, | |
| qb_inv, | |
| qb_amt, | |
| src_preview, | |
| qb_preview, | |
| status, | |
| ], | |
| ) | |
| run_btn.click( | |
| fn=run_reconciliation, | |
| inputs=[source_state, qb_state, src_inv, src_amt, qb_inv, qb_amt], | |
| outputs=[summary, exceptions, report, result_state], | |
| ) | |
| return demo | |
| demo = build_demo() | |
| if __name__ == "__main__": | |
| demo.launch() | |