dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/domain/rules.py
"""Partner functions, incompletion rules and block reasons: dated or configured rows the order reads, never code that decides for a customer.
* A customer's partners (SO-40): each ship-to, the bill-to and the payer are other business partners, from a date AFTER today (append-only).
An order copies them when it is raised, from the rows in force on its date.
* Incompletion rules (SO-38): per order type, which field blocks confirmation, delivery or billing, and under what condition.
* Block reasons (SO-39): the delivery and billing blocks an order admin can set; APPROVAL is seeded and released only by approval.
"""
import json
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_sales_order.domain.readiness import Gate, no, ok
FUNCTIONS = ("ship_to", "bill_to", "payer")
STAGES = ("confirm", "delivery", "billing")
CONDITIONS = ("always", "export_ship_to", "customer_policy_requires")
EPOCH = date(1970, 1, 1)
# a field named in an incompletion rule -> where the order keeps it, and where the person fixes it
FIELDS = {"delivery_terms": ("delivery_terms", "Order: delivery terms"), "customer_po_ref": ("customer_po_ref", "Order: Customer PO number"),
"return_reason": ("return_reason", "Order: return reason"), "reference_invoice": ("ref_invoice_id", "Order: the invoice it refers to")}
@dataclass
class CustomerPartner:
sold_to_bp_id: int
function: str
partner_bp_id: int
valid_from: date
is_default: bool = False
transit_days_json: str = "{}" # {warehouse_id: days} from each warehouse, for the promised date
status: str = "active" # active | ended
set_by: str = ""
def transit_days(self, warehouse_id: str) -> int:
return int(json.loads(self.transit_days_json or "{}").get(warehouse_id, 0))
@dataclass
class IncompletionRule:
order_type_code: str
field: str
valid_from: date
blocks: str = "confirm" # confirm | delivery | billing
condition: str = "always" # always | export_ship_to | customer_policy_requires
reason_text: str = ""
@dataclass
class BlockReason:
block_reason_code: str
kind: str = "delivery" # delivery | billing | any (APPROVAL blocks both)
name: str = ""
@dataclass
class ReturnReason:
"""Why goods come back (SO-41): whether they are inspected, the restocking fee kept from the credit, and when it is credited (receipt |
inspection; empty: the owner's return_credit_on)."""
return_reason_code: str
name: str = ""
inspect: bool = True
restocking_fee_pct: "Decimal" = Decimal(0)
credit_on: str = ""
def default_return_reasons() -> list:
return [ReturnReason("DAMAGED", "arrived damaged", True, Decimal(0)), ReturnReason("WRONG_ITEM", "wrong item sent", True, Decimal(0)),
ReturnReason("CHANGED_MIND", "the customer changed their mind", True, Decimal(10))]
def default_incompletion() -> list:
return [IncompletionRule("STANDARD", "delivery_terms", EPOCH, "delivery", "export_ship_to", "an export order needs its delivery terms (Incoterms)"),
IncompletionRule("STANDARD", "customer_po_ref", EPOCH, "billing", "customer_policy_requires", "this customer requires a PO number on the invoice"),
IncompletionRule("RETURNS", "return_reason", EPOCH, "confirm", "always", "a return needs its reason"),
IncompletionRule("CREDIT_REQ", "reference_invoice", EPOCH, "confirm", "always", "a credit request names the invoice it is about")]
def default_block_reasons() -> list:
return [BlockReason("APPROVAL", "any", "waiting for approval")]
def in_force_partners(rows: list, sold_to: int, function: str, on: date) -> list:
"""The partners of `sold_to` in `function` on a date: each partner's latest row on or before it, kept while it is active."""
latest: dict = {}
for r in sorted((r for r in rows if r.sold_to_bp_id == int(sold_to) and r.function == function and r.valid_from <= on), key=lambda r: r.valid_from):
latest[r.partner_bp_id] = r
return [r for r in latest.values() if r.status == "active"]
def resolve_partner(rows: list, sold_to: int, function: str, asked: Optional[int], on: date) -> tuple:
"""(bp_id, refusal): the asked partner when it is one of the sold-to's (or the sold-to itself, or the customer keeps no partner rows for the
function: partners_valid still checks it on the register at confirmation); else the default row in force; else the
sold-to. A partner that is not the customer's is refused in the words of the card."""
mine = in_force_partners(rows, sold_to, function, on)
if asked:
if not mine or int(asked) == int(sold_to) or any(r.partner_bp_id == int(asked) for r in mine):
return int(asked), ""
what = function.replace("_", "-")
return 0, f"{asked} is not a {what} of {sold_to}"
default = [r for r in mine if r.is_default]
return (max(default, key=lambda r: r.valid_from).partner_bp_id if default else int(sold_to)), ""
def transit_days(rows: list, sold_to: int, ship_to: int, warehouse_id: str, on: date) -> int:
row = next((r for r in in_force_partners(rows, sold_to, "ship_to", on) if r.partner_bp_id == int(ship_to)), None)
return row.transit_days(warehouse_id) if row else 0
def incomplete_gates(o, rules: list, stage: str, *, on: date, company_country: str, policy_requires_po: bool) -> list:
"""The incomplete:<field> gates of one stage, for the rules of the order's type in force on a date. A rule that does not apply (not an
export order, the customer asks for no PO) adds no gate; one that applies is met when the field is filled."""
latest: dict = {}
for r in sorted((r for r in rules if r.order_type_code == o.order_type_code and r.blocks == stage and r.valid_from <= on), key=lambda r: r.valid_from):
latest[r.field] = r
out: list = []
for field, r in sorted(latest.items()):
applies = (r.condition == "always"
or (r.condition == "export_ship_to" and bool(o.ship_to_country) and o.ship_to_country.upper() != (company_country or "").upper())
or (r.condition == "customer_policy_requires" and policy_requires_po))
if not applies:
continue
attr, fix = FIELDS.get(field, (field, f"Order: {field}"))
filled = bool(getattr(o, attr, ""))
out.append(ok(f"incomplete:{field}") if filled else no(f"incomplete:{field}", r.reason_text or f"{field} is needed before {stage}", fix))
return out
def block_gate(o, kind: str) -> Gate:
"""delivery_block / billing_block: met when none is set; unmet names the reason code."""
b = o.active_block(kind)
name = "delivery_block" if kind == "delivery" else "billing_block"
if b is None:
return ok(name)
return no(name, f"on hold: {b.reason_code}" + (f" ({b.note})" if b.note else ""),
"an order admin releases the block" if b.reason_code != "APPROVAL" else "the approval releases it")