dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/evaluation.py
"""The supplier scorecard (PUR-46): price, delivery, quantity, quality (derived from rows) and service (the buyer's 1-5 score, scaled), each 1 to
100, and the weighted overall. Freezing writes (bp_id, period_id, seq); a frozen evaluation never changes - a re-evaluation is a new row with its reason.
The weights are owner-controlled dated rows."""
import json
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, text
from brick_purchasing.application import base
from brick_purchasing.application.base import CONTROLS, EVAL_WRITE, PURCHASING, _decide, _today, _who, unit
from brick_purchasing.application.inputs import Reason
from brick_purchasing.domain.errors import RuleBroken
repository = lambda: base.repository() # noqa: E731
CRITERIA = ("price", "delivery", "quantity", "quality", "service")
DEFAULT_WEIGHTS = {"price": 25, "delivery": 25, "quantity": 20, "quality": 20, "service": 10}
def period_bounds(period_id: str) -> tuple:
"""'2026-Q3' (a quarter), '2026-09' (a month) or '2026' (a year) as (first day, last day)."""
try:
if "-Q" in period_id:
y, q = period_id.split("-Q")
m = (int(q) - 1) * 3 + 1
if not 1 <= int(q) <= 4:
raise ValueError
start = date(int(y), m, 1)
end = date(int(y) + (m + 3 > 12), (m + 3 - 1) % 12 + 1, 1) - timedelta(days=1)
elif len(period_id) == 7:
y, m = int(period_id[:4]), int(period_id[5:])
start = date(y, m, 1)
end = date(y + (m == 12), m % 12 + 1, 1) - timedelta(days=1)
else:
if not (len(period_id) == 4 and period_id.isdigit()): # a year is read as text: period_id is a VARCHAR key
raise ValueError
y = int(period_id[:4])
start, end = date(y, 1, 1), date(y, 12, 31)
except ValueError:
raise Refused(422, "period_id is a year (2026), a quarter (2026-Q3) or a month (2026-09)") from None
return start, end
def weights_on(repo, on: date) -> dict:
out = {}
for c in CRITERIA:
rows = [r for r in repo.weights.all(criterion=c) if r["valid_from"] <= on]
out[c] = int(max(rows, key=lambda r: r["valid_from"])["weight"]) if rows else DEFAULT_WEIGHTS[c]
return out
def _clip(x: float) -> int:
return int(max(1, min(100, round(x))))
def derive(repo, bp_id: int, period_id: str) -> dict:
start, end = period_bounds(period_id)
grace = int(repo.settings.value_on("evaluation_grace_days", end))
lines = on_time = returned = ordered_q = recvd_q = 0
pct_sum, pct_n = Decimal(0), 0
for h in repo.orders.repo.all_heads():
if h.bp_id != int(bp_id):
continue
o = repo.orders.get(h.purchase_order_id)[0]
for r in repo.receipts.of_order(h.purchase_order_id):
if not (start <= r.received_on <= end):
continue
if r.kind == "return":
if any(w in (r.reason or "").lower() for w in ("quality", "damag", "defect")):
returned += len(r.lines)
continue
for rl in r.lines:
x = o.line(rl.order_line_no)
lines += 1
on_time += 1 if (x is None or x.expected_on is None or r.received_on <= x.expected_on + timedelta(days=grace)) else 0
if x is not None:
ordered_q += float(x.quantity_base)
recvd_q += min(float(rl.quantity_base), float(x.quantity_base))
for b in repo.bills.of_order(h.purchase_order_id):
if b.status != "approved" or b.kind != "bill" or not (start <= b.invoice_date <= end):
continue
for ln in b.lines:
x = o.line(ln.order_line_no) if ln.order_line_no else None
if x is not None and x.unit_price and ln.pricing_unit_code == x.pricing_unit_code:
pct_sum += (Decimal(ln.unit_price) - Decimal(x.unit_price)) / Decimal(x.unit_price) * 100
pct_n += 1
avg_pct = (pct_sum / pct_n) if pct_n else Decimal(0)
svc = repo.service_scores.get(int(bp_id), period_id)
scores = {"delivery": _clip(100 * on_time / lines) if lines else 100, "quality": _clip(100 * (1 - returned / lines)) if lines else 100,
"price": _clip(100 - 10 * float(max(avg_pct, Decimal(0)))), "quantity": _clip(100 * recvd_q / ordered_q) if ordered_q else 100,
"service": _clip(int(svc["score"]) * 20) if svc else None}
w = weights_on(repo, end)
use = {c: w[c] for c in CRITERIA if scores[c] is not None}
total = sum(use.values()) or 1
overall = (sum(use[c] * scores[c] for c in use) / total)
return {"bp_id": str(bp_id), "period_id": period_id, "period_start": start, "period_end": end, "scores": scores, "weights": w,
"overall": Decimal(str(round(overall, 2))), "evidence": {"receipt_lines": lines, "on_time": on_time, "returned_for_quality": returned,
"avg_price_over_order_pct": str(round(avg_pct, 2)), "service_scored": svc is not None}}
@PURCHASING.use_case(kind="query", route="GET /suppliers/{bp_id}/evaluation", permission=EVAL_WRITE)
def preview_evaluation(caller: dict, bp_id: BigId, period_id: text(10, min_length=4)) -> dict:
"""The scorecard of a supplier for a period, derived from the rows now, with the evidence behind each score. Nothing is stored."""
with repository() as repo:
return derive(repo, int(bp_id), period_id)
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/service-score", permission=EVAL_WRITE, idempotent=True)
def set_service_score(caller: dict, bp_id: BigId, period_id: text(10, min_length=4), score: int) -> dict:
"""The buyer's service score for a period, 1 to 5 (scaled to 100 on the scorecard)."""
if not 1 <= int(score) <= 5:
raise Refused(422, "the service score is 1 to 5")
period_bounds(period_id)
def run():
with repository() as repo, _decide():
if repo.evaluations.get(int(bp_id), period_id, 1) is not None:
raise RuleBroken("frozen: the period is already evaluated; fix: set the score before freezing, or re-evaluate with a reason")
repo.service_scores.put({"bp_id": int(bp_id), "period_id": period_id, "score": int(score), "set_by_email": _who(caller)})
return {"bp_id": str(bp_id), "period_id": period_id, "score": int(score)}
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/evaluations", permission=EVAL_WRITE)
def freeze_evaluation(caller: dict, bp_id: BigId, period_id: text(10, min_length=4), reason: Optional[Reason] = None) -> dict:
"""Freeze the supplier's evaluation for a period: written once as (bp_id, period_id, seq) and never changed. A second evaluation of the same
period is a new row and needs a reason."""
def run():
with repository() as repo, _decide():
d = derive(repo, int(bp_id), period_id)
seq = 1
while repo.evaluations.get(int(bp_id), period_id, seq) is not None:
seq += 1
if seq > 1 and not (reason or "").strip():
raise RuleBroken("reason: this period is already evaluated; a re-evaluation is a new row and needs a reason")
s = d["scores"]
repo.evaluations.put({"bp_id": int(bp_id), "period_id": period_id, "seq": seq, "price_score": s["price"], "delivery_score": s["delivery"],
"quantity_score": s["quantity"], "quality_score": s["quality"], "service_score": s["service"], "overall": d["overall"],
"weights": json.dumps(d["weights"]), "frozen_by_email": _who(caller), "frozen_on": _today(repo), "reason": (reason or "").strip()})
return {**d, "seq": seq, "frozen": True, "reason": (reason or "").strip()}
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /suppliers/{bp_id}/evaluations", permission=EVAL_WRITE)
def list_evaluations(caller: dict, bp_id: BigId) -> dict:
"""The frozen evaluations of a supplier over time, oldest first, for comparison."""
with repository() as repo:
rows = sorted(repo.evaluations.all(bp_id=int(bp_id)), key=lambda r: (r["period_id"], r["seq"]))
return {"bp_id": str(bp_id), "items": [{**r, "bp_id": str(r["bp_id"]), "weights": json.loads(r["weights"] or "{}")} for r in rows]}
@PURCHASING.use_case(kind="command", route="POST /evaluation-weights", permission=CONTROLS)
def set_evaluation_weights(caller: dict, valid_from: IsoDate, price: int, delivery: int, quantity: int, quality: int, service: int) -> dict:
"""The criteria weights from a date AFTER today (owner-controlled): each 0 to 100 and together above zero. A frozen evaluation keeps the weights it used."""
w = {"price": price, "delivery": delivery, "quantity": quantity, "quality": quality, "service": service}
if any(not 0 <= int(v) <= 100 for v in w.values()) or sum(w.values()) <= 0:
raise Refused(422, "each weight is 0 to 100 and they are not all zero")
def run():
with repository() as repo, _decide():
start = base._day(valid_from, "valid_from")
if start <= _today(repo):
raise Refused(409, f"starts_after_today: weights start tomorrow at the earliest ({_today(repo)} is today)")
for c, v in w.items():
repo.weights.put({"criterion": c, "valid_from": start, "weight": int(v), "set_by_email": _who(caller)})
return {"valid_from": start, "weights": w}
return unit(None, run)