from __future__ import annotations import re from typing import Any FIELD_KEYS = [ "vendor_name", "invoice_number", "invoice_date", "currency", "subtotal", "tax", "total", "po_id", "receipt_id", "bank_account", ] INVESTIGATION_ACTIONS = [ "zoom", "get_doc_crop", "ocr", "lookup_vendor", "lookup_vendor_history", "lookup_policy", "lookup_po", "lookup_receipt", "search_ledger", "inspect_email_thread", "compare_bank_account", ] INTERVENTION_ACTIONS = [ "request_callback_verification", "freeze_vendor_profile", "request_bank_change_approval_chain", "request_po_reconciliation", "request_additional_receipt_evidence", "route_to_procurement", "route_to_security", "flag_duplicate_cluster_review", "create_human_handoff", ] FINAL_ACTIONS = ["submit_decision"] ALLOWED_ACTIONS = INVESTIGATION_ACTIONS + INTERVENTION_ACTIONS + FINAL_ACTIONS ALLOWED_DECISIONS = ["PAY", "HOLD", "NEEDS_REVIEW", "ESCALATE_FRAUD"] DISCREPANCY_TYPES = [ "price_mismatch", "quantity_mismatch", "missing_receipt", "duplicate_po_reference", "invalid_invoice_date", "total_mismatch", "tax_id_mismatch", "partial_receipt_only", "missing_po", "receipt_date_mismatch", "bank_account_mismatch", "vendor_master_mismatch", ] FRAUD_TYPES = [ "bank_override_attempt", "vendor_name_spoof", "sender_domain_spoof", "duplicate_near_match", "approval_threshold_evasion", "urgent_payment_pressure", "callback_verification_failed", "callback_suspicious_confirm", "callback_dispute_confirmed", "vendor_account_takeover_suspected", "policy_bypass_attempt", "shared_bank_account", "coordinated_timing", ] POLICY_CHECK_KEYS = [ "three_way_match", "bank_change_verification", "duplicate_check", "approval_threshold_check", "human_review_required", "callback_required", ] OUTCOME_TYPES = [ "safe_payment_cleared", "unsafe_payment_released", "fraud_prevented", "manual_review_created", "false_positive_operational_delay", "policy_breach", ] ALL_REASON_CODES = sorted(set(DISCREPANCY_TYPES + FRAUD_TYPES)) REASON_CODE_ALIAS_MAP = { "bank mismatch": "bank_override_attempt", "bank account mismatch": "bank_override_attempt", "remittance override": "bank_override_attempt", "bank change attempt": "bank_override_attempt", "spoofed sender domain": "sender_domain_spoof", "sender spoof": "sender_domain_spoof", "domain spoof": "sender_domain_spoof", "duplicate invoice": "duplicate_near_match", "duplicate cluster": "duplicate_near_match", "invoice splitting": "approval_threshold_evasion", "threshold evasion": "approval_threshold_evasion", "approval threshold splitting": "approval_threshold_evasion", "policy bypass": "policy_bypass_attempt", "callback bypass": "policy_bypass_attempt", "shared bank": "shared_bank_account", "same bank account": "shared_bank_account", "shared remittance account": "shared_bank_account", "coordinated invoice timing": "coordinated_timing", "linked invoice timing": "coordinated_timing", "coordinated invoices": "coordinated_timing", } def normalize_text(value: Any) -> str: if value is None: return "" return " ".join(str(value).strip().lower().split()) def normalize_id(value: Any) -> str: text = normalize_text(value) return "".join(ch for ch in text if ch.isalnum()) def safe_float(value: Any) -> float: try: if isinstance(value, str): cleaned = ( value.replace(",", "") .replace("₹", "") .replace("$", "") .replace("€", "") .strip() ) return float(cleaned) return float(value) except (ValueError, TypeError, ArithmeticError): return 0.0 def numeric_match(a: Any, b: Any, tolerance: float = 0.01) -> bool: return abs(safe_float(a) - safe_float(b)) <= tolerance def fuzzy_numeric_similarity(a: Any, b: Any) -> float: a_num = safe_float(a) b_num = safe_float(b) denom = max(abs(a_num), abs(b_num), 1.0) diff = abs(a_num - b_num) / denom return max(0.0, 1.0 - diff) def prefix_domain(value: Any) -> str: text = normalize_text(value) if "@" in text: return text.split("@", 1)[-1] return text def bbox_iou(box_a: list[float] | None, box_b: list[float] | None) -> float: if not box_a or not box_b or len(box_a) != 4 or len(box_b) != 4: return 0.0 ax1, ay1, ax2, ay2 = box_a bx1, by1, bx2, by2 = box_b inter_x1 = max(ax1, bx1) inter_y1 = max(ay1, by1) inter_x2 = min(ax2, bx2) inter_y2 = min(ay2, by2) inter_w = max(0.0, inter_x2 - inter_x1) inter_h = max(0.0, inter_y2 - inter_y1) inter_area = inter_w * inter_h area_a = max(0.0, ax2 - ax1) * max(0.0, ay2 - ay1) area_b = max(0.0, bx2 - bx1) * max(0.0, by2 - by1) denom = area_a + area_b - inter_area if denom <= 0: return 0.0 return inter_area / denom def token_overlap(pred_token_ids: list[str] | None, gold_token_ids: list[str] | None) -> float: pred = set(pred_token_ids or []) gold = set(gold_token_ids or []) if not pred or not gold: return 0.0 return len(pred & gold) / len(gold) def list_unique_normalized(values: list[Any]) -> list[str]: seen: set[str] = set() output: list[str] = [] for value in values: norm = normalize_text(value) if not norm or norm in seen: continue seen.add(norm) output.append(norm) return output def normalize_reason_code(value: Any) -> str: text = normalize_text(value) if not text: return "" slug = re.sub(r"[^a-z0-9]+", "_", text).strip("_") allowed = {normalize_text(x) for x in ALL_REASON_CODES} if slug in allowed: return slug return REASON_CODE_ALIAS_MAP.get(text, "") def canonical_reason_codes(values: list[Any]) -> list[str]: seen: set[str] = set() output: list[str] = [] for value in values: canonical = normalize_reason_code(value) if not canonical or canonical in seen: continue seen.add(canonical) output.append(canonical) return output def is_intervention_action(action_type: str) -> bool: return normalize_text(action_type) in {normalize_text(x) for x in INTERVENTION_ACTIONS} def is_investigation_action(action_type: str) -> bool: return normalize_text(action_type) in {normalize_text(x) for x in INVESTIGATION_ACTIONS}