src/Invoice Validation Functions/Bank validation.py

import re
from datetime import date, datetime
from hashlib import sha256
from unicodedata import normalize
from decimal import Decimal

from rapidfuzz import fuzz


def run_bank_validation(

    *,

    document,

    extracted_fields,

    vendor_bank_accounts,

    resolved_vendor_id,

    platform_configs=None,

    name_match_fn=None,

):
    """

    Bank validation logic based on Cat 2 section 7.4.



    Expected document fields:

        document.id

        document.tenant_id



    Expected extracted_fields fields:

        invoice_date

        currency

        bank_iban

        bank_acc_no

        bank_swift

        bank_acc_name



    Expected vendor_bank_accounts row fields:

        tenant_id

        vendor_id

        iban_canonical

        acc_no_canonical

        swift_bic

        bank_acc_name

        currency_code

        bank_country

        is_primary

        effective_from

        effective_to

        deleted_at



    name_match_fn:

        Optional function for bank account name comparison.

        Should return something like:

            {"decision": "matched" | "suggestion" | "unmatched"}

    """

    platform_configs = platform_configs or {}

    iban_checksum_required = platform_configs.get(
        "validation.iban_checksum_required",
        True,
    )

    iban_accno_exempt_countries = set(
        platform_configs.get(
            "validation.iban_accno_substring_exempt_countries",
            ["LC", "MT"],
        )
    )

    high_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_high_threshold", "0.90")))
    medium_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_medium_threshold", "0.75")))
    suggestion_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_suggestion_threshold", "0.60")))

    def hash_prefix(value, length=8):
        if not value:
            return None
        return sha256(str(value).encode("utf-8")).hexdigest()[:length]

    def canonicalize_text(value):
        if value is None:
            return None
        value = normalize("NFKC", str(value))
        value = value.strip().upper()
        return value or None

    def canonicalize_iban(value):
        if value is None:
            return None
        value = normalize("NFKC", str(value))
        value = re.sub(r"[\s\-\.,/\\:;_]+", "", value)
        value = value.upper()
        return value or None

    def canonicalize_acc_no(value):
        if value is None:
            return None
        value = normalize("NFKC", str(value))
        value = re.sub(r"[\s\-\.,/\\:;_]+", "", value)
        value = value.upper()
        return value or None

    def canonicalize_swift(value):
        if value is None:
            return None
        value = normalize("NFKC", str(value))
        value = re.sub(r"\s+", "", value)
        value = value.upper()
        return value or None

    def iban_country(iban):
        if not iban or len(iban) < 2:
            return None
        return iban[:2]

    def iban_checksum_mod97_valid(iban):
        """

        Standard IBAN mod-97 validation.



        Move first 4 chars to end.

        Replace letters A=10 ... Z=35.

        Result mod 97 must equal 1.

        """
        if not iban or len(iban) < 4:
            return False

        rearranged = iban[4:] + iban[:4]
        numeric = ""

        for char in rearranged:
            if char.isdigit():
                numeric += char
            elif "A" <= char <= "Z":
                numeric += str(ord(char) - ord("A") + 10)
            else:
                return False

        remainder = 0
        for digit in numeric:
            remainder = (remainder * 10 + int(digit)) % 97

        return remainder == 1

    def iban_acc_no_consistent(iban, acc_no):
        """

        Basic consistency check:

        remove country + check digits from IBAN,

        then check whether account number appears inside BBAN.



        Leading zeros are ignored for comparison.

        """
        if not iban or not acc_no:
            return True

        country = iban_country(iban)

        if country in iban_accno_exempt_countries:
            return True

        bban = iban[4:]
        stripped_acc_no = acc_no.lstrip("0") or acc_no

        return stripped_acc_no in bban

    def is_effective(row, invoice_date):
        if getattr(row, "deleted_at", None):
            return False

        if invoice_date is None:
            return False

        effective_from = getattr(row, "effective_from", None)
        effective_to = getattr(row, "effective_to", None)

        if effective_from is not None and invoice_date < effective_from:
            return False

        if effective_to and invoice_date >= effective_to:
            return False

        return True

    def default_name_match(invoice_name, master_name):
        if not invoice_name or not master_name:
            return {"decision": "unmatched"}

        score = Decimal(str(fuzz.WRatio(invoice_name, master_name) / 100)).quantize(
            Decimal("0.0001")
        )

        if score >= high_threshold:
            return {"decision": "matched", "score_tier": "high"}
        if score >= medium_threshold:
            return {"decision": "matched", "score_tier": "medium"}
        if score >= suggestion_threshold:
            return {"decision": "suggestion", "score_tier": "low"}
        return {"decision": "unmatched", "score_tier": "low"}

    def bank_account_name_matches(invoice_name, master_name):
        matcher = name_match_fn or default_name_match

        result = matcher(
            query_name=invoice_name,
            candidate_pool=[master_name],
            context={"purpose": "bank_acc_name_check"},
        ) if name_match_fn else matcher(invoice_name, master_name)

        return result.get("decision") == "matched"

    def add_flag(code, *, field=None, weight=None, match_result="failed", invoice_value=None,

                 master_value=None, currency_invoice=None, currency_master=None,

                 primary_or_non_primary=None):
        flag = {
            "code": code,
            "field": field,
            "severity": "advisory",
        }

        if weight is not None:
            flag["weight"] = weight

        risk_flags.append(flag)

        audit_entries.append(
            {
                "field": field,
                "match_result": match_result,
                "invoice_hash_prefix": hash_prefix(invoice_value),
                "master_hash_prefix": hash_prefix(master_value),
                "currency_invoice": currency_invoice,
                "currency_master": currency_master,
                "primary_or_non_primary": primary_or_non_primary,
            }
        )

    risk_flags = []
    advisory_flags = []
    skipped_steps = []
    audit_entries = []

    if resolved_vendor_id is None:
        skipped_steps.append(
            {
                "step": "bank_validation",
                "skip_reason": "vendor_unresolved",
            }
        )

        return {
            "risk_flags": risk_flags,
            "advisory_flags": advisory_flags,
            "skipped_steps": skipped_steps,
            "audit_entries": audit_entries,
            "match_result": "skipped",
        }

    invoice_date = getattr(extracted_fields, "invoice_date", None)

    if isinstance(invoice_date, datetime):
        invoice_date = invoice_date.date()


    invoice_iban = canonicalize_iban(getattr(extracted_fields, "bank_iban", None))
    invoice_acc_no = canonicalize_acc_no(getattr(extracted_fields, "bank_acc_no", None))
    invoice_swift = canonicalize_swift(getattr(extracted_fields, "bank_swift", None))
    invoice_bank_acc_name = getattr(extracted_fields, "bank_acc_name", None)
    invoice_currency = canonicalize_text(getattr(extracted_fields, "currency", None))

    # 1. Invoice-side pre-format checks.
    if invoice_iban and iban_checksum_required:
        if not iban_checksum_mod97_valid(invoice_iban):
            add_flag(
                "bank_iban_invalid_checksum",
                field="bank_iban",
                weight=Decimal("0.35"),
                invoice_value=invoice_iban,
            )

    if invoice_iban and invoice_acc_no:
        if not iban_acc_no_consistent(invoice_iban, invoice_acc_no):
            add_flag(
                "bank_iban_accno_inconsistent",
                field="bank_iban/bank_acc_no",
                weight=Decimal("0.35"),
                invoice_value=f"{invoice_iban}|{invoice_acc_no}",
            )

    # 2. Build same-tenant, same-vendor, active candidate pool.
    # all_vendor_rows: non-deleted rows for this vendor, no date filtering.
    # Used for new_bank_details_no_master — distinguishes "no rows at all" from
    # "rows exist but none are currently effective".
    all_vendor_rows = [
        row
        for row in vendor_bank_accounts
        if getattr(row, "tenant_id", None) == document.tenant_id
        and getattr(row, "vendor_id", None) == resolved_vendor_id
        and not getattr(row, "deleted_at", None)
    ]

    candidate_pool = [
        row for row in all_vendor_rows
        if is_effective(row, invoice_date)
    ]

    candidate_pool.sort(
        key=lambda row: (
            bool(getattr(row, "is_primary", False)),
            getattr(row, "effective_from", None) or date.max,
        ),
        reverse=True,
    )

    primary_rows = [
        row for row in candidate_pool
        if bool(getattr(row, "is_primary", False))
    ]

    non_primary_rows = [
        row for row in candidate_pool
        if not bool(getattr(row, "is_primary", False))
    ]

    primary = primary_rows[0] if primary_rows else None

    invoice_has_any_bank_detail = any(
        [
            invoice_iban,
            invoice_acc_no,
            invoice_swift,
            invoice_bank_acc_name,
        ]
    )

    if invoice_has_any_bank_detail and not all_vendor_rows:
        add_flag(
            "new_bank_details_no_master",
            field="bank_details",
            invoice_value="bank_details_present",
        )

        return {
            "risk_flags": risk_flags,
            "advisory_flags": advisory_flags,
            "skipped_steps": skipped_steps,
            "audit_entries": audit_entries,
            "match_result": "no_master_bank_details",
        }

    # 3. Compare against primary account.
    if primary:
        primary_match = True

        master_iban = getattr(primary, "iban_canonical", None)
        master_acc_no = getattr(primary, "acc_no_canonical", None)
        master_swift = canonicalize_swift(getattr(primary, "swift_bic", None))
        master_bank_acc_name = getattr(primary, "bank_acc_name", None)
        master_currency = canonicalize_text(getattr(primary, "currency_code", None))

        if invoice_iban:
            if not master_iban:
                primary_match = False
                add_flag(
                    "new_bank_field_type_no_master",
                    field="bank_iban",
                    invoice_value=invoice_iban,
                    primary_or_non_primary="primary",
                )
            elif invoice_iban != master_iban:
                primary_match = False
                add_flag(
                    "bank_iban_mismatch",
                    field="bank_iban",
                    invoice_value=invoice_iban,
                    master_value=master_iban,
                    primary_or_non_primary="primary",
                )

        if invoice_acc_no and master_acc_no and invoice_acc_no != master_acc_no:
            primary_match = False
            add_flag(
                "bank_acc_no_mismatch",
                field="bank_acc_no",
                invoice_value=invoice_acc_no,
                master_value=master_acc_no,
                primary_or_non_primary="primary",
            )

        if invoice_swift and master_swift and invoice_swift != master_swift:
            primary_match = False
            add_flag(
                "bank_swift_mismatch",
                field="bank_swift",
                invoice_value=invoice_swift,
                master_value=master_swift,
                primary_or_non_primary="primary",
            )

        if invoice_bank_acc_name and master_bank_acc_name:
            if not bank_account_name_matches(invoice_bank_acc_name, master_bank_acc_name):
                primary_match = False
                add_flag(
                    "bank_acc_name_mismatch",
                    field="bank_acc_name",
                    weight=Decimal("0.30"),
                    invoice_value=invoice_bank_acc_name,
                    master_value=master_bank_acc_name,
                    primary_or_non_primary="primary",
                )

        if invoice_currency and master_currency and invoice_currency != master_currency:
            primary_match = False
            add_flag(
                "bank_currency_mismatch",
                field="currency",
                weight=Decimal("0.20"),
                currency_invoice=invoice_currency,
                currency_master=master_currency,
                primary_or_non_primary="primary",
            )

        match_result = "primary_match" if primary_match else "primary_mismatch"

    else:
        match_result = "no_primary_bank_account"

        secondary_match = None
        for row in non_primary_rows:
            if invoice_iban and getattr(row, "iban_canonical", None) == invoice_iban:
                secondary_match = row
                break

        if secondary_match:
            add_flag(
                "bank_non_primary_match",
                field="bank_iban",
                weight=Decimal("0.25"),
                match_result="matched_non_primary",
                invoice_value=invoice_iban,
                master_value=getattr(secondary_match, "iban_canonical", None),
                primary_or_non_primary="non_primary",
            )

    return {
        "risk_flags": risk_flags,
        "advisory_flags": advisory_flags,
        "skipped_steps": skipped_steps,
        "audit_entries": audit_entries,
        "match_result": match_result,
        "candidate_count": len(candidate_pool),
        "primary_candidate_count": len(primary_rows),
        "non_primary_candidate_count": len(non_primary_rows),
    }