dodil/erp-brick-purchasing

DODIL ERP brick: purchasing

src/brick_purchasing/application/rules.py

"""Rules that read rows: the source list and the quota (PUR-44, PUR-45), what has been received and billed of an order (derived, never a
flag), and what is outstanding. Pure functions of a repository."""
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional

from brick_purchasing.domain.readiness import Gate, no, ok
from brick_purchasing.domain.values import as_dec, fr


def _in_force(rows: list, on: date) -> dict:
    """{bp_id: the source row in force on `on`} for a product's source list: of each supplier's rows the latest valid_from <= on, still
    running (valid_to empty or on/after `on`)."""
    best = {}
    for r in rows:
        if r["valid_from"] > on:
            continue
        if r["bp_id"] not in best or r["valid_from"] > best[r["bp_id"]]["valid_from"]:
            best[r["bp_id"]] = r
    return {bp: r for bp, r in best.items() if r.get("valid_to") is None or r["valid_to"] >= on}


def source_info(repo, product_id: str, on: date) -> dict:
    rows = repo.sources.all(product_id=product_id)
    force = _in_force(rows, on)
    required = any(r.get("required") for r in rows if r["valid_from"] <= on)
    return {"required": required, "allowed": sorted(bp for bp, r in force.items() if not r.get("blocked")),
            "fixed": sorted(bp for bp, r in force.items() if r.get("fixed") and not r.get("blocked")),
            "blocked": {bp: r["valid_from"] for bp, r in force.items() if r.get("blocked")}}


def source_gate(repo, product_id: str, bp_id: int, on: date) -> Gate:
    """source_allowed: a blocked supplier is never ordered from; where the product REQUIRES a source list, only a supplier on it."""
    if not product_id:
        return ok("source_allowed")
    info = source_info(repo, product_id, on)
    names = " or ".join(str(b) for b in info["allowed"]) or "nobody yet"
    if int(bp_id) in info["blocked"]:
        return no("source_allowed", f"{product_id} may be bought from {names}; {bp_id} is blocked from {info['blocked'][int(bp_id)]}",
                  "pick an allowed supplier, or ask the buyer who keeps the list (PUR-44)")
    if info["required"] and int(bp_id) not in info["allowed"]:
        return no("source_allowed", f"{product_id} may be bought from {names}; {bp_id} is not on the source list",
                  "pick an allowed supplier, or ask the buyer who keeps the list (PUR-44)")
    return ok("source_allowed")


def quota_proposal(repo, product_id: str, quantity_base: Decimal, on: date) -> Optional[dict]:
    """The supplier the quota proposes: the lowest rating (allocated + base) / quota; allocated is DERIVED from confirmed order lines since the
    quota's valid_from, never a counter. A blocked source is never proposed. None: no quota in force."""
    rows = [r for r in repo.quotas.all(product_id=product_id) if r["valid_from"] <= on]
    if not rows:
        return None
    start = max(r["valid_from"] for r in rows)
    mine = [r for r in rows if r["valid_from"] == start]
    blocked = source_info(repo, product_id, on)["blocked"]
    cands = []
    for r in mine:
        allocated = repo.orders.confirmed_quantity_since(product_id, r["bp_id"], start)
        quota = fr(r["quota_pct"])
        rating = (fr(allocated) + fr(r["base_quantity"] or 0)) / quota if quota else None
        cands.append({"bp_id": r["bp_id"], "quota_pct": r["quota_pct"], "allocated": allocated, "rating": None if rating is None else as_dec(rating, 4),
                      "blocked": r["bp_id"] in blocked, "_r": rating})
    open_ = [c for c in cands if not c["blocked"] and c["_r"] is not None]
    if not open_:
        return {"proposed": None, "candidates": [{k: v for k, v in c.items() if k != "_r"} for c in cands]}
    best = min(open_, key=lambda c: (c["_r"], c["bp_id"]))
    return {"proposed": {"bp_id": best["bp_id"], "rating": best["rating"]}, "valid_from": start,
            "candidates": [{k: v for k, v in c.items() if k != "_r"} for c in cands]}


def received_by_line(repo, purchase_order_id: str) -> dict:
    """{order line_no: Fraction received in the base unit}: receipts less returns, derived from the receipt lines."""
    out = {}
    for r in repo.receipts.lines_of_order(purchase_order_id):
        out[r["order_line_no"]] = out.get(r["order_line_no"], Fraction(0)) + fr(r["quantity_base"])
    return out


def returned_by_line(repo, purchase_order_id: str) -> dict:
    out = {}
    for r in repo.receipts.lines_of_order(purchase_order_id):
        if r["kind"] == "return":
            out[r["order_line_no"]] = out.get(r["order_line_no"], Fraction(0)) - fr(r["quantity_base"])
    return out


def billed_by_line(repo, purchase_order_id: str) -> dict:
    """{order line_no: Fraction base quantity claimed by approved bills (less credits)}."""
    out = repo.bills.claimed_by_line(purchase_order_id, "bill")
    for k, v in repo.bills.claimed_by_line(purchase_order_id, "credit").items():
        out[k] = out.get(k, Fraction(0)) - v
    return out


def line_status(line, received: Fraction, billed: Fraction) -> dict:
    """Derived per order line: received, billed and outstanding (in the line's unit and the base unit). Never stored."""
    factor = Fraction(line.unit_denominator, line.unit_numerator) if line.unit_numerator else Fraction(0)
    ordered = line.exact_base
    return {"received_base": as_dec(received), "billed_base": as_dec(billed), "outstanding_base": as_dec(max(ordered - received, Fraction(0))),
            "received": as_dec(received * factor), "billed": as_dec(billed * factor), "outstanding": as_dec(max(ordered - received, Fraction(0)) * factor)}