from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Optional, Union from two_way_match import ( InvoiceLine, MatchExceptionRecord, POHeader, POLine, TwoWayMatchResult, ) from three_way_match import GRLineItem from contract_match import ContractMatchResult # ============================================================ # Precondition violation # ============================================================ class MatchPreconditionError(Exception): """Raised when document.status is not 'validated' or 'matching' (SDD §3.1).""" # ============================================================ # Routing input models # ============================================================ @dataclass(frozen=True) class RoutingDocument: id: int tenant_id: int entity_id: int status: str # 'validated' | 'matching' — enforced by route_invoice doc_type: str # 'invoice' | 'credit_note' | ... @dataclass(frozen=True) class RoutingExtractedFields: po_number_canonical: Optional[str] resolved_vendor_id: Optional[int] # ============================================================ # Routing decision # ============================================================ MatchResult = Union[TwoWayMatchResult, ContractMatchResult] @dataclass class RouteDecision: route: str # 'two_way' | 'three_way' | 'contract' | 'stub' po: Optional[POHeader] = None # populated for two_way / three_way po_lines: Optional[list[POLine]] = None gr_rows: Optional[list[GRLineItem]] = None # populated for three_way stub_result: Optional[MatchResult] = None # populated when route='stub' # ============================================================ # Stub builders (SDD §3.4) # ============================================================ def _two_way_stub( inv_lines: list[InvoiceLine], exception_type: str, exception_detail: dict[str, Any], ) -> TwoWayMatchResult: res = TwoWayMatchResult( match_type="two_way", 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=exception_detail, ) ) return res def _contract_stub( inv_lines: list[InvoiceLine], exception_type: str, exception_detail: dict[str, Any], ) -> ContractMatchResult: res = ContractMatchResult( match_type="contract", 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=exception_detail, ) ) return res # ============================================================ # Status sets # ============================================================ _VALID_ENTRY_STATUSES = {"validated", "matching"} _PO_ACTIVE_STATUSES = {"open", "partial"} _PO_LINE_LIVE_STATUSES = {"open", "fully_invoiced"} _PO_LINE_DEAD_STATUSES = {"closed", "cancelled"} # ============================================================ # Decision tree (SDD §3.3) # ============================================================ def route_invoice( document: RoutingDocument, ef: RoutingExtractedFields, inv_lines: list[InvoiceLine], po: Optional[POHeader], po_lines: Optional[list[POLine]], gr_rows: Optional[list[GRLineItem]], ) -> RouteDecision: """ Deterministic match-type routing (SDD §3.3). First match wins. Caller responsibilities: - Acquire the advisory lock before calling (§3.1). - Pre-fetch po / po_lines from the DB (pass None if po_number_canonical is NULL or the PO row was not found). - Pre-fetch gr_rows (pass empty list or None if no GR rows exist for the PO). - After receiving RouteDecision, handle prior-match supersession (§3.5) before dispatching to the match function. Raises: MatchPreconditionError — document.status not in {'validated', 'matching'} (§3.1). """ # §3.1 — Entry precondition if document.status not in _VALID_ENTRY_STATUSES: raise MatchPreconditionError( f"document {document.id} has status={document.status!r}; " f"expected one of {sorted(_VALID_ENTRY_STATUSES)}" ) # §3.3 rule 2 — PO-based routing if ef.po_number_canonical is not None: if po is None: return RouteDecision( route="stub", stub_result=_two_way_stub( inv_lines, "po_not_found", { "reason": "no_purchase_order_matched_canonical", "tenant_id": document.tenant_id, "entity_id": document.entity_id, "po_number_canonical": ef.po_number_canonical, }, ), ) if po.status == "cancelled": return RouteDecision( route="stub", stub_result=_two_way_stub( inv_lines, "po_cancelled", { "reason": "purchase_order_is_cancelled", "po_id": po.id, "po_number_canonical": ef.po_number_canonical, }, ), ) lines: list[POLine] = po_lines or [] if po.status == "closed" and all( pl.status in _PO_LINE_DEAD_STATUSES for pl in lines ): return RouteDecision( route="stub", stub_result=_two_way_stub( inv_lines, "po_closed", { "reason": "purchase_order_closed_all_lines_dead", "po_id": po.id, "po_number_canonical": ef.po_number_canonical, "line_statuses": [pl.status for pl in lines], }, ), ) # §3.3 table: open/partial + every line closed/cancelled → po_all_lines_closed if po.status in _PO_ACTIVE_STATUSES and all( pl.status in _PO_LINE_DEAD_STATUSES for pl in lines ): return RouteDecision( route="stub", stub_result=_two_way_stub( inv_lines, "po_all_lines_closed", { "reason": "all_po_lines_closed_or_cancelled", "po_id": po.id, "po_number_canonical": ef.po_number_canonical, "line_statuses": [pl.status for pl in lines], }, ), ) # §3.3: at least one live line (open or fully_invoiced) must exist to proceed. # fully_invoiced lines reach matching (not stub) so credit-note reversals and # po_ceiling_exhausted can be raised correctly by §4 / §5. has_live_line = any(pl.status in _PO_LINE_LIVE_STATUSES for pl in lines) if not has_live_line: return RouteDecision( route="stub", stub_result=_two_way_stub( inv_lines, "po_all_lines_closed", { "reason": "no_live_po_lines_found", "po_id": po.id, "po_number_canonical": ef.po_number_canonical, "line_statuses": [pl.status for pl in lines], }, ), ) # §3.3: any GR row (any inspection_status) → three-way. # §5 adjudicates no_accepted_gr / inspection_rejected / timing_fraud internally. if gr_rows: return RouteDecision( route="three_way", po=po, po_lines=lines, gr_rows=gr_rows, ) return RouteDecision( route="two_way", po=po, po_lines=lines, ) # §3.3 rule 3 — No PO, no vendor anchor if ef.resolved_vendor_id is None: return RouteDecision( route="stub", stub_result=_contract_stub( inv_lines, "cannot_match_no_vendor", { "reason": "resolved_vendor_id_is_null_and_no_po", "tenant_id": document.tenant_id, "entity_id": document.entity_id, }, ), ) # §3.3 rule 4 — No PO, vendor resolved → contract match return RouteDecision(route="contract")