# ========================= # Invoice Extractor (Qwen3-VL via RunPod vLLM) - Batch Mode # Raw model output + deterministic computation post-processing # ========================= import os from pathlib import Path # ----------------------------- # Environment hardening (HF Spaces, /.cache issue) # ----------------------------- _home = os.environ.get("HOME", "") if _home in ("", "/", None): repo_dir = os.getcwd() safe_home = repo_dir if os.access(repo_dir, os.W_OK) else "/tmp" os.environ["HOME"] = safe_home print(f"[startup] HOME not set or unwritable — setting HOME={safe_home}") streamlit_dir = Path(os.environ["HOME"]) / ".streamlit" try: streamlit_dir.mkdir(parents=True, exist_ok=True) print(f"[startup] ensured {streamlit_dir}") except Exception as e: print(f"[startup] WARNING: could not create {streamlit_dir}: {e}") # ----------------------------- # Imports # ----------------------------- import json from io import BytesIO import hashlib import copy from dataclasses import dataclass from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN, getcontext from typing import Dict, Any, List import streamlit as st import pandas as pd from PIL import Image try: from pdf2image import convert_from_bytes except Exception: convert_from_bytes = None import requests import base64 import re # ----------------------------- # Embedded computation post-processing engine # Mirrors the repo Cat-2 Decimal solver; no external repo file required. # ----------------------------- _ISO_CURRENCY_DECIMALS = { "USD": 2, "EUR": 2, "GBP": 2, "INR": 2, "CAD": 2, "AUD": 2, "NZD": 2, "SGD": 2, "HKD": 2, "ZAR": 2, "CHF": 2, "SEK": 2, "DKK": 2, "NOK": 2, "PLN": 2, "BRL": 2, "MXN": 2, "ARS": 2, "CNY": 2, "PHP": 2, "MYR": 2, "JPY": 0, "KRW": 0, "BHD": 3, "KWD": 3, "OMR": 3, "TWD": 2, } _NULLISH = {"", "-", "—", "–", "n/a", "na", "null", "none", ".", "nan"} _OVERFLOW_HI = Decimal("1E14") _UNDERFLOW_LO = Decimal("1E-6") _MAX_SOLVER_ITERATIONS = 10 @dataclass class _SummaryState: currency: str | None = None subtotal: Decimal | None = None tax_rate: Decimal | None = None tax_amount: Decimal | None = None discount_rate: Decimal | None = None total_discount_amount: Decimal | None = None total_amount: Decimal | None = None vlm_subtotal: Decimal | None = None vlm_tax_amount: Decimal | None = None vlm_total_amount: Decimal | None = None @dataclass class _LineState: source: dict[str, Any] index: int quantity: Decimal | None = None unit_price: Decimal | None = None amount: Decimal | None = None discount_rate_per_item: Decimal | None = None discount_amount_per_item: Decimal | None = None taxable_base: Decimal | None = None tax_rate_per_item: Decimal | None = None tax_amount_per_item: Decimal | None = None line_total: Decimal | None = None def _is_blank(value: Any) -> bool: if value is None: return True if isinstance(value, float) and str(value).lower() == "nan": return True return str(value).strip().lower() in _NULLISH def _first_present(data: dict[str, Any], *keys: str) -> Any: for key in keys: if key in data: return data.get(key) return None def _parse_decimal(raw: Any) -> Decimal | None: if raw is None: return None if isinstance(raw, Decimal): return raw if raw.is_finite() else None if isinstance(raw, bool): return None if isinstance(raw, int): return Decimal(raw) if isinstance(raw, float): try: value = Decimal(str(raw)) except InvalidOperation: return None return value if value.is_finite() else None s = str(raw).strip() if s.lower() in _NULLISH: return None negative_paren = len(s) >= 2 and s[0] == "(" and s[-1] == ")" s = s.replace(" ", " ").replace(" ", " ").replace(" ", " ") s = re.sub(r"[^\d,.\-]", "", s) if not s or s in {"-", ".", ",", "-."}: return None if "," in s and "." in s: if s.rfind(",") > s.rfind("."): s = s.replace(".", "").replace(",", ".") else: s = s.replace(",", "") elif "," in s: left, _, right = s.partition(",") if len(right) == 3 and right.isdigit(): s = left + right elif right.isdigit() and len(right) in (1, 2): s = f"{left}.{right}" else: s = s.replace(",", "") try: value = Decimal(s) except InvalidOperation: return None if not value.is_finite(): return None return -abs(value) if negative_paren else value def _format_decimal(value: Decimal | None, *, min_places: int = 0) -> str: if value is None: return "" if value == 0: value = Decimal("0") s = format(value, "f") if "." in s: whole, frac = s.split(".", 1) frac = frac.rstrip("0") if len(frac) < min_places: frac += "0" * (min_places - len(frac)) return f"{whole}.{frac}" if frac else whole if min_places: return f"{s}.{'0' * min_places}" return s def _minor_unit_value(currency: str | None) -> Decimal: decimals = _ISO_CURRENCY_DECIMALS.get((currency or "").upper(), 2) return Decimal("1") if decimals == 0 else Decimal("1").scaleb(-decimals) def _per_line_tolerance(currency: str | None, num_lines: int) -> Decimal: minor_units = min(Decimal("100"), Decimal("1") + Decimal(num_lines)) return minor_units * _minor_unit_value(currency) def _summary_tolerance(currency: str | None, num_lines: int) -> Decimal: return _per_line_tolerance(currency, num_lines) * Decimal("1.5") def _finite_guard(value: Decimal | None) -> Decimal | None: if value is None: return None if not value.is_finite(): raise ArithmeticError("non-finite Decimal") if value == 0: return value av = abs(value) if av > _OVERFLOW_HI or av < _UNDERFLOW_LO: raise ArithmeticError("value outside financial bounds") return value def _flag(flags: list[str], name: str, detail: str) -> None: msg = f"{name}: {detail}" if msg not in flags: flags.append(msg) def _charge_applied(ef: _SummaryState, lines: list[_LineState], tolerance: Decimal) -> tuple[bool, bool]: total = ef.total_amount base = ef.subtotal if base is None: amounts = [li.amount for li in lines if li.amount is not None] base = sum(amounts, Decimal("0")) if amounts else None if total is None or base is None: return True, True known_disc = ef.total_discount_amount or Decimal("0") known_tax = ef.tax_amount or Decimal("0") return ( total > (base - known_disc) + tolerance, total < (base + known_tax) - tolerance, ) def _solve_line( li: _LineState, ef: _SummaryState, tol: Decimal, flags: list[str], *, tax_applied: bool = True, discount_applied: bool = True, ) -> None: hundred = Decimal("100") def credit_polarity() -> bool: if ef.total_amount is not None and ef.total_amount < 0: return True return any( v is not None and v < 0 for v in (li.amount, li.line_total, li.tax_amount_per_item, li.taxable_base) ) def amount_from_qty_unit() -> Decimal | None: if li.quantity is None or li.unit_price is None: return None computed = li.quantity * li.unit_price return -computed if credit_polarity() and computed > 0 else computed def quantity_from_amount_unit() -> Decimal | None: if li.amount is None or li.unit_price in (None, Decimal("0")): return None if credit_polarity() and li.unit_price > 0: mag = abs(li.amount) / li.unit_price return -mag if li.quantity is not None and li.quantity < 0 else mag return li.amount / li.unit_price def unit_price_from_amount_qty() -> Decimal | None: if li.amount is None or li.quantity in (None, Decimal("0")): return None if credit_polarity() and li.quantity > 0: mag = abs(li.amount) / li.quantity return -mag if li.unit_price is not None and li.unit_price < 0 else mag return li.amount / li.quantity def assign(field: str, computed: Decimal | None) -> None: if computed is None: return try: computed = _finite_guard(computed) except ArithmeticError as exc: _flag(flags, "math_error_line_items", f"line {li.index} {field}: {exc}") return existing = getattr(li, field) if existing is None: setattr(li, field, computed) elif abs(existing - computed) > tol: _flag( flags, "math_error_line_items", f"line {li.index} {field}: VLM={existing} solver={computed}", ) has_disc = ( (li.discount_rate_per_item is not None and discount_applied) or li.discount_amount_per_item is not None or ef.total_discount_amount is not None or (ef.discount_rate is not None and discount_applied) ) has_tax = ( (li.tax_rate_per_item is not None and tax_applied) or li.tax_amount_per_item is not None or ef.tax_amount is not None or (ef.tax_rate is not None and tax_applied) ) for _ in range(_MAX_SOLVER_ITERATIONS): before = ( li.amount, li.quantity, li.unit_price, li.discount_amount_per_item, li.discount_rate_per_item, li.taxable_base, li.tax_amount_per_item, li.tax_rate_per_item, li.line_total, ) assign("amount", amount_from_qty_unit()) assign("quantity", quantity_from_amount_unit()) assign("unit_price", unit_price_from_amount_qty()) if has_disc: if li.amount is not None and li.discount_rate_per_item is not None: assign("discount_amount_per_item", li.amount * li.discount_rate_per_item / hundred) if li.discount_amount_per_item is not None and li.amount not in (None, Decimal("0")): assign("discount_rate_per_item", li.discount_amount_per_item / li.amount * hundred) if li.discount_amount_per_item is not None and li.discount_rate_per_item not in (None, Decimal("0")): assign("amount", li.discount_amount_per_item * hundred / li.discount_rate_per_item) if li.amount is not None and li.discount_amount_per_item is not None: assign("taxable_base", li.amount - li.discount_amount_per_item) if li.taxable_base is not None and li.discount_amount_per_item is not None: assign("amount", li.taxable_base + li.discount_amount_per_item) if li.amount is not None and li.taxable_base is not None: assign("discount_amount_per_item", li.amount - li.taxable_base) if li.amount is not None and li.discount_rate_per_item is not None: assign("taxable_base", li.amount * (hundred - li.discount_rate_per_item) / hundred) if li.taxable_base is not None and li.discount_rate_per_item not in (None, hundred): assign("amount", li.taxable_base * hundred / (hundred - li.discount_rate_per_item)) if li.amount not in (None, Decimal("0")) and li.taxable_base is not None: assign("discount_rate_per_item", (li.amount - li.taxable_base) / li.amount * hundred) else: if li.amount is not None: assign("taxable_base", li.amount) if li.taxable_base is not None: assign("amount", li.taxable_base) if has_tax: if li.taxable_base is not None and li.tax_rate_per_item is not None: assign("tax_amount_per_item", li.taxable_base * li.tax_rate_per_item / hundred) if li.tax_amount_per_item is not None and li.taxable_base not in (None, Decimal("0")): assign("tax_rate_per_item", li.tax_amount_per_item / li.taxable_base * hundred) if li.tax_amount_per_item is not None and li.tax_rate_per_item not in (None, Decimal("0")): assign("taxable_base", li.tax_amount_per_item * hundred / li.tax_rate_per_item) if li.taxable_base is not None and li.tax_amount_per_item is not None: assign("line_total", li.taxable_base + li.tax_amount_per_item) if li.line_total is not None and li.tax_amount_per_item is not None: assign("taxable_base", li.line_total - li.tax_amount_per_item) if li.line_total is not None and li.taxable_base is not None: assign("tax_amount_per_item", li.line_total - li.taxable_base) if li.taxable_base is not None and li.tax_rate_per_item is not None: assign("line_total", li.taxable_base * (hundred + li.tax_rate_per_item) / hundred) if li.line_total is not None and li.tax_rate_per_item not in (None, -hundred): assign("taxable_base", li.line_total * hundred / (hundred + li.tax_rate_per_item)) if li.line_total is not None and li.taxable_base not in (None, Decimal("0")): assign("tax_rate_per_item", (li.line_total - li.taxable_base) / li.taxable_base * hundred) else: if li.taxable_base is not None: assign("line_total", li.taxable_base) if li.line_total is not None: assign("taxable_base", li.line_total) after = ( li.amount, li.quantity, li.unit_price, li.discount_amount_per_item, li.discount_rate_per_item, li.taxable_base, li.tax_amount_per_item, li.tax_rate_per_item, li.line_total, ) if before == after: break def _propagate_summary_rates( ef: _SummaryState, lines: list[_LineState], tax_applied: bool, discount_applied: bool, ) -> None: if ef.tax_rate is not None and tax_applied: for li in lines: if li.tax_rate_per_item is None and li.tax_amount_per_item is None: li.tax_rate_per_item = ef.tax_rate if ef.discount_rate is not None and discount_applied: for li in lines: if li.discount_rate_per_item is None and li.discount_amount_per_item is None: li.discount_rate_per_item = ef.discount_rate def _proportional( lines: list[_LineState], total: Decimal, *, weight, setter: str, signed_weight_base: bool = False, ) -> None: unit = Decimal("0.0001") raw_weights = [weight(li) for li in lines] weights = raw_weights if signed_weight_base else [abs(w) for w in raw_weights] wsum = sum(weights, Decimal("0")) if wsum == 0: return distributed = [] for li, w in zip(lines, weights): share = (total * (w / wsum)).quantize(unit) setattr(li, setter, share) distributed.append(share) residue = total - sum(distributed, Decimal("0")) if residue != 0: target = max(range(len(lines)), key=lambda i: abs(raw_weights[i])) current = getattr(lines[target], setter) setattr(lines[target], setter, current + residue) def _distribute(ef: _SummaryState, lines: list[_LineState], flags: list[str]) -> None: if ef.total_discount_amount is not None and lines: need = [li for li in lines if li.discount_amount_per_item is None and li.amount is not None] if need: _proportional( need, ef.total_discount_amount, weight=lambda x: x.amount, setter="discount_amount_per_item", ) if ef.tax_amount is not None: pool = [] for li in lines: if li.tax_rate_per_item == Decimal("0"): if li.tax_amount_per_item is None: li.tax_amount_per_item = Decimal("0") continue if li.tax_amount_per_item is not None: continue if li.taxable_base is None and li.amount is not None: li.taxable_base = li.amount - (li.discount_amount_per_item or Decimal("0")) if li.taxable_base is not None: pool.append(li) if pool: total_base = sum((li.taxable_base for li in pool), Decimal("0")) if total_base == 0: if ef.tax_amount != 0: _flag(flags, "tax_distribution_impossible", f"summary_tax={ef.tax_amount}") else: _proportional( pool, ef.tax_amount, weight=lambda x: x.taxable_base, setter="tax_amount_per_item", signed_weight_base=True, ) elif ef.tax_amount != 0: placed = sum((li.tax_amount_per_item for li in lines if li.tax_amount_per_item is not None), Decimal("0")) if placed != ef.tax_amount: _flag(flags, "tax_distribution_impossible", f"summary_tax={ef.tax_amount}") for li in lines: if li.amount is not None and li.discount_amount_per_item is not None and li.tax_amount_per_item is not None: li.line_total = li.amount - li.discount_amount_per_item + li.tax_amount_per_item def _aggregate_summary(ef: _SummaryState, lines: list[_LineState], tol: Decimal, flags: list[str]) -> None: def agg(attr: str) -> Decimal | None: vals = [getattr(li, attr) for li in lines if getattr(li, attr) is not None] return sum(vals, Decimal("0")) if vals else None discount_vals = [li.discount_amount_per_item for li in lines if li.discount_amount_per_item is not None] sum_amount = agg("amount") sum_disc = sum(discount_vals, Decimal("0")) if discount_vals else None sum_tax = agg("tax_amount_per_item") sum_base = agg("taxable_base") def reconcile(working_attr: str, vlm_attr: str, computed: Decimal | None) -> None: if computed is None: return if getattr(ef, working_attr) is None: setattr(ef, working_attr, computed) vlm_val = getattr(ef, vlm_attr) if vlm_val is not None and abs(vlm_val - computed) > tol: _flag(flags, "field_conflict", f"{working_attr}: VLM={vlm_val} solver={computed}") reconcile("subtotal", "vlm_subtotal", sum_amount) if sum_disc is not None and ef.total_discount_amount is None and any(v != 0 for v in discount_vals): ef.total_discount_amount = sum_disc if sum_tax is None and ef.vlm_tax_amount is not None and ef.vlm_tax_amount != 0: _flag(flags, "field_conflict", f"tax_amount: VLM={ef.vlm_tax_amount} solver=0") reconcile("tax_amount", "vlm_tax_amount", sum_tax) if ef.total_discount_amount is not None and ef.subtotal not in (None, Decimal("0")) and ef.discount_rate is None: ef.discount_rate = ef.total_discount_amount / ef.subtotal * Decimal("100") if sum_tax is not None and sum_base not in (None, Decimal("0")) and ef.tax_rate is None: ef.tax_rate = sum_tax / sum_base * Decimal("100") if ef.subtotal is not None: computed_total = ef.subtotal - (ef.total_discount_amount or Decimal("0")) + (ef.tax_amount or Decimal("0")) if ef.total_amount is None: ef.total_amount = computed_total if ef.vlm_total_amount is not None and abs(ef.vlm_total_amount - computed_total) > tol: _flag(flags, "field_conflict", f"total_amount: VLM={ef.vlm_total_amount} solver={computed_total}") def _solve_summary_chain(ef: _SummaryState, tol: Decimal, flags: list[str], tax_applied: bool, discount_applied: bool) -> None: syn = _LineState( source={}, index=0, amount=ef.subtotal, discount_amount_per_item=ef.total_discount_amount, discount_rate_per_item=ef.discount_rate, tax_amount_per_item=ef.tax_amount, tax_rate_per_item=ef.tax_rate, line_total=ef.total_amount, ) _solve_line(syn, ef, tol, flags, tax_applied=tax_applied, discount_applied=discount_applied) if ef.subtotal is None: ef.subtotal = syn.amount if ef.total_discount_amount is None: ef.total_discount_amount = syn.discount_amount_per_item if ef.discount_rate is None: ef.discount_rate = syn.discount_rate_per_item if ef.tax_amount is None: ef.tax_amount = syn.tax_amount_per_item if ef.tax_rate is None: ef.tax_rate = syn.tax_rate_per_item if ef.total_amount is None: ef.total_amount = syn.line_total def _snapshot(ef: _SummaryState, lines: list[_LineState]) -> tuple: return ( tuple( ( li.amount, li.quantity, li.unit_price, li.discount_amount_per_item, li.discount_rate_per_item, li.taxable_base, li.tax_amount_per_item, li.tax_rate_per_item, li.line_total, ) for li in lines ), (ef.subtotal, ef.total_discount_amount, ef.discount_rate, ef.tax_amount, ef.tax_rate, ef.total_amount), ) def postprocess_invoice_data(invoice_data: dict[str, Any]) -> dict[str, Any]: getcontext().prec = 28 getcontext().rounding = ROUND_HALF_EVEN data = copy.deepcopy(invoice_data or {}) raw_items = data.get("Itemized Data", []) or [] if not isinstance(raw_items, list): raw_items = [] ef = _SummaryState( currency=(str(data.get("Currency", "")).strip() or None), subtotal=_parse_decimal(data.get("Subtotal")), tax_rate=_parse_decimal(data.get("Tax Percentage")), tax_amount=_parse_decimal(data.get("Total Tax")), discount_rate=_parse_decimal(data.get("Discount Rate")), total_discount_amount=_parse_decimal(data.get("Total Discount Amount")), total_amount=_parse_decimal(data.get("Total Amount")), ) ef.vlm_subtotal = ef.subtotal ef.vlm_tax_amount = ef.tax_amount ef.vlm_total_amount = ef.total_amount lines = [] for idx, item in enumerate(raw_items, start=1): if not isinstance(item, dict): continue lines.append( _LineState( source=item, index=idx, quantity=_parse_decimal(_first_present(item, "Quantity", "Item Quantity")), unit_price=_parse_decimal(_first_present(item, "Unit Price", "Item Unit Price")), amount=_parse_decimal(_first_present(item, "Amount", "Item Amount")), discount_rate_per_item=_parse_decimal(item.get("Discount Rate Per Item")), discount_amount_per_item=_parse_decimal(item.get("Discount Amount Per Item")), tax_rate_per_item=_parse_decimal(item.get("Tax Rate Per Item")), tax_amount_per_item=_parse_decimal(item.get("Tax Amount Per Item")), line_total=_parse_decimal(_first_present(item, "Line Total", "Item Line Total")), ) ) num_lines = max(1, len(lines)) per_line_tol = _per_line_tolerance(ef.currency, num_lines) summary_tol = _summary_tolerance(ef.currency, num_lines) if not (ef.total_amount is not None and ef.total_amount < 0): for li in lines: if li.amount is None and li.quantity is not None and li.unit_price is not None: li.amount = li.quantity * li.unit_price flags: list[str] = [] tax_applied, discount_applied = _charge_applied(ef, lines, summary_tol) for _ in range(_MAX_SOLVER_ITERATIONS): mark = _snapshot(ef, lines) flags = [] for li in lines: _solve_line(li, ef, per_line_tol, flags, tax_applied=tax_applied, discount_applied=discount_applied) tax_applied, discount_applied = _charge_applied(ef, lines, summary_tol) _propagate_summary_rates(ef, lines, tax_applied, discount_applied) _distribute(ef, lines, flags) _aggregate_summary(ef, lines, summary_tol, flags) _solve_summary_chain(ef, summary_tol, flags, tax_applied, discount_applied) if _snapshot(ef, lines) == mark: break computed: list[str] = [] summary_fields = [ ("subtotal", "Subtotal", 2), ("tax_rate", "Tax Percentage", 0), ("tax_amount", "Total Tax", 2), ("discount_rate", "Discount Rate", 0), ("total_discount_amount", "Total Discount Amount", 2), ("total_amount", "Total Amount", 2), ] for attr, key, min_places in summary_fields: value = getattr(ef, attr) if value is not None and _is_blank(data.get(key)): data[key] = _format_decimal(value, min_places=min_places) computed.append(f"{key} = {data[key]}") line_fields = [ ("quantity", "Quantity", 0), ("unit_price", "Unit Price", 2), ("amount", "Amount", 2), ("discount_rate_per_item", "Discount Rate Per Item", 0), ("discount_amount_per_item", "Discount Amount Per Item", 2), ("tax_rate_per_item", "Tax Rate Per Item", 0), ("tax_amount_per_item", "Tax Amount Per Item", 2), ("line_total", "Line Total", 2), ] for li in lines: for attr, key, min_places in line_fields: value = getattr(li, attr) if value is None: continue old_value = _first_present(li.source, key, f"Item {key}") parsed_old = _parse_decimal(old_value) if _is_blank(old_value) or parsed_old != value: li.source[key] = _format_decimal(value, min_places=min_places) computed.append(f"line[{li.index}].{key} = {li.source[key]}") data["Postprocessing Notes"] = { "engine": "standalone_idp_cat2_computation", "computed_fields": computed, "risk_flags": flags, } return data POD_URL = os.getenv("POD_URL", "") VLLM_API_KEY = os.getenv("VLLM_API_KEY", "") MODEL_NAME = "phase2-v1-merged" MAX_PAGES_PER_REQUEST = 10 MAX_DIMENSION_PER_PAGE = 1560 MAX_TOTAL_PAYLOAD_MB = 40.0 if not POD_URL or not VLLM_API_KEY: st.error("⚠️ API credentials not configured. Please set POD_URL and VLLM_API_KEY in Space settings.") st.stop() # ----------------------------- # Page config & CSS # ----------------------------- st.set_page_config(page_title="Invoice Extractor (Qwen3-VL) - Batch Mode", layout="wide") st.title("Invoice Extraction") st.markdown( """ """, unsafe_allow_html=True ) DATA_EDITOR_HEIGHT = 380 # ----------------------------- # Helpers — date only, no numeric cleaning # ----------------------------- # ----------------------------- # Helpers — date parsing only, raw strings preserved for display # ----------------------------- try: import dateparser def parse_date_to_object(date_str, currency=None): if not date_str or str(date_str).strip() == "": return None settings = {} if currency and currency.upper() == "USD": settings["DATE_ORDER"] = "MDY" try: result = dateparser.parse(str(date_str).strip(), settings=settings) return result.date() if result else None except Exception: return None except ImportError: dateparser = None def parse_date_to_object(date_str, currency=None): return None # ============================================================================= # vLLM Inference — MULTI-IMAGE SINGLE REQUEST # ============================================================================= def _encode_page(page: Image.Image, page_idx: int, max_dim: int = MAX_DIMENSION_PER_PAGE): orig_w, orig_h = page.size api_img = page.copy() if orig_w > max_dim or orig_h > max_dim: ratio = min(max_dim / orig_w, max_dim / orig_h) new_size = (int(orig_w * ratio), int(orig_h * ratio)) api_img = api_img.resize(new_size, Image.Resampling.LANCZOS) else: new_size = (orig_w, orig_h) buf = BytesIO() api_img.save(buf, format="PNG", optimize=True) b64 = base64.b64encode(buf.getvalue()).decode("utf-8") data_url = f"data:image/png;base64,{b64}" return data_url, (orig_w, orig_h), new_size def run_inference_vllm(images: List[Image.Image]) -> str | None: EXTRACTION_PROMPT = """Extract structured data from the provided invoice image(s). Return a single JSON object with three sections: header, items, summary. ## Output Schema Return this exact structure. Use empty string "" for any field not found. json { "header": { "invoice_no": "", "po_no": "", "invoice_date": "", "payment_terms": "", "due_date": "", "sender_name": "", "sender_addr": "", "tax_id": "", "rcpt_name": "", "rcpt_addr": "", "bank_iban": "", "bank_name": "", "bank_acc_no": "", "bank_routing": "", "bank_swift": "", "bank_acc_name": "", "bank_branch": "" }, "items": [ { "service_name": "", "service_start_date": "", "service_end_date": "", "descriptions": "", "SKU": "", "quantity": "", "unit_price": "", "amount": "", "discount_rate_per_item": "", "discount_amount_per_item": "", "tax_rate_per_item": "", "tax_amount_per_item": "", "line_total": "" } ], "summary": { "subtotal": "", "tax_rate": "", "tax_amount": "", "discount_rate": "", "total_discount_amount": "", "total_amount": "", "currency": "" } } ## Core Rules 1. EXTRACT ONLY WHAT IS EXPLICITLY VISIBLE. Never compute, derive, infer totals, or fill missing values via arithmetic. The following format normalizations and explicit-data operations ARE permitted: - Bracketed amounts → negative: (100.00) → "-100.00". - Date ranges → expanded to full start/end dates using the invoice year. - Language normalization of payment terms → English. - Summing multiple tax AMOUNTS shown separately into one combined amount. - Summing multiple discount AMOUNTS shown separately into one combined amount. - Default quantity to "1.00" when unit_price and amount are both shown and equal, with no quantity printed. - Setting tax to "0.00" when "No VAT" / "VAT Exempt" is explicitly stated, OR when a tax rate is shown but tax amount is blank AND subtotal equals total as printed. 2. Return "" for any field not present or not clearly identifiable. 3. PRESERVE ORIGINAL FORMATTING for numbers (commas, decimals) and dates (except for date-range expansion noted above). For tax_rate and discount_rate, preserve exactly as printed including any % symbol. 4. STRIP CURRENCY SYMBOLS from all amount/price fields. Place currency code (USD, EUR, GBP, INR, etc.) only in summary.currency. If multiple currencies appear, use the currency of total_amount. If no currency is shown anywhere on the invoice, set summary.currency to "". 5. MULTI-PAGE: Merge all pages into one JSON object. Extract every line item as printed — do not deduplicate. Only skip a row if it is clearly a continuation header (e.g., the row repeats the table column titles like "Description / Qty / Price"). 6. NO DUPLICATION ACROSS SCHEMA SECTIONS: If a value appears in the line items area of the invoice, place it only in the items array of the schema. If it appears in the summary/totals area (the section at the bottom showing subtotal, tax, discount, and total — separate from the itemized table), place it only in the summary object of the schema. Do not copy the same value into both unless the invoice explicitly prints it in both areas. 7. OUTPUT ONLY THE JSON. No preamble, no markdown fences, no explanation. 8. INCLUDE ALL LINE ITEMS, even those with a zero amount, zero quantity, or negative amount. Never skip a line item because its value is 0 or negative. This applies equally to credit notes and adjustments. Negative quantities, if printed, are extracted as-is (preserve the negative sign). 9. NO LINE ITEMS: If the invoice shows no individual line items (e.g., a flat-fee invoice with only a total), return an empty array: "items": []. 10. CONFLICTING VALUES: If the invoice shows two different totals (e.g., "Total" and "Amount Due"), prefer the value labeled as the final payable amount ("Amount Due", "Balance Due", "Total Payable", "Grand Total"). ## Field Rules ### Sender & Recipient 1.sender_name: The invoicing company/person. Prefer company name over person name. 2.sender_addr: The full address of the invoicing entity, wherever it appears on the invoice. 3.tax_id: Always the SENDER's tax ID / VAT number. Never the recipient's. 4.rcpt_name: The name of the recipient, customer, or entity being billed. 5.rcpt_addr: The address of the recipient, customer, or entity being billed. ### PO Number 1.po_no: Extract if a single PO number is shown. If the same PO number appears multiple times on the invoice (e.g., header and footer), extract it once. If multiple DISTINCT PO numbers are present, return "". ### Bank Details 1.Prefer the bank account matching the invoice currency (USD invoice → USD account). 2.If no currency-matching account exists, extract the only available account. 3.If multiple non-matching accounts exist with no currency-matching account, return "" for all bank fields. 4.bank_acc_name: The name of the account holder. It may also appear in phrases such as “Cheques should be payable to” or “Remittance should be payable to.” Extract the mentioned company/entity name as the bank_acc_name. ### Dates 5.Preserve the original date format exactly as printed. 6.Date range exception: When a service period is shown as a range (e.g., "Feb 1-3"), expand to two full dates: service_start_date: "Feb 1 2026", service_end_date: "Feb 3 2026". Use the invoice year if the range omits a year. 7.due_date: If multiple due dates, extract only the first. If payment terms are "due on receipt" / "due on presentation" / "immediate", set due_date = invoice_date. 8.payment_terms: Normalize to English ("30 jours" → "30 days", "60 gg df" → "60 days"). ### Descriptions 1.service_name: A short label, product name, or service title (e.g., "Consulting", "Web Hosting", "Premium Plan"). 2.descriptions: The full detailed description text of the line item. 3.If the line item shows both a short label and a longer expanded text → put the short label in service_name and the longer text in descriptions. 4.If only one text is available for the line item → place it in descriptions and leave service_name as "". 5.If a SKU/code appears together with descriptive text (e.g., "SKU-12345 — Premium Web Hosting") → put the SKU in SKU, the descriptive portion in descriptions, and leave service_name as "" unless a separate short label also exists. 6.Do not extract handwritten annotations or stamps unless they are clearly legible and printed-quality. When in doubt, extract only the printed text. ### Amounts, Tax & Discounts 1.Preserve original number formatting (commas, decimals) exactly as printed. Do not reinterpret decimal separators — preserve as written (European 1.234,56 stays 1.234,56; US 1,234.56 stays 1,234.56). 2.Bracketed amounts are negative: (100.00) → "-100.00". 3."No VAT" / "VAT Exempt" → set relevant tax field to "0.00". 4.Tax rate shown but tax amount blank AND subtotal equals total as printed → tax amount = "0.00". 5.Tax-inclusive pricing: If the invoice states "all prices include VAT" or "tax-inclusive pricing", extract amount as printed without separating tax. Extract summary.tax_amount, summary.tax_rate, tax_rate_per_item, and tax_amount_per_item exactly as printed on the invoice — do not attempt to back-calculate. 6.Summing multiple values: - Multiple tax AMOUNTS shown separately → sum into one combined amount in the appropriate field. - Multiple DISCOUNT AMOUNTS shown separately → sum into one combined amount in the appropriate field. - Multiple tax RATES (e.g., 5% CGST + 12% SGST) → do NOT sum. Concatenate as printed (e.g., "5% + 12%") or extract the dominant rate as printed. Never produce a summed rate like "17%". 7.0Gross vs. Net amount columns: If both gross and net per-line amounts are shown, use the NET (pre-tax) amount as amount. 8.Discount/tax as a line item: If discount or tax appears as a row in the line items table — identified by labels such as "Discount", "Rebate", "Tax", "VAT", "Adjustment", or by being a negative-value row with no quantity/unit price — create a separate entry in the items array. Populate ONLY descriptions, the relevant discount/tax fields (discount_rate_per_item, discount_amount_per_item, tax_rate_per_item, tax_amount_per_item), and line_total. Leave quantity, unit_price, and amount as "". 9.Discount/tax in the totals/summary area: Place only in the summary object of the schema. Do not duplicate into the items array. ### Quantity & Unit Price 1.If unit_price and amount are shown and equal but no quantity is printed → quantity: "1.00". 2.If the invoice uses alternate labels for quantity (e.g., impressions, units, hours, sessions, clicks) and unit price (e.g., CPM, rate, cost per unit), map them to quantity and unit_price respectively. 3.If both a standard quantity and an alternate metric are shown, extract the standard quantity. 4.Negative quantities (e.g., on credit notes) are preserved as-is with the negative sign. ### Line Total line_total: Extract ONLY if ALL these conditions are met: 1. The invoice explicitly prints a dedicated line total column or value for the row. 2. The invoice has NO separate per-line tax column or per-line tax amount anywhere. 3. The invoice has NO separate per-line discount column or per-line discount amount anywhere. If any per-line tax or discount column exists anywhere on the invoice → set line_total to "" for ALL rows. If no line total column exists on the invoice → line_total = "".""" if not images: st.error("No images provided to run_inference_vllm.") return None if len(images) > MAX_PAGES_PER_REQUEST: st.warning( f"Invoice has {len(images)} pages — only the first " f"{MAX_PAGES_PER_REQUEST} will be sent per the request limit." ) images = images[:MAX_PAGES_PER_REQUEST] try: image_content_blocks: List[dict] = [] total_b64_bytes = 0 resize_info = [] for idx, page in enumerate(images): data_url, orig_size, new_size = _encode_page(page, idx) total_b64_bytes += len(data_url) if orig_size != new_size: resize_info.append( f"Page {idx+1}: {orig_size[0]}×{orig_size[1]} → {new_size[0]}×{new_size[1]}" ) image_content_blocks.append({"type": "image_url", "image_url": {"url": data_url}}) if resize_info: st.info("Resized for API payload:\n" + "\n".join(resize_info)) total_payload_mb = total_b64_bytes / (1024 * 1024) if total_payload_mb > MAX_TOTAL_PAYLOAD_MB: st.warning( f"Large multi-page payload ({total_payload_mb:.1f} MB across " f"{len(images)} page(s)). This may be slow or time-out." ) else: st.info(f"Sending {len(images)} page(s) in one request (payload ≈ {total_payload_mb:.2f} MB).") user_content = image_content_blocks + [ { "type": "text", "text": ( f"The {len(images)} image(s) above are all pages of the same invoice " f"(page 1 through page {len(images)}).\n" "Extract invoice data from ALL pages combined into a single JSON object." ) } ] payload = { "model": MODEL_NAME, "messages": [ {"role": "system", "content": EXTRACTION_PROMPT}, {"role": "user", "content": user_content} ], "temperature": 0, "max_tokens": 10000 } headers = { "Authorization": f"Bearer {VLLM_API_KEY}", "Content-Type": "application/json" } response = requests.post( f"{POD_URL}/v1/chat/completions", headers=headers, json=payload, timeout=600 ) if response.status_code == 200: result = response.json() return result["choices"][0]["message"]["content"] else: st.error(f"❌ API Error {response.status_code}") try: st.json(response.json()) except Exception: st.code(response.text) return None except Exception as e: st.error(f"Error calling vLLM: {str(e)}") return None # ----------------------------- # JSON Parser — raw strings, no numeric conversion # ----------------------------- def parse_vllm_json(raw_json_text): try: text_to_parse = raw_json_text.strip() if text_to_parse.startswith("```json"): text_to_parse = text_to_parse[7:] elif text_to_parse.startswith("```"): text_to_parse = text_to_parse[3:] if text_to_parse.endswith("```"): text_to_parse = text_to_parse[:-3] text_to_parse = text_to_parse.strip() data = json.loads(text_to_parse) header = data.get("header", {}) summary = data.get("summary", {}) items = data.get("items", []) currency = summary.get("currency", "") result = { "Invoice Number": header.get("invoice_no", ""), "PO Number": header.get("po_no", ""), "Invoice Date": header.get("invoice_date", ""), "Payment Terms": header.get("payment_terms", ""), "Due Date": header.get("due_date", ""), "Sender Name": header.get("sender_name", ""), "Sender Address": header.get("sender_addr", ""), "Tax ID": header.get("tax_id", ""), "Sender": { "Name": header.get("sender_name", ""), "Address": header.get("sender_addr", "") }, "Recipient Name": header.get("rcpt_name", ""), "Recipient Address": header.get("rcpt_addr", ""), "Recipient": { "Name": header.get("rcpt_name", ""), "Address": header.get("rcpt_addr", "") }, "Bank Details": { "bank_iban": header.get("bank_iban", ""), "bank_name": header.get("bank_name", ""), "bank_account_number": header.get("bank_acc_no", ""), "bank_routing": header.get("bank_routing", ""), "bank_swift": header.get("bank_swift", ""), "bank_acc_name": header.get("bank_acc_name", ""), "bank_branch": header.get("bank_branch", "") }, # All amounts kept as raw strings exactly as the model returned them "Subtotal": summary.get("subtotal", ""), "Tax Percentage": summary.get("tax_rate", ""), "Total Tax": summary.get("tax_amount", ""), "Discount Rate": summary.get("discount_rate", ""), "Total Discount Amount": summary.get("total_discount_amount", ""), "Total Amount": summary.get("total_amount", ""), "Currency": currency, "Itemized Data": [] } for item in items: result["Itemized Data"].append({ "Description": item.get("descriptions", ""), "Service Name": item.get("service_name", "") or item.get("Service_name", ""), "Service Start Date": item.get("service_start_date", ""), "Service End Date": item.get("service_end_date", ""), "SKU": item.get("SKU", ""), # All numeric item fields kept as raw strings "Quantity": str(item.get("quantity", "")), "Unit Price": str(item.get("unit_price", "")), "Amount": str(item.get("amount", "")), "Discount Rate Per Item": str(item.get("discount_rate_per_item", "")), "Discount Amount Per Item": str(item.get("discount_amount_per_item", "")), "Tax Rate Per Item": str(item.get("tax_rate_per_item", "")), "Tax Amount Per Item": str(item.get("tax_amount_per_item", "")), "Line Total": str(item.get("line_total", "")), }) return result except Exception as e: st.error(f"JSON parse error: {str(e)}") return None # ----------------------------- # Mapping logic — flatten for rows/CSV/Excel # ----------------------------- def flatten_invoice_to_rows(invoice_data) -> list: EXPECTED_BANK_FIELDS = [ "bank_name", "bank_account_number", "bank_acc_name", "bank_iban", "bank_swift", "bank_routing", "bank_branch" ] def fmt(value): if value is None or str(value).strip() == "": return "NA" return str(value).strip() rows = [] invoice_data = invoice_data or {} line_items = invoice_data.get("Itemized Data", []) or [] bank_details = {} nested = invoice_data.get("Bank Details", {}) or {} if isinstance(nested, dict): for k, v in nested.items(): key_name = k if str(k).startswith("bank_") else f"bank_{k}" bank_details[key_name] = v for k, v in invoice_data.items(): if isinstance(k, str) and k.lower().startswith("bank_"): bank_details[k] = v for f in EXPECTED_BANK_FIELDS: bank_details.setdefault(f, "") def base_invoice_info(): return { "Invoice Number": fmt(invoice_data.get("Invoice Number", "")), "PO Number": fmt(invoice_data.get("PO Number", "")), "Invoice Date": fmt(invoice_data.get("Invoice Date", "")), "Payment Terms": fmt(invoice_data.get("Payment Terms", "")), "Due Date": fmt(invoice_data.get("Due Date", "")), "Currency": fmt(invoice_data.get("Currency", "")), "Tax ID": fmt(invoice_data.get("Tax ID", "")), # Raw string values — exactly as extracted "Subtotal": invoice_data.get("Subtotal", ""), "Tax Percentage": invoice_data.get("Tax Percentage", ""), "Total Tax": invoice_data.get("Total Tax", ""), "Discount Rate": invoice_data.get("Discount Rate", ""), "Total Discount Amount": invoice_data.get("Total Discount Amount", ""), "Total Amount": invoice_data.get("Total Amount", ""), "Sender Name": fmt(invoice_data.get("Sender Name", "") or (invoice_data.get("Sender", {}) or {}).get("Name", "")), "Sender Address": fmt(invoice_data.get("Sender Address", "") or (invoice_data.get("Sender", {}) or {}).get("Address", "")), "Recipient Name": fmt(invoice_data.get("Recipient Name", "") or (invoice_data.get("Recipient", {}) or {}).get("Name", "")), "Recipient Address": fmt(invoice_data.get("Recipient Address", "") or (invoice_data.get("Recipient", {}) or {}).get("Address", "")), } if not line_items: row = base_invoice_info() for k in EXPECTED_BANK_FIELDS: row[k] = fmt(bank_details.get(k, "")) row.update({ "Item Description": "NA", "Service Name": "NA", "Service Start Date": "NA", "Service End Date": "NA", "Item Quantity": "", "Item Unit Price": "", "Item Amount": "", "Discount Rate Per Item": "", "Discount Amount Per Item": "", "Tax Rate Per Item": "", "Tax Amount Per Item": "", "Item Line Total": "", "IO Number/Cost Centre": "NA", }) rows.append(row) return rows for item in line_items: row = base_invoice_info() for k in EXPECTED_BANK_FIELDS: row[k] = fmt(bank_details.get(k, "")) row.update({ "Item Description": fmt(item.get("Description", "") if isinstance(item, dict) else ""), "Service Name": fmt(item.get("Service Name", "") if isinstance(item, dict) else ""), "Service Start Date": fmt(item.get("Service Start Date", "") if isinstance(item, dict) else ""), "Service End Date": fmt(item.get("Service End Date", "") if isinstance(item, dict) else ""), # Raw strings "Item Quantity": item.get("Quantity", "") if isinstance(item, dict) else "", "Item Unit Price": item.get("Unit Price", "") if isinstance(item, dict) else "", "Item Amount": item.get("Amount", "") if isinstance(item, dict) else "", "Discount Rate Per Item": item.get("Discount Rate Per Item", "") if isinstance(item, dict) else "", "Discount Amount Per Item": item.get("Discount Amount Per Item", "") if isinstance(item, dict) else "", "Tax Rate Per Item": item.get("Tax Rate Per Item", "") if isinstance(item, dict) else "", "Tax Amount Per Item": item.get("Tax Amount Per Item", "") if isinstance(item, dict) else "", "Item Line Total": item.get("Line Total", "") if isinstance(item, dict) else "", "IO Number/Cost Centre": fmt(item.get("IO Number/Cost Centre", "") if isinstance(item, dict) else ""), }) rows.append(row) return rows # ----------------------------- # JSONL conversion # ----------------------------- def build_page_filenames(file_name: str, num_pages: int) -> list[str]: """ Single page → ["PI-2025-Dec-040.png"] Multi page → ["PI-2025-Dec-051_page1.png", "PI-2025-Dec-051_page2.png"] Always a list, always .png, single page keeps original name with no suffix. """ stem = Path(file_name).stem if num_pages == 1: return [f"{stem}.png"] return [f"{stem}_page{i+1}.png" for i in range(num_pages)] def convert_to_jsonl_record(edited_data, file_name="", num_pages: int = 1): ed = edited_data or {} currency = ed.get("Currency", "") items = ed.get("Itemized Data", []) or [] bank = ed.get("Bank Details", {}) or {} header = { "invoice_no": ed.get("Invoice Number", ""), "po_no": ed.get("PO Number", ""), "invoice_date": ed.get("Invoice Date", ""), "payment_terms": ed.get("Payment Terms", ""), "due_date": ed.get("Due Date", ""), "sender_name": ed.get("Sender Name", "") or (ed.get("Sender", {}) or {}).get("Name", ""), "sender_addr": ed.get("Sender Address", "") or (ed.get("Sender", {}) or {}).get("Address", ""), "tax_id": ed.get("Tax ID", ""), "rcpt_name": ed.get("Recipient Name", "") or (ed.get("Recipient", {}) or {}).get("Name", ""), "rcpt_addr": ed.get("Recipient Address", "") or (ed.get("Recipient", {}) or {}).get("Address", ""), "bank_iban": bank.get("bank_iban", ""), "bank_name": bank.get("bank_name", ""), "bank_acc_no": bank.get("bank_account_number", ""), "bank_routing": bank.get("bank_routing", ""), "bank_swift": bank.get("bank_swift", ""), "bank_acc_name": bank.get("bank_acc_name", ""), "bank_branch": bank.get("bank_branch", ""), } jsonl_items = [] for it in items: if not isinstance(it, dict): continue jsonl_items.append({ "service_name": it.get("Service Name", ""), "service_start_date": it.get("Service Start Date", ""), "service_end_date": it.get("Service End Date", ""), "descriptions": it.get("Description", ""), "SKU": it.get("SKU", ""), "quantity": it.get("Quantity", ""), "unit_price": it.get("Unit Price", ""), "amount": it.get("Amount", ""), "discount_rate_per_item": it.get("Discount Rate Per Item", ""), "discount_amount_per_item": it.get("Discount Amount Per Item", ""), "tax_rate_per_item": it.get("Tax Rate Per Item", ""), "tax_amount_per_item": it.get("Tax Amount Per Item", ""), "line_total": it.get("Line Total", ""), }) summary = { "subtotal": ed.get("Subtotal", ""), "tax_rate": ed.get("Tax Percentage", ""), "tax_amount": ed.get("Total Tax", ""), "discount_rate": ed.get("Discount Rate", ""), "total_discount_amount": ed.get("Total Discount Amount", ""), "total_amount": ed.get("Total Amount", ""), "currency": currency, } return { "file_name": build_page_filenames(file_name, num_pages), "gt_parse": { "header": header, "items": jsonl_items, "summary": summary, } } def build_excel_bytes(df): buf = BytesIO() with pd.ExcelWriter(buf, engine="openpyxl") as writer: df.to_excel(writer, index=False, sheet_name="Invoices") ws = writer.sheets["Invoices"] for col_cells in ws.columns: max_len = 0 col_letter = col_cells[0].column_letter for cell in col_cells: try: if cell.value: max_len = max(max_len, len(str(cell.value))) except Exception: pass ws.column_dimensions[col_letter].width = min(max_len + 3, 50) return buf.getvalue() # ============================================================================= # Extraction helper — inference + parse + computation post-processing # ============================================================================= def _run_extraction(pages: List[Image.Image]) -> tuple[str | None, dict]: raw_json = run_inference_vllm(pages) if not raw_json: return None, {} parsed_data = parse_vllm_json(raw_json) if not parsed_data: st.warning("Failed to parse JSON response from model.") return raw_json, {} if isinstance(parsed_data, dict): parsed_data = postprocess_invoice_data(parsed_data) return raw_json, parsed_data if isinstance(parsed_data, dict) else {} # ----------------------------- # Session scaffolding # ----------------------------- if "batch_results" not in st.session_state: st.session_state.batch_results = {} if "current_file_hash" not in st.session_state: st.session_state.current_file_hash = None if "is_processing_batch" not in st.session_state: st.session_state.is_processing_batch = False if "confirm_back" not in st.session_state: st.session_state.confirm_back = False frame_left, frame_right = st.columns([1, 1], vertical_alignment="top") # ============================================================================= # UPLOAD & BATCH PROCESSING # ============================================================================= if not st.session_state.is_processing_batch and len(st.session_state.batch_results) == 0: with frame_left: st.header("📤 Upload Invoices") uploaded_files = st.file_uploader( "Upload invoice images or PDFs — all pages of a PDF are sent in one request", type=["png", "jpg", "jpeg", "pdf"], accept_multiple_files=True ) if uploaded_files: st.session_state.is_processing_batch = True progress_bar = st.progress(0) status_text = st.empty() for idx, uploaded_file in enumerate(uploaded_files): status_text.text(f"Processing {idx+1}/{len(uploaded_files)}: {uploaded_file.name}") uploaded_bytes = uploaded_file.read() file_hash = hashlib.sha256(uploaded_bytes).hexdigest() if file_hash in st.session_state.batch_results: progress_bar.progress((idx + 1) / len(uploaded_files)) continue pages = [] is_pdf = ( uploaded_file.name.lower().endswith(".pdf") or (hasattr(uploaded_file, "type") and uploaded_file.type == "application/pdf") ) if is_pdf: if convert_from_bytes is None: st.warning(f"PDF {uploaded_file.name} could not be rendered (pdf2image/poppler missing).") progress_bar.progress((idx + 1) / len(uploaded_files)) continue try: pdf_pages = convert_from_bytes(uploaded_bytes, dpi=300) pages = [p.convert("RGB") for p in pdf_pages] if not pages: st.warning(f"PDF {uploaded_file.name} has no pages.") progress_bar.progress((idx + 1) / len(uploaded_files)) continue st.info(f"{uploaded_file.name}: {len(pages)} page(s) detected — all sent in one request.") except Exception as exc: st.warning(f"Could not render PDF {uploaded_file.name}: {exc}") progress_bar.progress((idx + 1) / len(uploaded_files)) continue else: try: img = Image.open(BytesIO(uploaded_bytes)).convert("RGB") pages.append(img) except Exception: st.warning(f"Failed to open {uploaded_file.name}.") progress_bar.progress((idx + 1) / len(uploaded_files)) continue if not pages: progress_bar.progress((idx + 1) / len(uploaded_files)) continue raw_json, safe_mapped = _run_extraction(pages) if raw_json is None: st.warning(f"No response from vLLM for {uploaded_file.name}") st.session_state.batch_results[file_hash] = { "file_name": uploaded_file.name, "pages": pages, "current_page": 0, "raw_pred": raw_json, "mapped_data": safe_mapped, "edited_data": safe_mapped.copy() } progress_bar.progress((idx + 1) / len(uploaded_files)) status_text.text("✅ All files processed!") st.session_state.is_processing_batch = False st.rerun() with frame_right: st.caption("Preview & editor will appear here after extraction.") # ============================================================================= # REVIEW & EDIT # ============================================================================= elif len(st.session_state.batch_results) > 0: with frame_left: all_rows = [] for file_hash, result in st.session_state.batch_results.items(): rows = flatten_invoice_to_rows(result["edited_data"]) for r in rows: r["Source File"] = result.get("file_name", file_hash) all_rows.extend(rows) if all_rows: full_df = pd.DataFrame(all_rows) cols = list(full_df.columns) if "Source File" in cols: cols = ["Source File"] + [c for c in cols if c != "Source File"] full_df = full_df[cols] dl_cols = st.columns(3) with dl_cols[0]: st.download_button("📦 All CSV", full_df.to_csv(index=False).encode("utf-8"), file_name="all_extracted_invoices.csv", mime="text/csv", key="download_all_csv") with dl_cols[1]: st.download_button("📦 All Excel", build_excel_bytes(full_df), file_name="all_extracted_invoices.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key="download_all_xlsx") with dl_cols[2]: jsonl_lines = [ json.dumps( convert_to_jsonl_record( res["edited_data"], res.get("file_name", ""), len(res.get("pages", [])) ), ensure_ascii=False ) for res in st.session_state.batch_results.values() ] st.download_button("📦 All JSONL", "\n".join(jsonl_lines).encode("utf-8"), file_name="all_extracted_invoices.jsonl", mime="application/jsonl", key="download_all_jsonl") with frame_right: if st.button("⬅️ Back to Upload"): st.session_state[f"confirm_back"] = True if st.session_state.get("confirm_back"): st.warning("⚠️ All extracted data will be lost. Are you sure?") col_yes, col_no = st.columns(2) with col_yes: if st.button("✅ Yes, go back"): st.session_state.batch_results.clear() st.session_state.current_file_hash = None st.session_state.is_processing_batch = False st.session_state.confirm_back = False st.rerun() with col_no: if st.button("❌ Cancel"): st.session_state.confirm_back = False st.rerun() with frame_left: file_options = {f"{v['file_name']} ({k[:6]})": k for k, v in st.session_state.batch_results.items()} selected_display = st.selectbox("Select invoice to view/edit:", options=list(file_options.keys()), index=0, key="file_selector") selected_hash = file_options[selected_display] if st.session_state.current_file_hash != selected_hash: if st.session_state.current_file_hash is not None: old_hash = st.session_state.current_file_hash for k in [k for k in st.session_state.keys() if k.endswith(f"_{old_hash}")]: del st.session_state[k] st.session_state.current_file_hash = selected_hash current = st.session_state.batch_results[selected_hash] pages = current["pages"] num_pages = len(pages) cur_page_idx = current.get("current_page", 0) form_data = current["edited_data"] bank = form_data.get("Bank Details", {}) if isinstance(form_data.get("Bank Details", {}), dict) else {} form_currency = form_data.get("Currency", "") # State defaults — all monetary/percentage values stored as strings state_defaults = { f"Invoice Number_{selected_hash}": form_data.get("Invoice Number", ""), f"PO Number_{selected_hash}": form_data.get("PO Number", ""), f"Payment Terms_{selected_hash}": form_data.get("Payment Terms", ""), f"Tax ID_{selected_hash}": form_data.get("Tax ID", ""), f"Currency_{selected_hash}": form_data.get("Currency", "USD") or "USD", f"Currency_Custom_{selected_hash}": form_data.get("Currency", "") if form_data.get("Currency") not in ["USD","EUR","GBP","INR"] else "", f"Subtotal_{selected_hash}": str(form_data.get("Subtotal", "")), f"Tax Percentage_{selected_hash}": str(form_data.get("Tax Percentage", "")), f"Total Tax_{selected_hash}": str(form_data.get("Total Tax", "")), f"Discount Rate_{selected_hash}": str(form_data.get("Discount Rate", "")), f"Total Discount Amount_{selected_hash}": str(form_data.get("Total Discount Amount", "")), f"Total Amount_{selected_hash}": str(form_data.get("Total Amount", "")), f"Sender Name_{selected_hash}": form_data.get("Sender Name", ""), f"Sender Address_{selected_hash}": form_data.get("Sender Address", ""), f"Recipient Name_{selected_hash}": form_data.get("Recipient Name", ""), f"Recipient Address_{selected_hash}": form_data.get("Recipient Address", ""), f"Bank_bank_name_{selected_hash}": bank.get("bank_name", ""), f"Bank_bank_account_number_{selected_hash}": bank.get("bank_account_number", "") or bank.get("bank_acc_no", ""), f"Bank_bank_acc_name_{selected_hash}": bank.get("bank_acc_name", "") or bank.get("bank_account_holder", ""), f"Bank_bank_iban_{selected_hash}": bank.get("bank_iban", ""), f"Bank_bank_swift_{selected_hash}": bank.get("bank_swift", ""), f"Bank_bank_routing_{selected_hash}": bank.get("bank_routing", ""), f"Bank_bank_branch_{selected_hash}": bank.get("bank_branch", ""), } for key, default in state_defaults.items(): if key not in st.session_state: st.session_state[key] = default if f"Invoice Date_{selected_hash}" not in st.session_state: st.session_state[f"Invoice Date_{selected_hash}"] = parse_date_to_object(form_data.get("Invoice Date", ""), form_currency) if f"Due Date_{selected_hash}" not in st.session_state: st.session_state[f"Due Date_{selected_hash}"] = parse_date_to_object(form_data.get("Due Date", ""), form_currency) # Raw strings from model — used for ground truth output, never overwritten by date picker if f"Invoice Date_raw_{selected_hash}" not in st.session_state: st.session_state[f"Invoice Date_raw_{selected_hash}"] = form_data.get("Invoice Date", "") if f"Due Date_raw_{selected_hash}" not in st.session_state: st.session_state[f"Due Date_raw_{selected_hash}"] = form_data.get("Due Date", "") # ========================================================================= # LEFT COLUMN — image preview # ========================================================================= with frame_left: st.image(pages[cur_page_idx], caption=f"{current['file_name']} — Page {cur_page_idx + 1} of {num_pages}", use_container_width=True) if num_pages > 1: nav_cols = st.columns([1, 2, 1]) with nav_cols[0]: if st.button("⬅️ Prev", disabled=(cur_page_idx == 0), key=f"prev_page_{selected_hash}"): st.session_state.batch_results[selected_hash]["current_page"] = cur_page_idx - 1 st.rerun() with nav_cols[1]: st.markdown(f"

Page {cur_page_idx + 1} / {num_pages}

", unsafe_allow_html=True) with nav_cols[2]: if st.button("Next ➡️", disabled=(cur_page_idx == num_pages - 1), key=f"next_page_{selected_hash}"): st.session_state.batch_results[selected_hash]["current_page"] = cur_page_idx + 1 st.rerun() st.write(f"**File Hash:** `{selected_hash[:8]}…` | **Pages:** {num_pages}") col_a, col_b = st.columns(2) with col_a: if st.button("🔁 Re-Run (All Pages)", key=f"rerun_all_{selected_hash}", help=f"Re-send all {num_pages} page(s) in one request"): with st.spinner(f"Re-running inference on all {num_pages} page(s)…"): try: raw_json, safe_mapped = _run_extraction(pages) if raw_json is None: st.error("No response from vLLM.") else: st.session_state.batch_results[selected_hash]["raw_pred"] = raw_json st.session_state.batch_results[selected_hash]["mapped_data"] = safe_mapped st.session_state.batch_results[selected_hash]["edited_data"] = safe_mapped.copy() for k in [k for k in st.session_state.keys() if k.endswith(f"_{selected_hash}")]: del st.session_state[k] st.success("✅ Re-run complete (all pages)") st.rerun() except Exception as e: st.error(f"Re-run failed: {e}") with col_b: if st.button(f"🔍 Extract Page {cur_page_idx + 1} Only", key=f"extract_page_{selected_hash}", help="Send only the currently displayed page"): with st.spinner(f"Running inference on page {cur_page_idx + 1} only…"): try: raw_json, safe_mapped = _run_extraction([pages[cur_page_idx]]) if raw_json is None: st.error("No response from vLLM.") else: st.session_state.batch_results[selected_hash]["raw_pred"] = raw_json st.session_state.batch_results[selected_hash]["mapped_data"] = safe_mapped st.session_state.batch_results[selected_hash]["edited_data"] = safe_mapped.copy() for k in [k for k in st.session_state.keys() if k.endswith(f"_{selected_hash}")]: del st.session_state[k] st.success(f"✅ Page {cur_page_idx + 1} extracted!") st.rerun() except Exception as e: st.error(f"Extraction failed: {e}") with st.expander("🔍 Show raw model output"): raw_pred = current.get("raw_pred") if raw_pred is None: st.warning("No raw output available.") else: st.code(str(raw_pred), language="json") post_notes = (current.get("edited_data", {}) or {}).get("Postprocessing Notes", {}) or {} risk_flags = post_notes.get("risk_flags", []) or [] computed_fields = post_notes.get("computed_fields", []) or [] if risk_flags: st.warning(f"⚠️ Post-processing found {len(risk_flags)} reconciliation risk(s).") with st.expander("🧮 Post-processing details"): st.write(f"Computed fields: {len(computed_fields)}") if computed_fields: st.code("\n".join(computed_fields[:80])) if len(computed_fields) > 80: st.caption(f"Showing first 80 of {len(computed_fields)} computed fields.") if risk_flags: st.write("Risk flags:") st.code("\n".join(risk_flags)) elif not computed_fields: st.caption("No post-processing changes were required.") # ========================================================================= # RIGHT COLUMN — editable form # ========================================================================= with frame_right: st.subheader(f"Editable Invoice: {current['file_name']}") with st.form(key=f"edit_form_{selected_hash}", clear_on_submit=False): tabs = st.tabs(["Invoice Details", "Sender/Recipient", "Bank Details", "Line Items"]) with tabs[0]: st.text_input("Invoice Number", key=f"Invoice Number_{selected_hash}") st.text_input("PO Number", key=f"PO Number_{selected_hash}") st.write("**Invoice Date:**") raw_invoice_date = form_data.get("Invoice Date", "") if raw_invoice_date: st.info(f"📅 Model extracted: {raw_invoice_date}") st.date_input("Select date:", key=f"Invoice Date_{selected_hash}", format="DD/MM/YYYY", label_visibility="collapsed") st.text_input("Payment Terms", key=f"Payment Terms_{selected_hash}") st.write("**Due Date:**") raw_due_date = form_data.get("Due Date", "") if raw_due_date: st.info(f"📅 Model extracted: {raw_due_date}") st.date_input("Select date:", key=f"Due Date_{selected_hash}", format="DD/MM/YYYY", label_visibility="collapsed") st.text_input("Tax ID / VAT Number", key=f"Tax ID_{selected_hash}") curr_options = ["USD", "EUR", "GBP", "INR", "Other"] if st.session_state[f"Currency_{selected_hash}"] not in curr_options: st.session_state[f"Currency_{selected_hash}"] = "Other" st.selectbox("Currency", options=curr_options, key=f"Currency_{selected_hash}") if st.session_state.get(f"Currency_{selected_hash}") == "Other": st.text_input("Specify Currency", key=f"Currency_Custom_{selected_hash}") # Text inputs — raw strings, no conversion st.text_input("Subtotal", key=f"Subtotal_{selected_hash}") st.text_input("Tax %", key=f"Tax Percentage_{selected_hash}") st.text_input("Total Tax", key=f"Total Tax_{selected_hash}") st.text_input("Discount Rate %", key=f"Discount Rate_{selected_hash}") st.text_input("Total Discount Amount", key=f"Total Discount Amount_{selected_hash}") st.text_input("Total Amount", key=f"Total Amount_{selected_hash}") with tabs[1]: st.text_input("Sender Name", key=f"Sender Name_{selected_hash}") st.text_area ("Sender Address", key=f"Sender Address_{selected_hash}", height=80) st.text_input("Recipient Name", key=f"Recipient Name_{selected_hash}") st.text_area ("Recipient Address", key=f"Recipient Address_{selected_hash}", height=80) with tabs[2]: st.text_input("Bank Name", key=f"Bank_bank_name_{selected_hash}") st.text_input("Account Number", key=f"Bank_bank_account_number_{selected_hash}") st.text_input("Account Name", key=f"Bank_bank_acc_name_{selected_hash}") st.text_input("IBAN", key=f"Bank_bank_iban_{selected_hash}") st.text_input("SWIFT", key=f"Bank_bank_swift_{selected_hash}") st.text_input("Routing", key=f"Bank_bank_routing_{selected_hash}") st.text_input("Branch", key=f"Bank_bank_branch_{selected_hash}") with tabs[3]: items_state_key = f"items_df_{selected_hash}" if items_state_key not in st.session_state: item_rows = form_data.get("Itemized Data", []) or [] normalized = [] for it in item_rows: if not isinstance(it, dict): it = {} normalized.append({ "Description": it.get("Description", it.get("Item Description", "")), "Service Name": it.get("Service Name", ""), "Service Start Date": it.get("Service Start Date", ""), "Service End Date": it.get("Service End Date", ""), # All numeric columns as strings "Quantity": str(it.get("Quantity", it.get("Item Quantity", ""))), "Unit Price": str(it.get("Unit Price", it.get("Item Unit Price", ""))), "Amount": str(it.get("Amount", it.get("Item Amount", ""))), "Discount Rate Per Item": str(it.get("Discount Rate Per Item", "")), "Discount Amount Per Item": str(it.get("Discount Amount Per Item", "")), "Tax Rate Per Item": str(it.get("Tax Rate Per Item", "")), "Tax Amount Per Item": str(it.get("Tax Amount Per Item", "")), "IO Number/Cost Centre": it.get("IO Number/Cost Centre", ""), "Line Total": str(it.get("Line Total", it.get("Item Line Total", ""))), }) st.session_state[items_state_key] = ( pd.DataFrame(normalized) if normalized else pd.DataFrame(columns=[ "Description", "Service Name", "Service Start Date", "Service End Date", "Quantity", "Unit Price", "Amount", "Discount Rate Per Item", "Discount Amount Per Item", "Tax Rate Per Item", "Tax Amount Per Item", "IO Number/Cost Centre", "Line Total" ]) ) items_df = st.session_state[items_state_key] # All columns are TextColumn — no numeric conversion column_config = { "Description": st.column_config.TextColumn("Description", width="large"), "Service Name": st.column_config.TextColumn("Service Name", width="medium"), "Service Start Date": st.column_config.TextColumn("Svc Start", width="small"), "Service End Date": st.column_config.TextColumn("Svc End", width="small"), "Quantity": st.column_config.TextColumn("Qty", width="small"), "Unit Price": st.column_config.TextColumn("Unit Price", width="small"), "Amount": st.column_config.TextColumn("Amount", width="small"), "Discount Rate Per Item": st.column_config.TextColumn("Disc %", width="small"), "Discount Amount Per Item": st.column_config.TextColumn("Disc Amt", width="small"), "Tax Rate Per Item": st.column_config.TextColumn("Tax %", width="small"), "Tax Amount Per Item": st.column_config.TextColumn("Tax Amt", width="small"), "IO Number/Cost Centre": st.column_config.TextColumn("IO/Cost Centre", width="medium"), "Line Total": st.column_config.TextColumn("Line Total", width="small"), } edited_df = st.data_editor( items_df, num_rows="dynamic", key=f"items_editor_{selected_hash}", use_container_width=True, height=DATA_EDITOR_HEIGHT, column_config=column_config, ) st.session_state[items_state_key] = edited_df saved = st.form_submit_button("💾 Save All Edits") currency = st.session_state.get(f"Currency_{selected_hash}", "USD") if currency == "Other": currency = st.session_state.get(f"Currency_Custom_{selected_hash}", "") items_state_key = f"items_df_{selected_hash}" current_items_df = st.session_state.get(items_state_key, pd.DataFrame()) line_items_list = current_items_df.to_dict("records") def _build_updated_dict(): return { "Invoice Number": st.session_state.get(f"Invoice Number_{selected_hash}", ""), "PO Number": st.session_state.get(f"PO Number_{selected_hash}", ""), # Raw model string — not the date picker's reformatted value "Invoice Date": st.session_state.get(f"Invoice Date_raw_{selected_hash}", ""), "Payment Terms": st.session_state.get(f"Payment Terms_{selected_hash}", ""), "Due Date": st.session_state.get(f"Due Date_raw_{selected_hash}", ""), "Tax ID": st.session_state.get(f"Tax ID_{selected_hash}", ""), "Currency": currency, # Raw strings preserved as-is "Subtotal": st.session_state.get(f"Subtotal_{selected_hash}", ""), "Tax Percentage": st.session_state.get(f"Tax Percentage_{selected_hash}", ""), "Total Tax": st.session_state.get(f"Total Tax_{selected_hash}", ""), "Discount Rate": st.session_state.get(f"Discount Rate_{selected_hash}", ""), "Total Discount Amount": st.session_state.get(f"Total Discount Amount_{selected_hash}", ""), "Total Amount": st.session_state.get(f"Total Amount_{selected_hash}", ""), "Sender Name": st.session_state.get(f"Sender Name_{selected_hash}", ""), "Sender Address": st.session_state.get(f"Sender Address_{selected_hash}", ""), "Recipient Name": st.session_state.get(f"Recipient Name_{selected_hash}", ""), "Recipient Address": st.session_state.get(f"Recipient Address_{selected_hash}", ""), "Bank Details": { "bank_name": st.session_state.get(f"Bank_bank_name_{selected_hash}", ""), "bank_account_number": st.session_state.get(f"Bank_bank_account_number_{selected_hash}", ""), "bank_acc_name": st.session_state.get(f"Bank_bank_acc_name_{selected_hash}", ""), "bank_iban": st.session_state.get(f"Bank_bank_iban_{selected_hash}", ""), "bank_swift": st.session_state.get(f"Bank_bank_swift_{selected_hash}", ""), "bank_routing": st.session_state.get(f"Bank_bank_routing_{selected_hash}", ""), "bank_branch": st.session_state.get(f"Bank_bank_branch_{selected_hash}", ""), }, "Itemized Data": line_items_list, "Sender": {"Name": st.session_state.get(f"Sender Name_{selected_hash}", ""), "Address": st.session_state.get(f"Sender Address_{selected_hash}", "")}, "Recipient": {"Name": st.session_state.get(f"Recipient Name_{selected_hash}", ""), "Address": st.session_state.get(f"Recipient Address_{selected_hash}", "")}, } if saved: updated = _build_updated_dict() # Persist user-edited date picker values back as the date strings invoice_date = st.session_state.get(f"Invoice Date_{selected_hash}") due_date = st.session_state.get(f"Due Date_{selected_hash}") if invoice_date is not None: try: updated["Invoice Date"] = invoice_date.strftime("%d-%b-%Y") st.session_state[f"Invoice Date_raw_{selected_hash}"] = invoice_date.strftime("%d-%b-%Y") except (AttributeError, ValueError): pass if due_date is not None: try: updated["Due Date"] = due_date.strftime("%d-%b-%Y") st.session_state[f"Due Date_raw_{selected_hash}"] = due_date.strftime("%d-%b-%Y") except (AttributeError, ValueError): pass updated = postprocess_invoice_data(updated) st.session_state.batch_results[selected_hash]["edited_data"] = updated for k in [k for k in list(st.session_state.keys()) if k.endswith(f"_{selected_hash}")]: del st.session_state[k] st.success("✅ Saved") st.rerun() download_data = postprocess_invoice_data(_build_updated_dict()) rows = flatten_invoice_to_rows(download_data) file_df = pd.DataFrame(rows) file_stem = Path(current["file_name"]).stem per_dl_cols = st.columns(3) with per_dl_cols[0]: st.download_button("📥 CSV", file_df.to_csv(index=False).encode("utf-8"), file_name=f"{file_stem}_full.csv", mime="text/csv", key=f"dl_csv_{selected_hash}") with per_dl_cols[1]: st.download_button("📥 Excel", build_excel_bytes(file_df), file_name=f"{file_stem}_full.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", key=f"dl_xlsx_{selected_hash}") with per_dl_cols[2]: record = convert_to_jsonl_record(download_data, current["file_name"], num_pages) jsonl_one = json.dumps(record, ensure_ascii=False).encode("utf-8") st.download_button("📥 JSONL", jsonl_one, file_name=f"{file_stem}.jsonl", mime="application/jsonl", key=f"dl_jsonl_{selected_hash}") # ============================================================================= # Processing placeholder # ============================================================================= elif st.session_state.is_processing_batch: with frame_left: st.info("⏳ Processing batch… Please wait.") st.progress(0) with frame_right: st.caption("Preview & editor will appear here after extraction.") else: with frame_left: st.caption("Ready when you are.") with frame_right: st.caption("Preview & editor will appear here after extraction.")