"""Deterministic boundary between AI extraction and the public v1 result.""" from __future__ import annotations from decimal import Decimal, InvalidOperation from pathlib import Path import re from typing import Any EXTRACTOR_VERSION = "prompts-v1.2" MATH_TOLERANCE = Decimal("0.01") def build_canonical_result( *, extracted: dict[str, Any], invoice_id: str, filename: str, content_type: str, size_bytes: int, artifact_id: str | None, engine: str, ) -> dict[str, Any]: """Normalize legacy prompt output without retaining model reasoning or paths.""" header = _mapping(extracted.get("header")) supplier = _mapping(header.get("supplier")) invoice = _mapping(header.get("invoice")) invoice_date = _mapping(invoice.get("date")) totals = _mapping(extracted.get("totals")) items = [_normalize_item(item) for item in _list(extracted.get("items")) if isinstance(item, dict)] subtotal = _money(_field_value(totals.get("subtotal"))) tax = _money(_field_value(totals.get("gst"))) total = _money(_field_value(totals.get("grand_total"))) validations = _math_validations(items, subtotal, tax, total) warnings = _warnings( supplier_name=_text(_field_value(supplier.get("name"))), invoice_number=_text(_field_value(invoice.get("number"))), invoice_date=_text(_field_value(invoice_date.get("issue_date"))), items=items, total=total, validations=validations, ) document_type_confidence = _confidence(extracted.get("invoice_type")) field_confidences = _collect_confidences(extracted) overall_confidence = ( round(sum(field_confidences) / len(field_confidences), 4) if field_confidences else document_type_confidence ) return { "invoice_id": str(invoice_id), "status": "completed", "document": { "filename": _safe_filename(filename), "content_type": str(content_type or "application/octet-stream"), "size_bytes": max(0, int(size_bytes)), "artifact_id": str(artifact_id) if artifact_id else None, }, "supplier": { "name": _text(_field_value(supplier.get("name"))), "tax_id": _text(_field_value(supplier.get("abn") or supplier.get("tax_id"))), }, "invoice_number": _text(_field_value(invoice.get("number"))), "invoice_date": _text(_field_value(invoice_date.get("issue_date"))), "currency": _currency(extracted), "items": items, "subtotal": subtotal, "tax": tax, "total": total, "validations": validations, "warnings": warnings, "confidence": { "document_type": document_type_confidence, "overall": overall_confidence, }, "processing": { "engine": str(engine or "mini"), "extractor_version": EXTRACTOR_VERSION, }, } def build_canonical_result_v2(**kwargs) -> dict[str, Any]: """Build the in-development trust-aware result without changing v1.""" from apps.wp_invoices.services.quality_gate import ( evaluate_invoice_quality, normalize_invoice_type, ) from apps.wp_invoices.services.reconciliation import reconcile_invoice extracted = kwargs.get("extracted") or {} result = build_canonical_result(**kwargs) result["items"] = [_normalize_item_v2(item) for item in _list(extracted.get("items")) if isinstance(item, dict)] header = _mapping(extracted.get("header")) supplier = _mapping(header.get("supplier")) references = [ref for ref in _list(extracted.get("reference_candidates")) if isinstance(ref, dict)] adjustments = [_normalize_adjustment(value) for value in _list(extracted.get("adjustments")) if isinstance(value, dict)] tax_summary = _normalize_tax_summary(extracted.get("tax_summary"), result["tax"]) result["supplier"].update({ "legal_entity": _text(_field_value(supplier.get("legal_entity"))), "store": _text(_field_value(header.get("store"))), }) result.update({ "transaction_time": _text(_field_value(header.get("transaction_time"))), "payment_method": _text(_field_value(header.get("payment_method"))), "payment_reference": _text(_field_value(header.get("payment_reference"))), "receipt_reference": _best_reference(references), "reference_candidates": references, "item_count": len(result["items"]), "observed_item_line_count": _integer(extracted.get("observed_item_line_count")), "adjustments": adjustments, "tax_summary": tax_summary, "tax_inclusive": tax_summary["tax_inclusive"], "document_subtype": _text(_field_value(extracted.get("document_subtype"))), }) reconciliation = reconcile_invoice( items=result["items"], adjustments=adjustments, tax_summary=tax_summary, subtotal=result["subtotal"], total=result["total"], observed_item_line_count=result["observed_item_line_count"], ) adjustments = reconciliation["reconciled_adjustments"] result["adjustments"] = adjustments result["tax"] = tax_summary["tax_total"] quality = evaluate_invoice_quality( items=result["items"], subtotal=result["subtotal"], tax=result["tax"], total=result["total"], supplier_name=result["supplier"]["name"], invoice_date=result["invoice_date"], adjustments=adjustments, tax_inclusive=tax_summary["tax_inclusive"], reconciliation=reconciliation, ) return { **result, "result_contract_version": "2.0", "invoice_type": normalize_invoice_type(extracted.get("invoice_type")), "review_status": quality["review_status"], "quality": quality, "reconciliation": reconciliation, } def _normalize_item(item: dict[str, Any]) -> dict[str, Any]: confidences = [ _confidence(item.get(name)) for name in ("description", "qty", "unit_price", "gst_line", "line_total") if isinstance(item.get(name), dict) ] return { "description": _text(_field_value(item.get("description"))), "quantity": _number(_field_value(item.get("qty") or item.get("quantity"))), "unit_price": _money(_field_value(item.get("unit_price"))), "tax": _money(_field_value(item.get("gst_line") or item.get("tax"))), "line_total": _money(_field_value(item.get("line_total"))), "confidence": round(sum(confidences) / len(confidences), 4) if confidences else 0.0, } def _normalize_item_v2(item: dict[str, Any]) -> dict[str, Any]: from apps.wp_invoices.services.reconciliation import evidence normalized = _normalize_item(item) package = _mapping(item.get("package_size")) source_text = _text(item.get("source_text")) source_line = item.get("source_line") bounding_box = item.get("bounding_box") product_code = _product_code(source_text) product_code_evidence = { "raw_text": product_code, "source_line_id": None if source_line is None else str(source_line), "bbox": bounding_box if isinstance(bounding_box, list) else None, "confidence": _confidence(item.get("description")) if product_code else 0.0, "evidence_state": "observed_unverified" if product_code else "missing", } tax_code = _text(_field_value(item.get("tax_code"))) tax_code_evidence = { "raw_text": tax_code, "source_line_id": None if source_line is None else str(source_line), "bbox": bounding_box if isinstance(bounding_box, list) else None, "confidence": _confidence(item.get("tax_code")), "evidence_state": "observed_unverified" if tax_code else "missing", } return {**normalized, "product_code": product_code, "product_code_evidence": product_code_evidence, "transaction_unit": _text(_field_value(item.get("transaction_unit"))) or "each", "package_size": {"value": _number(package.get("value")), "unit": _text(package.get("unit"))} if package and package.get("value") is not None else None, "unit_price_basis": _text(_field_value(item.get("unit_price_basis"))), "tax_code": tax_code, "tax_code_evidence": tax_code_evidence, "source_text": source_text, "source_line": source_line, "bounding_box": bounding_box if isinstance(bounding_box, list) else None, "evidence": { "quantity": _field_evidence(item.get("qty"), source_text, source_line, bounding_box, evidence), "unit_price": _field_evidence(item.get("unit_price"), source_text, source_line, bounding_box, evidence), "tax": _field_evidence(item.get("gst_line"), source_text, source_line, bounding_box, evidence), "line_total": _field_evidence(item.get("line_total"), source_text, source_line, bounding_box, evidence), }} def _product_code(source_text: str | None) -> str | None: match = re.match(r"^\s*(\d{4,8})\b", str(source_text or "")) return match.group(1) if match else None def _normalize_adjustment(value: dict[str, Any]) -> dict[str, Any]: kind = str(value.get("type") or "other") return {"type": kind if kind in {"payment_surcharge", "discount", "rounding", "other"} else "other", "description": _text(value.get("description")), "amount": _money(value.get("amount")), "source_text": _text(value.get("source_text")), "confidence": _confidence(value)} def _normalize_tax_summary(value: Any, fallback_tax: float | None) -> dict[str, Any]: summary = _mapping(value) lines = [] for line in _list(summary.get("lines")): if isinstance(line, dict): lines.append({"tax_code": _text(line.get("tax_code")), "rate": _tax_rate(line.get("rate")), "net_amount": _money(line.get("net_amount")), "tax_amount": _money(line.get("tax_amount")), "evidence": {"status": "observed_unverified", "origin": "model_semantic_extraction"}}) tax_amounts = [line["tax_amount"] for line in lines] derived_total = round(sum(tax_amounts), 2) if lines and all(amount is not None for amount in tax_amounts) else None return {"tax_inclusive": summary.get("tax_inclusive") if isinstance(summary.get("tax_inclusive"), bool) else None, "tax_total": derived_total, "tax_total_origin": "derived_from_tax_summary_rows" if derived_total is not None else "not_available", "declared_tax_total": _money(_field_value(summary.get("tax_total"))) if summary else None, "declared_tax_total_evidence": {"status": "observed_unverified", "origin": "model_semantic_extraction"}, "lines": lines} def _tax_rate(value: Any) -> float | None: rate = _number(value) if rate is None: return None return round(rate / 100 if abs(rate) > 1 else rate, 6) def _best_reference(references: list[dict[str, Any]]) -> str | None: if not references: return None best = max(references, key=lambda value: float(value.get("confidence") or 0)) return _text(best.get("value")) def _field_evidence(field, source_text, source_line, bounding_box, factory): mapping = _mapping(field) raw = mapping.get("verbatim") origin = "derived" if raw is None and mapping.get("computed") is not None else "observed" value = factory(raw, source_text or source_line, mapping.get("confidence"), origin, bounding_box) value["status"] = "derived" if origin == "derived" else ("missing" if raw is None else "observed_unverified") return value def _integer(value: Any) -> int | None: try: return None if value is None else max(0, int(value)) except (TypeError, ValueError): return None def _math_validations(items, subtotal, tax, total) -> dict[str, bool | None]: item_totals = [item["line_total"] for item in items] items_check = None if items and subtotal is not None and all(value is not None for value in item_totals): items_check = _close(sum(Decimal(str(value)) for value in item_totals), Decimal(str(subtotal))) total_check = None if subtotal is not None and tax is not None and total is not None: total_check = _close(Decimal(str(subtotal)) + Decimal(str(tax)), Decimal(str(total))) available = [value for value in (items_check, total_check) if value is not None] return { "items_sum_matches_subtotal": items_check, "subtotal_plus_tax_matches_total": total_check, "mathematically_consistent": bool(available) and all(available), } def _warnings(*, supplier_name, invoice_number, invoice_date, items, total, validations): warnings: list[str] = [] for missing, message in ( (not supplier_name, "supplier_name_missing"), (not invoice_number, "invoice_number_missing"), (not invoice_date, "invoice_date_missing"), (not items, "line_items_missing"), (total is None, "total_missing"), ): if missing: warnings.append(message) if validations["items_sum_matches_subtotal"] is False: warnings.append("items_do_not_match_subtotal") if validations["subtotal_plus_tax_matches_total"] is False: warnings.append("subtotal_plus_tax_does_not_match_total") return warnings def _field_value(value: Any) -> Any: if not isinstance(value, dict): return value computed = value.get("computed") return computed if computed is not None else value.get("verbatim") def _money(value: Any) -> float | None: number = _decimal(value) return float(number.quantize(Decimal("0.01"))) if number is not None else None def _number(value: Any) -> float | None: number = _decimal(value) return float(number) if number is not None else None def _decimal(value: Any) -> Decimal | None: if value is None or isinstance(value, bool): return None cleaned = str(value).strip().replace(",", "") for token in ("AUD", "A$", "$"): cleaned = cleaned.replace(token, "") cleaned = cleaned.strip() if cleaned.startswith("(") and cleaned.endswith(")"): cleaned = f"-{cleaned[1:-1]}" try: return Decimal(cleaned) except (InvalidOperation, ValueError): return None def _confidence(value: Any) -> float: raw = value.get("confidence") if isinstance(value, dict) else 0 try: return min(1.0, max(0.0, float(raw or 0))) except (TypeError, ValueError): return 0.0 def _collect_confidences(value: Any) -> list[float]: if isinstance(value, dict): found = [] if "confidence" in value: found.append(_confidence(value)) for child in value.values(): found.extend(_collect_confidences(child)) return found if isinstance(value, list): found = [] for child in value: found.extend(_collect_confidences(child)) return found return [] def _currency(extracted: dict[str, Any]) -> str: currency = _text(_field_value(extracted.get("currency"))) or "AUD" normalized = currency.upper().strip() return normalized if len(normalized) == 3 and normalized.isalpha() else "AUD" def _safe_filename(filename: str) -> str: name = Path(str(filename or "invoice")).name.replace("\x00", "") return (name or "invoice")[:255] def _text(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None def _mapping(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def _list(value: Any) -> list[Any]: return value if isinstance(value, list) else [] def _close(left: Decimal, right: Decimal) -> bool: return abs(left - right) <= MATH_TOLERANCE