from datetime import date, datetime from decimal import Decimal, InvalidOperation def parse_date(value): if value in (None, ""): return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value try: return datetime.strptime(str(value), "%Y-%m-%d").date() except ValueError: return None def parse_decimal(value): if value in (None, ""): return None if isinstance(value, Decimal): return value try: return Decimal(str(value)) except (InvalidOperation, ValueError): return None def run_invoice_validation(*, document, extracted_fields, line_items, currency_seed_map, platform_configs=None): """ Business validation logic for: 7.7.1 Required fields + due-date derivation path 7.7.2 Currency membership 7.7.3 Math cross-check with currency-aware tolerance """ platform_configs = platform_configs or {} cap_minor_units = int( platform_configs.get("validation.solver_tolerance.cap_minor_units", 0) ) base_per_minor_unit = int( platform_configs.get("validation.solver_tolerance.base_per_minor_unit", 0) ) per_line_minor_units = int( platform_configs.get("validation.solver_tolerance.per_line_minor_units", 0) ) risk_flags = [] advisory_flags = [] skipped_steps = [] def add_flag(code, *, weight, detail=None): risk_flags.append( { "code": code, "severity": "advisory", "weight": weight, "detail": detail or {}, } ) invoice_no = getattr(extracted_fields, "invoice_no", None) invoice_date = parse_date(getattr(extracted_fields, "invoice_date", None)) total_amount = parse_decimal(getattr(extracted_fields, "total_amount", None)) subtotal = parse_decimal(getattr(extracted_fields, "subtotal", None)) sender_name = getattr(extracted_fields, "sender_name", None) currency = getattr(extracted_fields, "currency", None) due_date = parse_date(getattr(extracted_fields, "due_date", None)) payment_terms = getattr(extracted_fields, "payment_terms", None) missing_fields = [] def mark_missing(field_name): if field_name not in missing_fields: missing_fields.append(field_name) if not invoice_no: mark_missing("invoice_no") if invoice_date is None: mark_missing("invoice_date") if total_amount is None: mark_missing("total_amount") if not sender_name: mark_missing("sender_name") if not currency: mark_missing("currency") due_date_derivable = invoice_date is not None and payment_terms not in (None, "") if due_date is None and not due_date_derivable: mark_missing("due_date") if invoice_date is None: mark_missing("invoice_date") if payment_terms in (None, ""): mark_missing("payment_terms") if missing_fields: add_flag( "incomplete_fields", weight=0.25, detail={"missing_fields": missing_fields}, ) currency_key = str(currency).strip().upper() if currency not in (None, "") else None currency_row = currency_seed_map.get(currency_key) if currency_key else None if currency_row is None: add_flag( "invalid_currency", weight=0.20, detail={ "currency": currency, "reason": "currency_missing_or_not_in_iso_currencies_seed", }, ) line_amounts = [] for li in line_items: amount = parse_decimal(getattr(li, "amount", None)) if amount is not None: line_amounts.append(amount) num_line_items = len(line_amounts) math_ran = False if subtotal is not None and num_line_items > 0: if currency_row is None: skipped_steps.append( {"step": "math_cross_check", "skip_reason": "currency_minor_unit_unknown"} ) else: minor_unit_value = parse_decimal(currency_row.get("minor_unit_value")) if minor_unit_value is None: skipped_steps.append( {"step": "math_cross_check", "skip_reason": "minor_unit_value_missing_in_seed"} ) else: math_ran = True line_sum = sum(line_amounts, Decimal("0")) tolerance_minor_units = min( cap_minor_units, base_per_minor_unit + per_line_minor_units * num_line_items, ) tolerance_decimal = Decimal(tolerance_minor_units) * minor_unit_value difference = abs(line_sum - subtotal) if difference > tolerance_decimal: add_flag( "math_error_line_items", weight=0.20, detail={ "subtotal": str(subtotal), "line_items_sum": str(line_sum), "difference": str(difference), "tolerance_minor_units": tolerance_minor_units, "tolerance_decimal": str(tolerance_decimal), "currency": currency_key, "minor_unit_value": str(minor_unit_value), "num_line_items": num_line_items, }, ) return { "risk_flags": risk_flags, "advisory_flags": advisory_flags, "skipped_steps": skipped_steps, "result": "flagged" if risk_flags else "clean", "missing_fields": missing_fields, "currency_checked": currency_row is not None, "math_checked": math_ran, }