src/Invoice Validation Functions/completeness.py

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,
    }