dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

src/brick_purchasing/domain/matching.py

"""The three-way match (order <-> receipt <-> bill), DERIVED on read and on write; never a stored status (PUR-11, PUR-36, PUR-52).

THE control of this module. Price is compared per unit of the order line's PRICING unit; quantity in that pricing unit through the line's
own fixed conversion, so a bill in kilograms matches an order in bars and a receipt in bars. Every comparison is an exact Fraction,
rounded only where an amount is shown. Each failure names its tolerance key (PP price, DQ quantity, AN a line with no order, TX tax,
KW delivery cost, BD small difference) for the supplier's tolerance group, in force on the invoice date."""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional

from brick_purchasing.domain.values import as_dec, exact_str, fr, round_half_up

STANDARD = "STANDARD"


@dataclass(frozen=True)
class Tol:
    lower_amount: Optional[Decimal] = None
    upper_amount: Optional[Decimal] = None
    lower_pct: Optional[Decimal] = None
    upper_pct: Optional[Decimal] = None

    def view(self) -> dict:
        return {k: (None if v is None else str(v)) for k, v in self.__dict__.items()}


def tolerance_set(rows: list, group: str, on: date, defaults: dict) -> dict:
    """{key: Tol} in force on `on` for the supplier's group. Each (group, key) takes its latest row valid_from <= on; a key the
    group has no row for falls back to the STANDARD group's row, then to the setting-derived default."""
    out = dict(defaults)
    for g in (STANDARD, group or STANDARD):
        latest = {}
        for r in rows:
            if r["tolerance_group"] == g and r["valid_from"] <= on and (r["key"] not in latest or r["valid_from"] > latest[r["key"]]["valid_from"]):
                latest[r["key"]] = r
        for k, r in latest.items():
            out[k] = Tol(r.get("lower_amount"), r.get("upper_amount"), r.get("lower_pct"), r.get("upper_pct"))
    return out


def allowance(tol: Tol, base_amount: Fraction) -> Fraction:
    """The most a DIFFERENCE upward may be worth: the LARGER of the percent of `base_amount` and the amount (signed decision 2026-10-22: "2% or 0.00 per line, whichever is larger"; the PUR-52 text saying "stricter" is a spec typo); none set: nothing."""
    cands = []
    if tol.upper_pct:
        cands.append(base_amount * fr(tol.upper_pct) / 100)
    if tol.upper_amount:
        cands.append(fr(tol.upper_amount))
    return max(cands) if cands else Fraction(0)


def lower_allowance(tol: Tol, base_amount: Fraction) -> Optional[Fraction]:
    cands = []
    if tol.lower_pct:
        cands.append(base_amount * fr(tol.lower_pct) / 100)
    if tol.lower_amount:
        cands.append(fr(tol.lower_amount))
    return max(cands) if cands else None


def _check(kind: str, key: str, passed: bool, difference: Fraction, tol: Tol, **more) -> dict:
    return {"kind": kind, "key": key, "passed": passed, "difference": exact_str(difference), "difference_shown": str(as_dec(difference, 4)),
            "tolerance": tol.view(), **more}


def price_per_unit(line, order_line) -> Fraction:
    """The bill's price per ONE unit of the order line's pricing unit: via the base unit, exact."""
    per_bill_unit = fr(line.unit_price) / int(line.pricing_quantity or 1)
    per_base = per_bill_unit / Fraction(line.pricing_numerator, line.pricing_denominator)
    return per_base * Fraction(order_line.pricing_numerator, order_line.pricing_denominator)


def billed_priced(line, order_line) -> Fraction:
    """The bill line's quantity in the order line's pricing unit."""
    base = fr(line.quantity) * Fraction(line.unit_numerator, line.unit_denominator)
    return base * Fraction(order_line.pricing_denominator, order_line.pricing_numerator)


def match_line(line, order_line, *, basis_base: Optional[Fraction], claimed_other: Fraction, tol: dict, credit: bool = False) -> dict:
    """Checks for one bill line against its order line. `basis_base`: what may be billed in the base unit (received for a
    three-way match, ordered for a two-way, returned for a credit); None when the line has no basis (a service with no receipt)."""
    checks = []
    op = fr(order_line.unit_price) / int(order_line.pricing_quantity or 1)
    bp = price_per_unit(line, order_line)
    qty_priced = billed_priced(line, order_line)
    diff = bp - op                                    # per one pricing unit, exact
    line_diff = diff * qty_priced                     # in the bill currency
    t = tol.get("PP", Tol())
    up = allowance(t, op * qty_priced)
    low = lower_allowance(t, op * qty_priced)
    pct = (diff / op * 100) if op else Fraction(0)
    ok_price = line_diff <= up and (low is None or -line_diff <= low)
    checks.append(_check("price", "PP", ok_price, diff, t, difference_pct=str(round_half_up(pct, 2)), amount=str(as_dec(line_diff, 2)),
                         order_price=str(as_dec(op, 4)), bill_price=str(as_dec(bp, 4))))
    if basis_base is not None:
        basis_priced = (basis_base - claimed_other) * Fraction(order_line.pricing_denominator, order_line.pricing_numerator)
        excess = qty_priced - basis_priced
        qt = tol.get("DQ", Tol())
        allow = (basis_priced * fr(qt.upper_pct) / 100 if qt.upper_pct else Fraction(0))
        if qt.upper_amount:
            allow = max(allow, fr(qt.upper_amount))
        checks.append(_check("quantity", "DQ", excess <= max(allow, Fraction(0)), excess, qt,
                             available=str(as_dec(max(basis_priced, Fraction(0)), 6)), billed=str(as_dec(qty_priced, 6))))
    return {"line_no": line.line_no, "order_line_no": line.order_line_no, "checks": checks, "passed": all(c["passed"] for c in checks),
            "amount_difference": line_diff}


def match_unordered(line, tol: dict) -> dict:
    """A line with no order line: tolerance AN (default nothing) on its net."""
    t = tol.get("AN", Tol())
    allowed = fr(t.upper_amount) if t.upper_amount else Fraction(0)
    over = fr(line.net_amount) - allowed
    return {"line_no": line.line_no, "order_line_no": None, "passed": over <= 0, "amount_difference": Fraction(0),
            "checks": [_check("amount", "AN", over <= 0, fr(line.net_amount), t, note="a line with no order line")]}


def match_delivery_cost(line, planned: Optional[Fraction], tol: dict) -> dict:
    """A freight or duty line against its planned cost: key KW (percent of the plan)."""
    t = tol.get("KW", Tol())
    if planned is None:
        return {"line_no": line.line_no, "order_line_no": line.order_line_no, "passed": False, "amount_difference": Fraction(0),
                "checks": [_check("delivery_cost", "KW", False, fr(line.net_amount), t, note="no planned delivery cost to match")]}
    diff = fr(line.net_amount) - planned
    allow = planned * fr(t.upper_pct) / 100 if t.upper_pct else Fraction(0)
    if t.upper_amount:
        allow = max(allow, fr(t.upper_amount))
    pct = diff / planned * 100 if planned else Fraction(0)
    return {"line_no": line.line_no, "order_line_no": line.order_line_no, "passed": diff <= allow, "amount_difference": diff,
            "checks": [_check("delivery_cost", "KW", diff <= allow, diff, t, difference_pct=str(round_half_up(pct, 2)),
                              planned=str(as_dec(planned, 2)))]}


def match_withdrawals(line, withdrawal_amount: Optional[Fraction]) -> dict:
    """A consignment line billed from withdrawals: two-way against the withdrawals' amount, exact."""
    t = Tol()
    if withdrawal_amount is None:
        return {"line_no": line.line_no, "order_line_no": line.order_line_no, "passed": False, "amount_difference": Fraction(0),
                "checks": [_check("amount", "AN", False, fr(line.net_amount), t, note="no withdrawal to bill")]}
    diff = fr(line.net_amount) - withdrawal_amount
    return {"line_no": line.line_no, "order_line_no": line.order_line_no, "passed": diff == 0, "amount_difference": diff,
            "checks": [_check("amount", "AN", diff == 0, diff, t, note="against the consignment withdrawals")]}


def match_tax(supplier_tax: Optional[Decimal], computed_tax: Decimal, tol: dict) -> Optional[dict]:
    """The supplier's stated VAT against the computed VAT: tolerance TX (an amount). Only when the supplier stated one."""
    if supplier_tax is None:
        return None
    t = tol.get("TX", Tol())
    diff = abs(fr(supplier_tax) - fr(computed_tax))
    allow = fr(t.upper_amount) if t.upper_amount else Fraction(0)
    return {"line_no": 0, "order_line_no": None, "passed": diff <= allow, "amount_difference": Fraction(0),
            "checks": [_check("tax", "TX", diff <= allow, diff, t, supplier=str(supplier_tax), computed=str(computed_tax))]}


def small_difference(results: list, tol: dict) -> Optional[Fraction]:
    """The total difference to take as a SMALL DIFFERENCE (BD): when every failure is a price failure and their amounts together are
    within the BD amount, the system accepts them. None when it does not apply."""
    bd = tol.get("BD", Tol())
    if not bd.upper_amount:
        return None
    failed = [(r, c) for r in results for c in r["checks"] if not c["passed"]]
    if not failed or any(c["kind"] != "price" for _, c in failed):
        return None
    total = sum((abs(r["amount_difference"]) for r, _ in failed), Fraction(0))
    return total if total <= fr(bd.upper_amount) else None


def failures(results: list) -> list:
    """[(line_no, kind, key)] of every failed check."""
    return [(r["line_no"], c["kind"], c["key"]) for r in results for c in r["checks"] if not c["passed"]]


def view(results: list) -> list:
    return [{k: v for k, v in r.items() if k != "amount_difference"} | {"amount_difference": str(as_dec(r["amount_difference"], 2))}
            for r in results]