dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/domain/gates.py
"""The readiness of each transition, as {gate, met, reason, fix} — one rule, so a refusal and the screen's answer
can never disagree (module.yaml invariant: a transition refused for a gate names the gate, its reason and its fix)."""
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Optional
from brick_sales_order.domain.readiness import Gate, no, ok
from brick_sales_order.domain.values import MEMO_TYPES, REPEATING, RETURN_TYPES, SalesOrderState, num
ZERO = Decimal(0)
@dataclass
class Facts:
"""What the order cannot know itself: answers from neighbours and from rows, gathered by the use case."""
customer_active: Optional[bool] = True # None: the partner register did not answer
customer_note: str = ""
credit: Optional[dict] = None # None: payments is not part of this installation
po_required: bool = False
invoicing_installed: bool = True
issued_deposits: list = field(default_factory=list) # source_ids of deposit requests whose invoice is ISSUED
issued: dict = field(default_factory=dict) # line_no -> quantity on ISSUED invoices, read back from invoicing
multiples: Optional[dict] = None # line_no -> product's rounding answer (or "unanswered"); None: nothing was asked
delivery_mode: str = "warn" # warn | block: delivery_multiple_mode in force on the date
extra: list = field(default_factory=list) # gates the use case gathered: partners_valid, incomplete:* for confirmation
def _lines_ok(o) -> Gate:
if not o.lines:
return no("lines_priced", "an order needs at least one line", "Order > add a line")
bare = [str(ln.line_no) for ln in o.lines if not ln.priced_from]
if bare:
return no("lines_priced", f"line {', '.join(bare)} has no price source (priced_from)", "Order > re-add the line: prices come from the product module")
return ok("lines_priced")
def delivery_multiple_gate(o, f: Facts) -> Gate:
"""SO-28: each asked line off its product's delivery multiple. Warn mode is MET with the reason and the fix (shown and recorded on the
order); block mode is not met with the same words. A product that did not answer is unmet either way: a rule nobody read is not a pass."""
reasons, fixes, silent = [], [], []
for ln in o.lines:
a = (f.multiples or {}).get(ln.line_no)
if a is None:
continue
if a == "unanswered":
silent.append(str(ln.line_no))
elif not a.get("met", True):
reasons.append(a.get("reason") or f"line {ln.line_no} is not a multiple of its product's delivery rule")
near = [num(a[k]) for k in ("next_down", "next_up") if a.get(k) not in (None, "")]
where = "" if len(o.lines) == 1 else f"line {ln.line_no} "
fixes.append(f"change {where}to " + (" or ".join(near) if near else "a multiple"))
if silent:
return no("delivery_multiple", f"the product module did not answer for line {', '.join(silent)}", "try again")
if not reasons:
return ok("delivery_multiple")
reason, fix = "; ".join(reasons), "; ".join(fixes)
if f.delivery_mode == "block":
return no("delivery_multiple", reason, fix)
return ok("delivery_multiple", reason, fix + ", or confirm as it is")
def confirm_gates(o, f: Facts) -> list:
gates = []
if f.customer_active is True:
gates.append(ok("customer_active"))
else:
gates.append(no("customer_active", f.customer_note or ("the customer is not active on the partner register"
if f.customer_active is False else "the partner register did not confirm the customer"),
"Business partner > the customer, or try again"))
gates.append(_lines_ok(o))
bad = [str(ln.line_no) for ln in o.lines if not (ln.unit_code and ln.unit_numerator > 0 and ln.unit_denominator > 0)]
gates.append(no("units_valid", f"line {', '.join(bad)} has no unit or conversion", "Order > the line's unit") if bad else ok("units_valid"))
if f.multiples is not None:
gates.append(delivery_multiple_gate(o, f))
gates.append(no("freight_computable", f"FREIGHT cannot be computed: line {o.freight_unknown} has no known weight", "record the unit's weight in Products")
if o.freight_unknown else ok("freight_computable"))
untaxed = [str(ln.line_no) for ln in o.lines if not ln.tax_code]
gates.append(no("lines_taxed", f"line {', '.join(untaxed)} has no tax code", "Order > recalculate tax (the tax module computes it)") if untaxed else ok("lines_taxed"))
c = f.credit
if c is None or not c.get("hold"):
gates.append(ok("credit", "payments is not part of this installation: no credit check" if c is None else ""))
elif (c.get("mode") or "block") == "warn":
gates.append(ok("credit", f"warning: {c.get('reason', 'on credit hold')}"))
elif o.credit_override_by:
gates.append(ok("credit", f"overridden by {o.credit_override_by}: {o.credit_override_reason}"))
else:
gates.append(no("credit", f"on hold: {c.get('reason', 'credit hold')}", "ask an order admin to override"))
gates.append(no("po_ref", "this customer requires a PO number", "Order > Customer PO number") if f.po_required and not o.customer_po_ref else ok("po_ref"))
open_ = [s for s in o.suggestions if s.decision == "open"]
gates.append(no("no_open_suggestion", f"{len(open_)} AI suggestion(s) are waiting for you", "Order > accept, change or reject each") if open_ else ok("no_open_suggestion"))
gates.extend(f.extra)
if any(ln.billing_kind in REPEATING for ln in o.lines):
gates.append(ok("billing_handoff") if f.invoicing_installed else
no("billing_handoff", "invoicing is not part of this installation, so a repeating line cannot be scheduled", "install invoicing (ar)"))
stages = o.plan_complete()
if any(ln.milestone for ln in o.lines):
gates.append(no("plan_complete", stages, "set the milestones so they total 100%") if stages else ok("plan_complete"))
if o.order_type_code == "CASH" and any(ln.repeating or ln.milestone for ln in o.lines):
gates.append(no("cash_no_plan", "a CASH order is paid on the spot: it takes no billing plan (schedule, instalments or milestones)", "raise a STANDARD order for a plan"))
return gates
def _same_person(a: tuple, b: tuple) -> bool:
"""Two people are one if ANY identity (email or sub) of one is any of the other's."""
mine = {x.strip().lower() for x in a if x and x.strip()}
return bool(mine & {x.strip().lower() for x in b if x and x.strip()})
def amend_gates(o, f: Facts, changes=(), proposer: str = "", approver: str = "", needs_approval: bool = True,
proposer_ids: tuple = (), approver_ids: tuple = ()) -> list:
gates = [ok("not_done") if o.lifecycle == SalesOrderState.CONFIRMED.value else
no("not_done", f"the order is {o.lifecycle}" + (": raise a new order" if o.lifecycle == "done" else ""), "raise a new order")]
gates.append(no("amendable_type", f"a {o.order_type_code} is never amended: its quantity is held against the original delivery or invoice line, and its price is the one it references",
"cancel it and raise a new one") if o.order_type_code in RETURN_TYPES or o.order_type_code in MEMO_TYPES else ok("amendable_type"))
pend = [r for r in o.revisions if r.status == "pending"]
gates.append(no("no_pending_amendment", f"revision {pend[0].revision} is waiting for approval", "approve or reject it first") if pend and not approver else ok("no_pending_amendment"))
low = []
for ch in changes:
ln = o.line(ch["line_no"])
if ln is None:
low.append(f"no line {ch['line_no']}")
continue
done = max(o.delivered(ln.line_no), o.invoiced_qty(ln.line_no, f.issued))
if ch.get("quantity") is not None and ch["quantity"] < done:
low.append(f"line {ln.line_no}: {num(ch['quantity'])} is below the {num(done)} already delivered or invoiced")
if ch.get("unit_code") and ch["unit_code"] != ln.unit_code and done > 0:
low.append(f"line {ln.line_no}: the unit cannot change once something is delivered or invoiced")
gates.append(no("not_below_done", "; ".join(low), "change the line to at least what is delivered or invoiced") if low else ok("not_below_done"))
if f.multiples:
gates.append(delivery_multiple_gate(o, f))
if approver and needs_approval:
gates.append(no("approver_not_amender", "an amendment is never approved by the person who made it", "ask another holder of so:order:approve")
if _same_person((proposer, *proposer_ids), (approver, *approver_ids)) else ok("approver_not_amender"))
return gates
def cancel_gates(o, f: Facts) -> list:
gates = [ok("not_done") if o.lifecycle == SalesOrderState.CONFIRMED.value else
no("not_done", f"the order is {o.lifecycle}", "")]
gates.append(no("nothing_delivered", "something is already delivered: record a correction, or close the order", "Deliveries") if any(o.delivered(ln.line_no) > 0 for ln in o.lines) else ok("nothing_delivered"))
gates.append(no("nothing_invoiced", "an invoice is already issued: credit the invoice, then close the order", "Invoicing > credit note")
if any(v > 0 for v in f.issued.values()) else ok("nothing_invoiced"))
return gates
def deposit_gates(o, amount: Optional[Decimal] = None) -> list:
from brick_sales_order.domain.values import money
want = amount if amount is not None else money(o.subtotal * o.deposit_pct / 100)
gates = [ok("confirmed") if o.lifecycle == "confirmed" else no("confirmed", f"the order is {o.lifecycle}", "confirm it first")]
if o.order_type_code == "CASH":
gates.append(no("cash_no_deposit", "a CASH order is paid on the spot: it takes no deposit", "raise a STANDARD order for a deposit"))
gates.append(ok("deposit_set") if want > 0 else no("deposit_set", "the order's deposit percentage is zero", "Order > deposit, or the customer's policy"))
already = o.deposit_total()
gates.append(no("within_total", f"deposits would reach {num(already + want, 2)}, more than the order total {num(o.subtotal, 2)}", "a smaller deposit")
if already + want > o.subtotal else ok("within_total"))
return gates
def invoice_gates(o, f: Facts, skip=frozenset()) -> list:
gates = [ok("confirmed") if o.lifecycle == "confirmed" else no("confirmed", f"the order is {o.lifecycle}", "")]
due = any(o.due_qty(ln.line_no, f.issued) > 0 for ln in o.lines if ln.line_no not in skip) \
or any(m.released_on and not m.billed_request_no for m in o.milestones)
gates.append(ok("something_due") if due else no("something_due", "no line has quantity left to invoice (ordered or delivered, per its policy)", "record a delivery, or nothing more is due"))
return gates