src/Invoice Validation Functions/Vendor Validation.py

from decimal import Decimal
from hashlib import sha256
from unicodedata import normalize
from uuid import uuid4
from datetime import datetime, timezone

from rapidfuzz import fuzz, process


def run_vendor_validation(
    *,
    document,
    extracted_fields,
    vendors,
    platform_configs,
    route_contexts=None,
    actor_id=None,
):
    """
    Runs Cat 2 vendor validation using RapidFuzz instead of the platform name-match algorithm.

    Expected document fields:
        document.id
        document.tenant_id

    Expected extracted_fields fields:
        sender_name
        sender_name_canonical
        sender_tax_id_canonical
        bank_iban_canonical

    Expected vendor fields:
        id
        tenant_id
        legal_name
        vendor_name
        legal_name_canonical
        billing_country
        billing_address
        vendor_status
        deleted_at

    Returns a dict containing:
        resolved_vendor_id
        resolved_sender_name
        resolved_sender_addr
        risk_flags
        advisory_flags
        skipped_steps
        audit_row
        match_result
    """

    route_contexts = set(route_contexts or [])

    require_high_tier_contexts = set(
        platform_configs.get(
            "validation.vendor_match_require_high_tier_contexts",
            ["contract_match", "bank_validation", "approval_revalidation"],
        )
    )

    algorithm_version = platform_configs.get(
        "validation.name_match_algorithm_version",
        "rapidfuzz.local.v1",
    )

    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 canonicalize(value):
        if not value:
            return ""
        return normalize("NFKC", str(value)).lower().strip()

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

    def get_vendor_names(vendor):
        names = []

        for name in (
            getattr(vendor, "legal_name", None),
            getattr(vendor, "vendor_name", None),
        ):
            canonical = canonicalize(name)
            if canonical and canonical not in names:
                names.append(canonical)

        return names

    def build_candidate_pool():
        pool = []

        for vendor in vendors:
            if getattr(vendor, "tenant_id", None) != document.tenant_id:
                continue

            if getattr(vendor, "vendor_status", None) == "deleted":
                continue

            if getattr(vendor, "deleted_at", None) is not None:
                continue

            names = get_vendor_names(vendor)
            if not names:
                continue

            pool.append(
                {
                    "record_id": vendor.id,
                    "vendor": vendor,
                    "names": names,
                    "auxiliary": {
                        "country": getattr(vendor, "billing_country", None),
                        "tax_id_hash": hash_prefix(
                            getattr(extracted_fields, "sender_tax_id_canonical", None)
                        ),
                        "iban_hash": hash_prefix(
                            getattr(extracted_fields, "bank_iban_canonical", None)
                        ),
                    },
                }
            )

        return pool

    def rapidfuzz_name_match(query_name, candidate_pool):
        choices = []

        for candidate in candidate_pool:
            for name in candidate["names"]:
                choices.append(
                    {
                        "name": name,
                        "record_id": candidate["record_id"],
                        "vendor": candidate["vendor"],
                    }
                )

        if not query_name or not choices:
            return {
                "decision": "unmatched",
                "matched_record_id": None,
                "matched_vendor": None,
                "runner_up_record_id": None,
                "runner_up_vendor": None,
                "score": Decimal("0"),
                "score_tier": "low",
                "algorithm_version": algorithm_version,
                "trace_id": str(uuid4()),
            }

        raw_matches = process.extract(
            query_name,
            [choice["name"] for choice in choices],
            scorer=fuzz.WRatio,
            limit=5,
        )

        ranked = []
        seen_record_ids = set()

        for matched_name, score, choice_index in raw_matches:
            choice = choices[choice_index]
            record_id = choice["record_id"]

            if record_id in seen_record_ids:
                continue

            seen_record_ids.add(record_id)

            ranked.append(
                {
                    "record_id": record_id,
                    "vendor": choice["vendor"],
                    "matched_name": matched_name,
                    "score": Decimal(str(score / 100)).quantize(Decimal("0.0001")),
                }
            )

        if not ranked:
            best = None
            runner_up = None
            score = Decimal("0")
        else:
            best = ranked[0]
            runner_up = ranked[1] if len(ranked) > 1 else None
            score = best["score"]

        if score >= high_threshold:
            decision = "matched"
            score_tier = "high"
            matched_record_id = best["record_id"]
            matched_vendor = best["vendor"]
        elif score >= medium_threshold:
            decision = "matched"
            score_tier = "medium"
            matched_record_id = best["record_id"]
            matched_vendor = best["vendor"]
        elif score >= suggestion_threshold:
            decision = "suggestion"
            score_tier = "low"
            matched_record_id = None
            matched_vendor = None
        else:
            decision = "unmatched"
            score_tier = "low"
            matched_record_id = None
            matched_vendor = None

        return {
            "decision": decision,
            "matched_record_id": matched_record_id,
            "matched_vendor": matched_vendor,
            "runner_up_record_id": (
                best["record_id"] if decision == "suggestion" and best
                else (runner_up["record_id"] if runner_up else None)
            ),
            "runner_up_vendor": (
                best["vendor"] if decision == "suggestion" and best
                else (runner_up["vendor"] if runner_up else None)
            ),
            "score": score,
            "score_tier": score_tier,
            "algorithm_version": algorithm_version,
            "trace_id": str(uuid4()),
        }

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

    resolved_vendor_id = None
    resolved_sender_name = None
    resolved_sender_addr = None

    query_name = canonicalize(
        getattr(extracted_fields, "sender_name_canonical", None)
        or getattr(extracted_fields, "sender_name", None)
    )

    candidate_pool = build_candidate_pool()

    match_result = rapidfuzz_name_match(
        query_name=query_name,
        candidate_pool=candidate_pool,
    )

    decision = match_result["decision"]
    score_tier = match_result["score_tier"]

    control_context_requires_high = bool(route_contexts & require_high_tier_contexts)

    if decision == "matched" and score_tier == "high":
        resolved_vendor_id = match_result["matched_record_id"]

    elif decision == "matched" and score_tier == "medium":
        if control_context_requires_high:
            match_result["decision"] = "suggestion"
            advisory_flags.append(
                {
                    "code": "vendor_suggestion",
                    "severity": "advisory",
                    "record_id": match_result["matched_record_id"],
                    "reason": "medium_tier_match_downgraded_for_control_context",
                }
            )
            skipped_steps.extend(
                [
                    {"step": "vendor_tax_validation", "skip_reason": "vendor_unresolved"},
                    {"step": "vendor_bank_validation", "skip_reason": "vendor_unresolved"},
                    {"step": "vendor_status_validation", "skip_reason": "vendor_unresolved"},
                ]
            )
        else:
            resolved_vendor_id = match_result["matched_record_id"]
            risk_flags.append(
                {
                    "code": "low_confidence_vendor_resolution",
                    "severity": "medium",
                    "record_id": resolved_vendor_id,
                    "score_tier": score_tier,
                }
            )

    elif decision == "suggestion":
        advisory_flags.append(
            {
                "code": "vendor_suggestion",
                "severity": "advisory",
                "record_id": match_result["runner_up_record_id"],
                "reason": "possible_vendor_match",
            }
        )
        skipped_steps.extend(
            [
                {"step": "sender_field_override", "skip_reason": "vendor_unresolved"},
                {"step": "vendor_tax_validation", "skip_reason": "vendor_unresolved"},
                {"step": "vendor_bank_validation", "skip_reason": "vendor_unresolved"},
                {"step": "vendor_status_validation", "skip_reason": "vendor_unresolved"},
            ]
        )

    else:
        risk_flags.append(
            {
                "code": "unknown_vendor",
                "severity": "high",
                "reason": "no_vendor_match_found",
            }
        )
        skipped_steps.extend(
            [
                {"step": "sender_field_override", "skip_reason": "vendor_unresolved"},
                {"step": "vendor_tax_validation", "skip_reason": "vendor_unresolved"},
                {"step": "vendor_bank_validation", "skip_reason": "vendor_unresolved"},
                {"step": "vendor_status_validation", "skip_reason": "vendor_unresolved"},
            ]
        )

    blocked_vendor = None

    if match_result["decision"] in ("matched", "suggestion"):
        if match_result.get("matched_vendor") and match_result["matched_vendor"].vendor_status == "blocked":
            blocked_vendor = match_result["matched_vendor"]

        if match_result.get("runner_up_vendor") and match_result["runner_up_vendor"].vendor_status == "blocked":
            blocked_vendor = match_result["runner_up_vendor"]

    if blocked_vendor:
        risk_flags.append(
            {
                "code": "vendor_blocked",
                "severity": "high",
                "record_id": blocked_vendor.id,
                "reason": "matched_or_suggested_vendor_is_blocked",
            }
        )

    if resolved_vendor_id:
        resolved_vendor = match_result["matched_vendor"]

        resolved_sender_name = (
            getattr(resolved_vendor, "legal_name", None)
            or getattr(resolved_vendor, "vendor_name", None)
        )

        billing_address = getattr(resolved_vendor, "billing_address", None)
        if billing_address is not None:
            resolved_sender_addr = billing_address

    audit_row = {
        "document_id": document.id,
        "query_name_hash_prefix_8": hash_prefix(query_name, length=8),
        "algorithm_version": algorithm_version,
        "decision": match_result["decision"],
        "matched_record_id": match_result["matched_record_id"],
        "runner_up_record_id": match_result["runner_up_record_id"],
        "score_tier": match_result["score_tier"],
        "score": str(match_result["score"]),
        "trace_id": match_result["trace_id"],
        "called_at": datetime.now(timezone.utc),
        "actor_id": actor_id,
    }

    return {
        "resolved_vendor_id": resolved_vendor_id,
        "resolved_sender_name": resolved_sender_name,
        "resolved_sender_addr": resolved_sender_addr,
        "risk_flags": risk_flags,
        "advisory_flags": advisory_flags,
        "skipped_steps": skipped_steps,
        "audit_row": audit_row,
        "match_result": {
            "decision": match_result["decision"],
            "matched_record_id": resolved_vendor_id,
            "runner_up_record_id": match_result["runner_up_record_id"],
            "score": match_result["score"],
            "score_tier": match_result["score_tier"],
            "algorithm_version": match_result["algorithm_version"],
            "trace_id": match_result["trace_id"],
        },
    }