from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal, ROUND_HALF_EVEN, getcontext from typing import Any, Callable, Optional import re import unicodedata from rapidfuzz import fuzz from two_way_match import MatchExceptionRecord, InvoiceLine, to_decimal # ============================================================ # Decimal hygiene # ============================================================ getcontext().prec = 28 getcontext().rounding = ROUND_HALF_EVEN D0 = Decimal("0") D100 = Decimal("100") # ============================================================ # Data models # ============================================================ @dataclass(frozen=True) class RateCardItem: id: int description: Optional[str] = None description_canonical: Optional[str] = None unit_price: Optional[Decimal] = None item_type: Optional[str] = None # goods | services | None @dataclass(frozen=True) class ContractRecord: id: int tenant_id: int vendor_id: int entity_id: int currency: str status: str = "active" deleted_at: Any = None effective_from: Optional[date] = None effective_to: Optional[date] = None created_at: Optional[datetime] = None total_value: Optional[Decimal] = None cumulative_billed_amount: Optional[Decimal] = None rate_card_items: Optional[list[RateCardItem]] = None @dataclass(frozen=True) class ContractInvoiceHeader: id: int tenant_id: int entity_id: int resolved_vendor_id: Optional[int] currency: Optional[str] invoice_date: Optional[date] service_start_date: Optional[date] = None service_end_date: Optional[date] = None total_amount: Optional[Decimal] = None advisory_flags: tuple[str, ...] = () @dataclass class ContractLineResult: invoice_line_id: int rate_card_item_id: Optional[int] line_status: str = "matched" # matched | no_match | exception line_zone: str = "green" # green | amber | red price_var_pct: Optional[Decimal] = None allocated_inv_amount: Optional[Decimal] = None allocated_rate_amount: Optional[Decimal] = None exception_type: Optional[str] = None exception_detail: dict[str, Any] = field(default_factory=dict) @dataclass class ContractMatchResult: match_type: str = "contract" overall_status: str = "exception" # matched | partial_match | exception match_zone: str = "red" # green | amber | red | critical documents_status: str = "pending_approval" lines_total: int = 0 lines_matched: int = 0 lines_in_exception: int = 0 advisory_flags: list[str] = field(default_factory=list) risk_flags: list[str] = field(default_factory=list) skipped_steps: list[dict] = field(default_factory=list) header_exceptions: list[MatchExceptionRecord] = field(default_factory=list) match_exceptions: list[MatchExceptionRecord] = field(default_factory=list) match_line_results: list[ContractLineResult] = field(default_factory=list) selected_contract_id: Optional[int] = None @dataclass(frozen=True) class ContractConfig: # RapidFuzz thresholds for rate-card / tiebreaker matching high_threshold: float = 90.0 medium_threshold: float = 80.0 score_cutoff: float = 80.0 # Price variance bands (same idea as 2-way; tune to your config table) price_goods_green: Decimal = Decimal("5") price_goods_amber: Decimal = Decimal("10") price_services_green: Decimal = Decimal("10") price_services_amber: Decimal = Decimal("15") precision: int = 28 # ============================================================ # Helpers # ============================================================ def canonicalise_identifier(value: Optional[str]) -> Optional[str]: if value is None: return None value = unicodedata.normalize("NFKC", str(value)).upper() value = re.sub(r"[^A-Z0-9]+", "", value) return value or None def norm_currency(value: Optional[str]) -> Optional[str]: return value.upper().strip() if value else None def price_variance_pct(inv_price: Optional[Decimal], ref_price: Optional[Decimal]) -> Optional[Decimal]: if inv_price is None or ref_price is None or ref_price == 0: return None return (abs(inv_price - ref_price) / ref_price * D100).quantize(Decimal("0.0001")) def _bands(item_type: Optional[str], cfg: ContractConfig) -> tuple[Decimal, Decimal]: item_type = (item_type or "services").lower() if item_type == "goods": return cfg.price_goods_green, cfg.price_goods_amber return cfg.price_services_green, cfg.price_services_amber def _line_zone_from_price(price_pct: Optional[Decimal], item_type: Optional[str], cfg: ContractConfig) -> str: if price_pct is None: return "green" green, amber = _bands(item_type, cfg) if price_pct <= green: return "green" if price_pct <= amber: return "amber" return "red" def _is_service_window_contained( contract: ContractRecord, invoice_start: Optional[date], invoice_end: Optional[date], ) -> bool: if invoice_start is None or invoice_end is None: return False if contract.effective_from is None: return False effective_to = contract.effective_to or date(9999, 12, 31) return contract.effective_from <= invoice_start and invoice_end <= effective_to def _joined_invoice_descriptions(inv_lines: list[InvoiceLine]) -> str: parts = [] for line in inv_lines: if line.description_canonical: parts.append(line.description_canonical) elif line.description: parts.append(canonicalise_identifier(line.description) or "") return " ".join(p for p in parts if p).strip() def rapidfuzz_best_match( query: str, candidates: list[dict[str, Any]], *, high_threshold: float, medium_threshold: float, score_cutoff: float, ) -> dict[str, Any]: """ Simple local matcher. Swap this out with your platform name-match algorithm later if required. """ if not query or not candidates: return {"decision": "unmatched", "score_tier": None, "score": 0.0, "record_id": None} best_record_id = None best_score = 0.0 for cand in candidates: record_id = cand.get("record_id") names = cand.get("names") or [] for name in names: if not name: continue score = float(fuzz.WRatio(query, name, score_cutoff=score_cutoff)) if score > best_score: best_score = score best_record_id = record_id if best_record_id is None or best_score < score_cutoff: return {"decision": "unmatched", "score_tier": None, "score": 0.0, "record_id": None} if best_score >= high_threshold: tier = "high" elif best_score >= medium_threshold: tier = "medium" else: tier = "low" return { "decision": "matched" if tier in {"high", "medium"} else "suggestion", "score_tier": tier, "score": round(best_score, 2), "record_id": best_record_id, } def _contract_stub_result( invoice: ContractInvoiceHeader, inv_lines: list[InvoiceLine], exception_type: str, detail: dict[str, Any], ) -> ContractMatchResult: res = ContractMatchResult( overall_status="exception", match_zone="red", documents_status="pending_approval", lines_total=len(inv_lines), lines_matched=0, lines_in_exception=len(inv_lines), ) res.header_exceptions.append( MatchExceptionRecord( exception_type=exception_type, exception_detail=detail, ) ) return res # ============================================================ # Contract candidate selection # ============================================================ def select_contract_candidate( invoice: ContractInvoiceHeader, inv_lines: list[InvoiceLine], contracts: list[ContractRecord], cfg: ContractConfig, ) -> tuple[Optional[ContractRecord], Optional[MatchExceptionRecord]]: """ Returns: - selected contract - or a MatchExceptionRecord for stub paths """ inv_currency = norm_currency(invoice.currency) initial_candidates: list[dict[str, Any]] = [] for c in contracts: if c.deleted_at is not None: continue if c.status != "active": continue if c.tenant_id != invoice.tenant_id: continue if c.vendor_id != invoice.resolved_vendor_id: continue if c.entity_id != invoice.entity_id: continue if norm_currency(c.currency) != norm_currency(invoice.currency): continue if invoice.invoice_date is None: continue if c.effective_from is None: continue effective_to = c.effective_to or date(9999, 12, 31) if not (c.effective_from <= invoice.invoice_date < effective_to): continue initial_candidates.append( { "contract": c, "elimination_reason": None, } ) if not initial_candidates: return None, MatchExceptionRecord( exception_type="contract_not_found", exception_detail={ "reason": "initial_query_returned_zero_rows", "tenant_id": invoice.tenant_id, "vendor_id": invoice.resolved_vendor_id, "entity_id": invoice.entity_id, "currency": inv_currency, "invoice_date": str(invoice.invoice_date) if invoice.invoice_date else None, }, ) remaining = [x["contract"] for x in initial_candidates] # Tiebreaker 1: service-period containment (soft preference — only when both dates present) if invoice.service_start_date is not None and invoice.service_end_date is not None: contained = [c for c in remaining if _is_service_window_contained(c, invoice.service_start_date, invoice.service_end_date)] if contained: remaining = contained # If none pass containment, keep all and let run_contract_match_core raise billing_period_outside_contract # Tiebreaker 2: rate-card best fit via RapidFuzz # Query uses joined canonical invoice descriptions. joined_desc = _joined_invoice_descriptions(inv_lines) if joined_desc: scored_candidates: list[dict[str, Any]] = [] for c in remaining: if not c.rate_card_items: continue item_names = [ (item.description_canonical or canonicalise_identifier(item.description) or "") for item in c.rate_card_items ] item_names = [n for n in item_names if n] if not item_names: continue scored_candidates.append({"record_id": c.id, "names": item_names}) if scored_candidates: # Contracts without rate card items cannot participate; exclude them now # so they don't survive to tiebreaker 3 ahead of contracts that do. scored_ids = {x["record_id"] for x in scored_candidates} remaining = [c for c in remaining if c.id in scored_ids] # Evaluate each contract individually: keep only those whose rate card # descriptions yield decision='matched' against the joined invoice text. matched_ids = { x["record_id"] for x in scored_candidates if rapidfuzz_best_match( joined_desc, [x], high_threshold=cfg.high_threshold, medium_threshold=cfg.medium_threshold, score_cutoff=cfg.score_cutoff, )["decision"] == "matched" } if matched_ids: remaining = [c for c in remaining if c.id in matched_ids] else: return None, MatchExceptionRecord( exception_type="cannot_match_no_contract", exception_detail={ "reason": "rate_card_best_fit_filtered_all_candidates", "joined_invoice_descriptions": joined_desc, "considered_contract_ids": [c.id for c in remaining], }, ) # Tiebreaker 3: newest created_at DESC, id DESC remaining.sort( key=lambda c: ( c.created_at or datetime.min, c.id, ), reverse=True, ) if len(remaining) == 1: return remaining[0], None if len(remaining) == 0: return None, MatchExceptionRecord( exception_type="cannot_match_no_contract", exception_detail={ "reason": "all_candidates_removed_by_tiebreakers", }, ) return None, MatchExceptionRecord( exception_type="ambiguous_active_contract", exception_detail={ "reason": "multiple_contracts_remain_after_tiebreakers", "surviving_contract_ids": [c.id for c in remaining], }, ) # ============================================================ # Config builder # ============================================================ def _cfg_from_contract_dict(d: dict) -> ContractConfig: def _d(key: str, default: str) -> Decimal: return Decimal(str(d.get(key, default))) def _f(key: str, default: float) -> float: return float(d.get(key, default)) def _i(key: str, default: int) -> int: return int(d.get(key, default)) return ContractConfig( high_threshold=_f("contract_fuzzy_high_threshold", 90.0), medium_threshold=_f("contract_fuzzy_medium_threshold", 80.0), score_cutoff=_f("contract_fuzzy_score_cutoff", 80.0), price_goods_green=_d("contract_price_goods_green", "5"), price_goods_amber=_d("contract_price_goods_amber", "10"), price_services_green=_d("contract_price_services_green", "10"), price_services_amber=_d("contract_price_services_amber", "15"), precision=_i("contract_precision", 28), ) # ============================================================ # Main contract match core # ============================================================ def run_contract_match_core( invoice: ContractInvoiceHeader, inv_lines: list[InvoiceLine], contracts: list[ContractRecord], *, matching_configs: Optional[dict] = None, ) -> ContractMatchResult: """ Contract match used when po_number_canonical is NULL. Returns: - a stub exception result if no unique contract can be selected - otherwise runs billing-period, rate-card, and budget checks Note: - cumulative_billed_amount is NOT incremented here - this is matching-only, not approval """ cfg = _cfg_from_contract_dict(matching_configs or {}) getcontext().prec = cfg.precision result = ContractMatchResult( lines_total=len(inv_lines), documents_status="pending_approval", match_type="contract", ) result.advisory_flags = list(invoice.advisory_flags) if invoice.resolved_vendor_id is None: return _contract_stub_result( invoice, inv_lines, "cannot_match_no_vendor", { "reason": "resolved_vendor_id_is_required", "tenant_id": invoice.tenant_id, "entity_id": invoice.entity_id, }, ) selected_contract, stub_exc = select_contract_candidate(invoice, inv_lines, contracts, cfg) if stub_exc is not None or selected_contract is None: # contract_not_found / cannot_match_no_contract / ambiguous_active_contract result.header_exceptions.append(stub_exc or MatchExceptionRecord( exception_type="cannot_match_no_contract", exception_detail={"reason": "no_selected_contract"}, )) result.overall_status = "exception" result.match_zone = "red" result.lines_in_exception = len(inv_lines) return result result.selected_contract_id = selected_contract.id # --------------------------------------------------------- # Contract checks # --------------------------------------------------------- # 1) Billing period missing if invoice.service_start_date is None or invoice.service_end_date is None: result.match_exceptions.append( MatchExceptionRecord( exception_type="billing_period_unverified", exception_detail={ "reason": "service_start_date_or_service_end_date_missing", "service_start_date": str(invoice.service_start_date) if invoice.service_start_date else None, "service_end_date": str(invoice.service_end_date) if invoice.service_end_date else None, }, ) ) # 2) Billing period outside contract if invoice.service_start_date is not None and invoice.service_end_date is not None: if not _is_service_window_contained(selected_contract, invoice.service_start_date, invoice.service_end_date): result.match_exceptions.append( MatchExceptionRecord( exception_type="billing_period_outside_contract", exception_detail={ "reason": "invoice_service_window_not_contained_in_contract", "service_start_date": str(invoice.service_start_date), "service_end_date": str(invoice.service_end_date), "contract_effective_from": str(selected_contract.effective_from) if selected_contract.effective_from else None, "contract_effective_to": str(selected_contract.effective_to) if selected_contract.effective_to else None, }, ) ) # 3) Rate card exists if not selected_contract.rate_card_items: result.match_exceptions.append( MatchExceptionRecord( exception_type="no_rate_card_configured", exception_detail={ "reason": "contract_has_no_rate_card_items", "contract_id": selected_contract.id, }, ) ) result.overall_status = "exception" result.match_zone = "red" result.lines_in_exception = len(inv_lines) return result # 4) Per-line rate check rate_card_candidates = [ { "record_id": item.id, "names": [item.description_canonical or canonicalise_identifier(item.description) or ""], } for item in selected_contract.rate_card_items if (item.description_canonical or item.description) ] line_matched_count = 0 line_exception_count = 0 for inv_line in inv_lines: query = inv_line.description_canonical or canonicalise_identifier(inv_line.description) or "" if not query: result.match_exceptions.append( MatchExceptionRecord( exception_type="no_rate_card_match", line_id=inv_line.id, exception_detail={ "reason": "invoice_line_description_missing", "invoice_line_id": inv_line.id, }, ) ) line_exception_count += 1 continue rf = rapidfuzz_best_match( query=query, candidates=rate_card_candidates, high_threshold=cfg.high_threshold, medium_threshold=cfg.medium_threshold, score_cutoff=cfg.score_cutoff, ) if rf["decision"] != "matched" or rf["record_id"] is None: result.match_exceptions.append( MatchExceptionRecord( exception_type="no_rate_card_match", line_id=inv_line.id, exception_detail={ "reason": "rate_card_name_match_failed", "invoice_line_id": inv_line.id, "query": query, }, ) ) line_exception_count += 1 continue rate_item = next((x for x in selected_contract.rate_card_items if x.id == rf["record_id"]), None) if rate_item is None: result.match_exceptions.append( MatchExceptionRecord( exception_type="no_rate_card_match", line_id=inv_line.id, exception_detail={ "reason": "matched_rate_card_item_not_found", "invoice_line_id": inv_line.id, "matched_rate_card_item_id": rf["record_id"], }, ) ) line_exception_count += 1 continue # Compare unit price if present if inv_line.unit_price is None: if "rate_card_match_no_price_compare" not in result.advisory_flags: result.advisory_flags.append("rate_card_match_no_price_compare") result.match_line_results.append( ContractLineResult( invoice_line_id=inv_line.id, rate_card_item_id=rate_item.id, line_status="matched", line_zone="green", price_var_pct=None, allocated_inv_amount=inv_line.amount, allocated_rate_amount=rate_item.unit_price, ) ) line_matched_count += 1 continue if rate_item.unit_price is None: # Name match succeeded; price comparison is impossible. Treat as an # advisory (not a hard exception) because the match itself is valid. if "rate_card_match_no_price_compare" not in result.advisory_flags: result.advisory_flags.append("rate_card_match_no_price_compare") result.match_line_results.append( ContractLineResult( invoice_line_id=inv_line.id, rate_card_item_id=rate_item.id, line_status="matched", line_zone="green", price_var_pct=None, allocated_inv_amount=inv_line.amount, allocated_rate_amount=None, ) ) line_matched_count += 1 continue p_var = price_variance_pct(inv_line.unit_price, rate_item.unit_price) zone = _line_zone_from_price(p_var, rate_item.item_type, cfg) if zone == "red": result.match_exceptions.append( MatchExceptionRecord( exception_type="rate_card_price_mismatch", line_id=inv_line.id, po_line_id=rate_item.id, exception_detail={ "reason": "price_outside_tolerance", "invoice_line_id": inv_line.id, "rate_card_item_id": rate_item.id, "invoice_unit_price": str(inv_line.unit_price), "rate_card_unit_price": str(rate_item.unit_price), "price_var_pct": str(p_var), }, ) ) result.match_line_results.append( ContractLineResult( invoice_line_id=inv_line.id, rate_card_item_id=rate_item.id, line_status="exception", line_zone="red", price_var_pct=p_var, allocated_inv_amount=inv_line.amount, allocated_rate_amount=rate_item.unit_price, exception_type="rate_card_price_mismatch", ) ) line_exception_count += 1 continue result.match_line_results.append( ContractLineResult( invoice_line_id=inv_line.id, rate_card_item_id=rate_item.id, line_status="matched", line_zone=zone, price_var_pct=p_var, allocated_inv_amount=inv_line.amount, allocated_rate_amount=rate_item.unit_price, ) ) line_matched_count += 1 result.lines_matched = line_matched_count result.lines_in_exception = line_exception_count + (len(inv_lines) - line_matched_count - line_exception_count) # --------------------------------------------------------- # Budget ceiling # --------------------------------------------------------- inv_currency = norm_currency(invoice.currency) contract_currency = norm_currency(selected_contract.currency) if contract_currency != inv_currency: result.match_exceptions.append( MatchExceptionRecord( exception_type="contract_currency_mismatch", exception_detail={ "reason": "contract_currency_changed_or_mismatch", "invoice_currency": inv_currency, "contract_currency": contract_currency, "contract_id": selected_contract.id, }, ) ) result.overall_status = "exception" result.match_zone = "red" return result if invoice.total_amount is not None: cumulative = selected_contract.cumulative_billed_amount or D0 total_value = selected_contract.total_value if total_value is not None and (cumulative + invoice.total_amount) > total_value: result.match_exceptions.append( MatchExceptionRecord( exception_type="contract_budget_exceeded", exception_detail={ "reason": "budget_ceiling_breached", "contract_id": selected_contract.id, "invoice_total_amount": str(invoice.total_amount), "cumulative_billed_amount": str(cumulative), "contract_total_value": str(total_value), }, ) ) result.overall_status = "exception" result.match_zone = "red" return result # --------------------------------------------------------- # Overall status / zone # --------------------------------------------------------- if any(exc.exception_type in { "billing_period_unverified", "billing_period_outside_contract", "no_rate_card_configured", "no_rate_card_match", "rate_card_price_mismatch", "contract_budget_exceeded", "contract_currency_mismatch", } for exc in result.match_exceptions): # If any hard contract exception exists, keep it red. if result.lines_matched > 0 and result.lines_in_exception > 0: result.overall_status = "partial_match" else: result.overall_status = "exception" result.match_zone = "red" else: if result.lines_matched == result.lines_total and result.lines_in_exception == 0: result.overall_status = "matched" result.match_zone = "amber" if any(r.line_zone == "amber" for r in result.match_line_results) else "green" elif result.lines_matched > 0: result.overall_status = "partial_match" result.match_zone = "amber" if any(r.line_zone == "amber" for r in result.match_line_results) else "green" else: result.overall_status = "exception" result.match_zone = "red" return result