from __future__ import annotations from dataclasses import dataclass, field from datetime import date from decimal import Decimal, ROUND_HALF_EVEN, getcontext from typing import Any, Callable, Optional import re import unicodedata from rapidfuzz import fuzz # ============================================================ # Decimal / arithmetic hygiene # ============================================================ getcontext().prec = 28 getcontext().rounding = ROUND_HALF_EVEN D0 = Decimal("0") D100 = Decimal("100") # ============================================================ # Data models # ============================================================ @dataclass(frozen=True) class InvoiceLine: id: int description: Optional[str] = None description_canonical: Optional[str] = None sku: Optional[str] = None sku_canonical: Optional[str] = None gl_account: Optional[str] = None quantity: Optional[Decimal] = None unit_price: Optional[Decimal] = None amount: Optional[Decimal] = None @dataclass(frozen=True) class POLine: id: int description: Optional[str] = None description_canonical: Optional[str] = None sku: Optional[str] = None sku_canonical: Optional[str] = None gl_account: Optional[str] = None quantity: Optional[Decimal] = None invoiced_qty: Optional[Decimal] = None unit_price: Optional[Decimal] = None item_type: Optional[str] = None # 'goods' | 'services' | None status: str = "open" # 'open' | 'fully_invoiced' | 'closed' | 'cancelled' deleted_at: Any = None @dataclass(frozen=True) class POHeader: id: int vendor_id: Optional[int] = None currency: Optional[str] = None status: str = "open" deleted_at: Any = None @dataclass(frozen=True) class InvoiceHeader: id: int status: str vendor_id: Optional[int] = None currency: Optional[str] = None total_amount: Optional[Decimal] = None advisory_flags: tuple[str, ...] = () doc_type: str = "invoice" tenant_id: Optional[int] = None invoice_date: Optional[date] = None @dataclass class MatchExceptionRecord: exception_type: str line_id: Optional[int] = None po_line_id: Optional[int] = None exception_detail: dict[str, Any] = field(default_factory=dict) @dataclass class MatchLineResult: invoice_line_id: int po_line_id: Optional[int] match_tier: int allocated_inv_amount: Optional[Decimal] = None allocated_po_amount: Optional[Decimal] = None price_var_pct: Optional[Decimal] = None qty_var_pct: Optional[Decimal] = None line_zone: str = "green" # green | amber | red line_status: str = "matched" # matched | no_match | exception match_type: str = "two_way" @dataclass class TwoWayMatchResult: match_type: str = "two_way" overall_status: str = "exception" # matched | partial_match | exception match_zone: str = "green" # green | amber | red documents_status: str = "pending_approval" lines_total: int = 0 lines_matched: int = 0 lines_in_exception: int = 0 risk_flags: list[str] = field(default_factory=list) advisory_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[MatchLineResult] = field(default_factory=list) @dataclass(frozen=True) class TwoWayConfig: tier3_price_variance_threshold: Decimal = Decimal("0.10") price_goods_green: Decimal = Decimal("5") price_goods_amber: Decimal = Decimal("10") price_services_green: Decimal = Decimal("10") price_services_amber: Decimal = Decimal("15") qty_goods_green: Decimal = Decimal("2") qty_goods_amber: Decimal = Decimal("4") qty_services_green: Decimal = Decimal("3") qty_services_amber: Decimal = Decimal("6") tier4_gl_group_max_lines: int = 100 fuzzy_high_threshold: float = 90.0 fuzzy_medium_threshold: float = 80.0 fuzzy_score_cutoff: float = 80.0 precision: int = 28 def _cfg_from_dict(d: dict) -> TwoWayConfig: """Build TwoWayConfig from matching_configs dict. Falls back to dataclass defaults for missing keys.""" def _d(key: str, default: str) -> Decimal: return Decimal(str(d.get(key, default))) def _i(key: str, default: int) -> int: return int(d.get(key, default)) def _f(key: str, default: float) -> float: return float(d.get(key, default)) return TwoWayConfig( tier3_price_variance_threshold=_d("tier3_price_variance_threshold", "0.10"), price_goods_green=_d("price_goods_green", "5"), price_goods_amber=_d("price_goods_amber", "10"), price_services_green=_d("price_services_green", "10"), price_services_amber=_d("price_services_amber", "15"), qty_goods_green=_d("qty_goods_green", "2"), qty_goods_amber=_d("qty_goods_amber", "4"), qty_services_green=_d("qty_services_green", "3"), qty_services_amber=_d("qty_services_amber", "6"), tier4_gl_group_max_lines=_i("tier4_gl_group_max_lines", 100), fuzzy_high_threshold=_f("fuzzy_high_threshold", 90.0), fuzzy_medium_threshold=_f("fuzzy_medium_threshold", 80.0), fuzzy_score_cutoff=_f("fuzzy_score_cutoff", 80.0), precision=_i("precision", 28), ) # ============================================================ # Helper functions # ============================================================ def to_decimal(value: Any) -> Optional[Decimal]: if value is None or value == "": return None if isinstance(value, Decimal): return value return Decimal(str(value)) def canonicalise_identifier(value: Optional[str]) -> Optional[str]: """ NFKC + uppercase + strip non-alphanumeric. """ 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], po_price: Optional[Decimal]) -> Optional[Decimal]: """ abs(inv - po) / po * 100 Skip if invoice price is NULL or po price is NULL or po price == 0. """ if inv_price is None or po_price is None or po_price == 0: return None return (abs(inv_price - po_price) / po_price * D100).quantize(Decimal("0.0001")) def qty_over_pct(inv_qty: Optional[Decimal], available_qty: Optional[Decimal]) -> Optional[Decimal]: """ max(0, (inv_qty - available_qty) / available_qty * 100) Skip if inv_qty is NULL or available_qty is NULL/zero. """ if inv_qty is None or available_qty is None or available_qty == 0: return None return (max(D0, (inv_qty - available_qty) / available_qty * D100)).quantize(Decimal("0.0001")) def available_qty(po_line: POLine) -> Decimal: q = to_decimal(po_line.quantity) or D0 iq = to_decimal(po_line.invoiced_qty) or D0 return q - iq def remaining_po_value(po_lines: list[POLine]) -> Decimal: total = D0 for line in po_lines: if line.deleted_at is not None: continue if line.status != "open": continue q = to_decimal(line.quantity) or D0 iq = to_decimal(line.invoiced_qty) or D0 up = to_decimal(line.unit_price) or D0 total += (q - iq) * up return total def same_vendor_merge_chain( invoice_vendor_id: Optional[int], po_vendor_id: Optional[int], merge_chain_fn: Optional[Callable[[Optional[int], Optional[int]], bool]] = None, ) -> bool: if invoice_vendor_id is None or po_vendor_id is None: return False if merge_chain_fn is not None: return bool(merge_chain_fn(invoice_vendor_id, po_vendor_id)) return invoice_vendor_id == po_vendor_id def zone_from_pct( price_pct: Optional[Decimal], qty_pct: Optional[Decimal], item_type: Optional[str], cfg: TwoWayConfig, ) -> str: """ Worst-wins among price and qty, ignoring NULL dimensions. green -> amber -> red """ item_type = (item_type or "services").lower() if item_type == "goods": price_green, price_amber = cfg.price_goods_green, cfg.price_goods_amber qty_green, qty_amber = cfg.qty_goods_green, cfg.qty_goods_amber else: price_green, price_amber = cfg.price_services_green, cfg.price_services_amber qty_green, qty_amber = cfg.qty_services_green, cfg.qty_services_amber def single_zone(pct: Optional[Decimal], green: Decimal, amber: Decimal) -> Optional[str]: if pct is None: return None if pct <= green: return "green" if pct <= amber: return "amber" return "red" zones = [ single_zone(price_pct, price_green, price_amber), single_zone(qty_pct, qty_green, qty_amber), ] zones = [z for z in zones if z is not None] if not zones: return "green" if "red" in zones: return "red" if "amber" in zones: return "amber" return "green" def worse_zone(a: str, b: str) -> str: rank = {"green": 0, "amber": 1, "red": 2} return a if rank[a] >= rank[b] else b def _price_bands(item_type: Optional[str], cfg: TwoWayConfig) -> 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 _qty_bands(item_type: Optional[str], cfg: TwoWayConfig) -> tuple[Decimal, Decimal]: item_type = (item_type or "services").lower() if item_type == "goods": return cfg.qty_goods_green, cfg.qty_goods_amber return cfg.qty_services_green, cfg.qty_services_amber def rapidfuzz_best_description_match( query: str, candidates: list[dict[str, Any]], *, high_threshold: float = 90.0, medium_threshold: float = 80.0, score_cutoff: float = 80.0, ) -> dict[str, Any]: """ Returns: { "decision": "matched" | "suggestion" | "unmatched", "score_tier": "high" | "medium" | "low" | None, "score": float, "record_id": int | None } """ 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 cand_name in names: if not cand_name: continue score = float( fuzz.WRatio( query, cand_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: score_tier = "high" elif best_score >= medium_threshold: score_tier = "medium" else: score_tier = "low" decision = "matched" if score_tier in {"high", "medium"} else "suggestion" return { "decision": decision, "score_tier": score_tier, "score": round(best_score, 2), "record_id": best_record_id, } # ============================================================ # Main 2-way matching function # ============================================================ def run_two_way_match_core( invoice: InvoiceHeader, inv_lines: list[InvoiceLine], po: POHeader, po_lines: list[POLine], *, matching_configs: Optional[dict] = None, merge_chain_fn: Optional[Callable[[Optional[int], Optional[int]], bool]] = None, lock_candidate_po_lines_fn: Optional[Callable[[list[int]], None]] = None, ) -> TwoWayMatchResult: """ Core 2-way matching engine. Implements: - entry precondition on invoice.status - header checks: currency, vendor merge-chain aware, PO value ceiling advisory - Tier 1: SKU exact canonical match - Tier 2: RapidFuzz description match - Tier 3: GL exact + price tolerance - Tier 4: aggregate GL matching over residue - variance computation and zones - exceptions and advisory flags - analytical pool consumption only (no DB writes) """ cfg = _cfg_from_dict(matching_configs or {}) getcontext().prec = cfg.precision result = TwoWayMatchResult() result.lines_total = len(inv_lines) result.advisory_flags = list(invoice.advisory_flags) invoice_currency = norm_currency(invoice.currency) po_currency = norm_currency(po.currency) # --------------------------------------------------------- # Entry preconditions # --------------------------------------------------------- if invoice.status not in {"validated", "matching"}: raise ValueError( f"Precondition violation: invoice.status={invoice.status!r} " f"must be in ('validated', 'matching')." ) # --------------------------------------------------------- # Header Check 1: Currency # --------------------------------------------------------- currency_ok = True if invoice_currency is None: currency_ok = False result.skipped_steps.append({"step": "currency_check", "skip_reason": "ef_currency_null"}) elif po_currency is None or invoice_currency != po_currency: currency_ok = False result.header_exceptions.append( MatchExceptionRecord( exception_type="currency_mismatch", exception_detail={ "invoice_currency": invoice_currency, "po_currency": po_currency, }, ) ) # --------------------------------------------------------- # Header Check 2: Vendor (merge-chain aware) # --------------------------------------------------------- vendor_ok = same_vendor_merge_chain(invoice.vendor_id, po.vendor_id, merge_chain_fn) if not vendor_ok: result.header_exceptions.append( MatchExceptionRecord( exception_type="vendor_mismatch", exception_detail={ "invoice_vendor_id": invoice.vendor_id, "po_vendor_id": po.vendor_id, "merge_chain_aware": bool(merge_chain_fn is not None), }, ) ) # --------------------------------------------------------- # Header Check 3: PO value ceiling (advisory only) # --------------------------------------------------------- if not currency_ok: result.skipped_steps.append({"step": "po_value_ceiling", "skip_reason": "currency_mismatch_or_null"}) elif invoice.total_amount is not None: rem_value = remaining_po_value(po_lines) if invoice.total_amount > rem_value: result.advisory_flags.append("exceeds_remaining_po_value_at_match") # --------------------------------------------------------- # Build matching pool # --------------------------------------------------------- pool: list[POLine] = [ pl for pl in po_lines if pl.deleted_at is None and pl.status in {"open", "fully_invoiced"} ] # Optional lock hook for caller-side DB advisory locks if lock_candidate_po_lines_fn is not None: lock_candidate_po_lines_fn([pl.id for pl in sorted(pool, key=lambda x: x.id)]) consumed_po_line_ids: set[int] = set() def pool_remaining() -> list[POLine]: return [pl for pl in pool if pl.id not in consumed_po_line_ids] def consume_po_line(po_line_id: int) -> None: consumed_po_line_ids.add(po_line_id) # --------------------------------------------------------- # Tier 1-3 per-line waterfall # --------------------------------------------------------- residue_lines: list[InvoiceLine] = [] line_exception_map: dict[int, list[MatchExceptionRecord]] = {} item_type_unresolved_flagged = False def add_line_exception(line_id: int, exc: MatchExceptionRecord) -> None: line_exception_map.setdefault(line_id, []).append(exc) result.match_exceptions.append(exc) for inv_line in inv_lines: inv_sku_canon = inv_line.sku_canonical or canonicalise_identifier(inv_line.sku) inv_desc_canon = inv_line.description_canonical or canonicalise_identifier(inv_line.description) matched_this_line = False # ------------------------- # Tier 1: SKU exact match # ------------------------- if inv_sku_canon: candidates = [ pl for pl in pool_remaining() if (pl.sku_canonical or canonicalise_identifier(pl.sku)) == inv_sku_canon ] candidates.sort(key=lambda x: x.id) if candidates: chosen = candidates[0] if chosen.item_type is None and not item_type_unresolved_flagged: result.advisory_flags.append("item_type_unresolved") item_type_unresolved_flagged = True matched, line_result = _finalize_tier1_3_line_match( inv_line=inv_line, po_line=chosen, tier=1, cfg=cfg, ) if matched: result.match_line_results.append(line_result) consume_po_line(chosen.id) matched_this_line = True else: add_line_exception( inv_line.id, MatchExceptionRecord( exception_type="po_ceiling_exhausted", line_id=inv_line.id, po_line_id=chosen.id, exception_detail={ "reason": "available_qty_zero_or_negative", "po_line_id": chosen.id, }, ), ) matched_this_line = True if matched_this_line: continue # ------------------------- # Tier 2: RapidFuzz description match # ------------------------- pool_now = pool_remaining() if inv_desc_canon and pool_now: tier2_candidates = [ { "record_id": pl.id, "names": [ pl.description_canonical or canonicalise_identifier(pl.description) or "" ], } for pl in pool_now ] tier2_candidates = [c for c in tier2_candidates if c["names"][0]] if tier2_candidates: rf_result = rapidfuzz_best_description_match( query=inv_desc_canon, candidates=tier2_candidates, high_threshold=cfg.fuzzy_high_threshold, medium_threshold=cfg.fuzzy_medium_threshold, score_cutoff=cfg.fuzzy_score_cutoff, ) if rf_result["decision"] == "matched" and rf_result["record_id"] is not None: chosen = next((pl for pl in pool_now if pl.id == rf_result["record_id"]), None) if chosen is not None: if chosen.item_type is None and not item_type_unresolved_flagged: result.advisory_flags.append("item_type_unresolved") item_type_unresolved_flagged = True matched, line_result = _finalize_tier1_3_line_match( inv_line=inv_line, po_line=chosen, tier=2, cfg=cfg, ) if matched: result.match_line_results.append(line_result) consume_po_line(chosen.id) matched_this_line = True else: add_line_exception( inv_line.id, MatchExceptionRecord( exception_type="po_ceiling_exhausted", line_id=inv_line.id, po_line_id=chosen.id, exception_detail={ "reason": "available_qty_zero_or_negative", "po_line_id": chosen.id, }, ), ) matched_this_line = True if matched_this_line: continue # ------------------------- # Tier 3: GL exact + price tolerance # ------------------------- tier3_candidates: list[POLine] = [] if inv_line.gl_account is not None: for pl in pool_remaining(): if pl.gl_account is None: continue if inv_line.gl_account != pl.gl_account: continue if pl.unit_price is None or pl.unit_price == 0: continue tier3_candidates.append(pl) tier3_candidates.sort(key=lambda x: x.id) for chosen in tier3_candidates: p_var = price_variance_pct(to_decimal(inv_line.unit_price), to_decimal(chosen.unit_price)) if p_var is None: continue if p_var <= (cfg.tier3_price_variance_threshold * D100): if chosen.item_type is None and not item_type_unresolved_flagged: result.advisory_flags.append("item_type_unresolved") item_type_unresolved_flagged = True matched, line_result = _finalize_tier1_3_line_match( inv_line=inv_line, po_line=chosen, tier=3, cfg=cfg, ) if matched: result.match_line_results.append(line_result) consume_po_line(chosen.id) matched_this_line = True else: add_line_exception( inv_line.id, MatchExceptionRecord( exception_type="po_ceiling_exhausted", line_id=inv_line.id, po_line_id=chosen.id, exception_detail={ "reason": "available_qty_zero_or_negative", "po_line_id": chosen.id, }, ), ) matched_this_line = True break if matched_this_line: continue # No Tier 1-3 hit residue_lines.append(inv_line) # --------------------------------------------------------- # Tier 4: Aggregate GL match on residue # --------------------------------------------------------- residue_lines_by_gl: dict[str, list[InvoiceLine]] = {} for line in residue_lines: if line.gl_account is None: add_line_exception( line.id, MatchExceptionRecord( exception_type="no_match", line_id=line.id, exception_detail={ "reason": "gl_account_null_tier4_skip", }, ), ) continue residue_lines_by_gl.setdefault(line.gl_account, []).append(line) residue_po_lines = pool_remaining() po_lines_by_gl: dict[str, list[POLine]] = {} for pl in residue_po_lines: if pl.gl_account is None: continue po_lines_by_gl.setdefault(pl.gl_account, []).append(pl) for gl, group_inv_lines in residue_lines_by_gl.items(): group_po_lines = po_lines_by_gl.get(gl, []) if not group_po_lines: for il in group_inv_lines: add_line_exception( il.id, MatchExceptionRecord( exception_type="no_match", line_id=il.id, exception_detail={ "reason": "no_po_lines_in_same_gl_group", "gl_account": gl, }, ), ) continue group_inv_lines.sort(key=lambda x: x.id) group_po_lines.sort(key=lambda x: x.id) # Tier 4 group cap if len(group_inv_lines) > cfg.tier4_gl_group_max_lines: for il in group_inv_lines: add_line_exception( il.id, MatchExceptionRecord( exception_type="no_match", line_id=il.id, exception_detail={ "reason": "tier4_group_cap_exceeded", "group_gl_account": gl, "group_line_count": len(group_inv_lines), "cap": cfg.tier4_gl_group_max_lines, }, ), ) continue # Invoice amount sum if all(il.amount is None for il in group_inv_lines): for il in group_inv_lines: add_line_exception( il.id, MatchExceptionRecord( exception_type="no_match", line_id=il.id, exception_detail={ "reason": "all_invoice_amounts_null", "group_gl_account": gl, }, ), ) continue invoice_sum = sum((il.amount if il.amount is not None else D0 for il in group_inv_lines), D0) # PO group sum — SDD §4.3: status='open' only. # any_goods checks all lines (fully_invoiced included) because goods-first is a # type-classification rule, not an availability rule. open_group_po_lines = [pl for pl in group_po_lines if pl.status == "open"] po_group_sum = D0 any_goods = False for pl in group_po_lines: if (pl.item_type or "").lower() == "goods": any_goods = True for pl in open_group_po_lines: rem_qty = available_qty(pl) up = to_decimal(pl.unit_price) if up is None: continue po_group_sum += rem_qty * up if any(pl.item_type is None for pl in group_po_lines) and not item_type_unresolved_flagged: result.advisory_flags.append("item_type_unresolved") item_type_unresolved_flagged = True if po_group_sum == 0: all_zero_price = ( bool(open_group_po_lines) and all((to_decimal(pl.unit_price) or D0) == D0 for pl in open_group_po_lines) ) exc_type = "zero_price_po_line_at_match" if all_zero_price else "no_match" for il in group_inv_lines: add_line_exception( il.id, MatchExceptionRecord( exception_type=exc_type, line_id=il.id, exception_detail={ "reason": "po_group_sum_zero", "group_gl_account": gl, "po_line_ids": [pl.id for pl in group_po_lines], }, ), ) continue group_variance_pct = (abs(invoice_sum - po_group_sum) / po_group_sum * D100).quantize(Decimal("0.0001")) group_item_type = "goods" if any_goods else "services" # Quantity sanity (before accepting Tier 4) inv_qty_sum = sum((to_decimal(il.quantity) or D0 for il in group_inv_lines), D0) po_available_qty_sum = sum((available_qty(pl) for pl in group_po_lines), D0) if inv_qty_sum > 0 and po_available_qty_sum > 0: qty_var = qty_over_pct(inv_qty_sum, po_available_qty_sum) if qty_var is not None: _, qty_amber = _qty_bands(group_item_type, cfg) if qty_var > qty_amber: for il in group_inv_lines: add_line_exception( il.id, MatchExceptionRecord( exception_type="no_match", line_id=il.id, exception_detail={ "reason": "tier4_qty_sanity_failed", "group_gl_account": gl, "inv_qty_sum": str(inv_qty_sum), "po_available_qty_sum": str(po_available_qty_sum), "qty_over_pct": str(qty_var), }, ), ) continue # Price tolerance for Tier 4 price_green, price_amber = _price_bands(group_item_type, cfg) if group_variance_pct <= price_amber: # Match the whole group and persist deterministic proportional allocations. for il in group_inv_lines: il_amount = il.amount if il.amount is not None else D0 raw_allocations: list[tuple[POLine, Decimal]] = [] for pl in open_group_po_lines: rem_qty = available_qty(pl) up = to_decimal(pl.unit_price) or D0 allocated_po_amount = rem_qty * up raw_allocations.append((pl, allocated_po_amount)) po_sum = sum((amt for _, amt in raw_allocations), D0) if po_sum == 0: add_line_exception( il.id, MatchExceptionRecord( exception_type="zero_price_po_line_at_match", line_id=il.id, exception_detail={ "reason": "po_sum_zero_in_allocation", "group_gl_account": gl, "po_line_ids": [pl.id for pl, _ in raw_allocations], }, ), ) continue quant = Decimal("0.01") alloc_rows: list[tuple[POLine, Decimal, Decimal]] = [] for pl, allocated_po_amount in raw_allocations: allocated_inv_amount = (allocated_po_amount / po_sum * il_amount).quantize(quant) alloc_rows.append((pl, allocated_po_amount, allocated_inv_amount)) allocated_inv_total = sum((row[2] for row in alloc_rows), D0) residue = (il_amount - allocated_inv_total).quantize(quant) if residue != 0: # Largest allocated PO amount, tie-break by smallest po_line.id target_idx = min( range(len(alloc_rows)), key=lambda i: (-alloc_rows[i][1], alloc_rows[i][0].id), ) pl, a_po, a_inv = alloc_rows[target_idx] alloc_rows[target_idx] = (pl, a_po, (a_inv + residue).quantize(quant)) group_line_zone = "green" if group_variance_pct <= price_green else ("amber" if group_variance_pct <= price_amber else "red") for pl, allocated_po_amount, allocated_inv_amount in alloc_rows: result.match_line_results.append( MatchLineResult( invoice_line_id=il.id, po_line_id=pl.id, match_tier=4, allocated_inv_amount=allocated_inv_amount, allocated_po_amount=allocated_po_amount.quantize(Decimal("0.01")), price_var_pct=group_variance_pct, qty_var_pct=None, line_zone=group_line_zone, line_status="matched", match_type="two_way", ) ) for pl in group_po_lines: consume_po_line(pl.id) continue # Outside tolerance -> no_match for every invoice line in the group for il in group_inv_lines: add_line_exception( il.id, MatchExceptionRecord( exception_type="no_match", line_id=il.id, exception_detail={ "reason": "tier4_outside_tolerance", "group_gl_account": gl, "invoice_sum": str(invoice_sum), "po_sum": str(po_group_sum), "variance_pct": str(group_variance_pct), "po_line_ids": [pl.id for pl in group_po_lines], }, ), ) # --------------------------------------------------------- # Final summary # --------------------------------------------------------- matched_line_ids = {r.invoice_line_id for r in result.match_line_results if r.line_status == "matched"} result.lines_matched = len(matched_line_ids) exception_line_ids = set(line_exception_map.keys()) # Any invoice line not matched and not explicitly captured is a no_match exception for line in inv_lines: if line.id not in matched_line_ids and line.id not in exception_line_ids: exception_line_ids.add(line.id) result.lines_in_exception = len(exception_line_ids) # Final overall status and zone if result.header_exceptions: result.overall_status = "exception" result.match_zone = "red" else: any_amber = any(r.line_zone == "amber" for r in result.match_line_results) any_red = any(r.line_zone == "red" for r in result.match_line_results) if result.lines_matched == result.lines_total and result.lines_in_exception == 0: result.overall_status = "matched" result.match_zone = "red" if any_red else ("amber" if any_amber else "green") elif result.lines_matched > 0: result.overall_status = "partial_match" if any_red or result.match_exceptions: result.match_zone = "red" elif any_amber: result.match_zone = "amber" else: result.match_zone = "green" else: result.overall_status = "exception" result.match_zone = "red" # documents.status is always pending_approval after task_match result.documents_status = "pending_approval" result.match_type = "two_way" return result # ============================================================ # Internal helper for Tier 1-3 finalization # ============================================================ def _finalize_tier1_3_line_match( *, inv_line: InvoiceLine, po_line: POLine, tier: int, cfg: TwoWayConfig, ) -> tuple[bool, MatchLineResult]: """ Finalize a Tier 1-3 match candidate. Returns: (matched_successfully, result_row) If available_qty == 0, the line is blocked with po_ceiling_exhausted. """ inv_qty = to_decimal(inv_line.quantity) inv_price = to_decimal(inv_line.unit_price) po_price = to_decimal(po_line.unit_price) avail_qty = available_qty(po_line) if avail_qty < 0: return False, MatchLineResult( invoice_line_id=inv_line.id, po_line_id=po_line.id, match_tier=tier, line_zone="red", line_status="exception", match_type="two_way", ) if avail_qty == 0: return False, MatchLineResult( invoice_line_id=inv_line.id, po_line_id=po_line.id, match_tier=tier, price_var_pct=price_variance_pct(inv_price, po_price), qty_var_pct=None, line_zone="red", line_status="exception", match_type="two_way", ) p_var = price_variance_pct(inv_price, po_price) if inv_qty is None: q_var = None else: q_var = qty_over_pct(inv_qty, avail_qty) item_type = (po_line.item_type or "services").lower() line_zone = zone_from_pct(p_var, q_var, item_type, cfg) allocated_inv_amount = inv_line.amount if inv_line.amount is not None else None allocated_po_amount = ((to_decimal(po_line.quantity) or D0) - (to_decimal(po_line.invoiced_qty) or D0)) * (po_price or D0) return True, MatchLineResult( invoice_line_id=inv_line.id, po_line_id=po_line.id, match_tier=tier, allocated_inv_amount=allocated_inv_amount, allocated_po_amount=allocated_po_amount.quantize(Decimal("0.01")), price_var_pct=p_var, qty_var_pct=q_var, line_zone=line_zone, line_status="matched", match_type="two_way", )