dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/domain/sales_order.py
"""The SalesOrder aggregate — the rules, and nothing else.
What a customer has committed to buy, and on what terms. No database, no HTTP, no identity: a rule here is true
whoever calls it and however it is stored. "Today" is always handed in (the use case reads it from
business_timezone); prices, tax and invoiced quantities come from neighbours and are handed in too.
The order is a COMMITMENT separate from the quote. Fulfilment and invoice status are DERIVED from quantities on
read and never stored. The order asks invoicing for an invoice; it never writes one.
"""
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from brick_sales_order.domain import billing as bill
from brick_sales_order.domain import events as ev
from brick_sales_order.domain import gates as gate_rules
from brick_sales_order.domain import pricing as pr
from brick_sales_order.domain.errors import GateAnswer, GateUnmet, RuleBroken
from brick_sales_order.domain.readiness import no, require
from brick_sales_order.domain.rules import block_gate
from brick_sales_order.domain.values import MEMO_TYPES, REPEATING, RETURN_TYPES, ZERO, money, num
D = Decimal
@dataclass
class LineSpec:
"""A priced line as the use case hands it over: from find_price, from an accepted quote, or from Odoo."""
product_id: str
description: str
unit_code: str
quantity: Decimal
unit_numerator: int
unit_denominator: int
list_price: Decimal
unit_price: Decimal
discount_pct: Decimal
priced_on: Optional[date]
priced_from: str
tax_category: str = ""
invoicing_policy: str = "ordered"
billing_kind: str = ""
billing_plan: str = ""
quote_ref: str = ""
odoo_line_id: str = ""
tax_code: str = ""
net_amount: Optional[Decimal] = None # a migrated line carries the amounts Odoo computed
tax_amount: Optional[Decimal] = None
pricing_quantity: int = 1 # unit_price is per this many of pricing_unit_code (a price per 100)
pricing_unit_code: str = "" # "" = the line's own unit
pricing_numerator: int = 1 # the pricing unit's conversion to the base unit
pricing_denominator: int = 1
category_code: str = "" # the product's category (conditions are found by it)
product_type_code: str = "" # TRADING | SERVICE ...: what the item category is determined from
usage: str = "normal" # normal | drop_ship
item_category_code: str = ""
gross_weight: Optional[Decimal] = None # per ONE of the line's unit, copied from product's measures for freight
weight_unit_code: str = ""
measure_valid_from: Optional[date] = None
procedural: bool = False # priced by the pricing procedure (a quote's or Odoo's prices are copied, never re-priced)
ref_sales_order_id: str = "" # what a return, a memo request or a consignment movement refers to (v5)
ref_line_no: int = 0
ref_delivery_ref: str = ""
ref_invoice_id: str = ""
ref_invoice_line_no: int = 0
instalments: int = 0 # equal instalments of the line's total (a schedule at invoicing)
def net(self) -> Decimal:
"""The line's net, rounded once: the quantity in the PRICING unit x unit_price / pricing_quantity (a price per 100 is
never divided down to a piece price). A migrated line carries Odoo's own net."""
if self.net_amount is not None:
return self.net_amount
return money(priced_quantity(self.quantity, self.unit_code, self.unit_numerator, self.unit_denominator, self.pricing_unit_code,
self.pricing_numerator, self.pricing_denominator) * self.unit_price / self.pricing_quantity)
def priced_quantity(quantity: Decimal, unit_code: str, un: int, ud: int, pricing_unit_code: str, pn: int, pd: int) -> Decimal:
"""The quantity expressed in the pricing unit: exact (a fraction through the base unit), unchanged where the units agree."""
if not pricing_unit_code or pricing_unit_code == unit_code:
return quantity
return quantity * un * pd / (ud * pn)
@dataclass
class OrderLine:
"""One product, quantity and price on the order. Description, unit and its conversion, tax category,
invoicing policy and billing are COPIES taken when the line was added."""
sales_order_id: str
line_no: int
product_id: str = ""
description: str = ""
unit_code: str = ""
quantity: Decimal = ZERO
unit_numerator: int = 1
unit_denominator: int = 1
list_price: Decimal = ZERO
unit_price: Decimal = ZERO
discount_pct: Decimal = ZERO
priced_on: Optional[date] = None
priced_from: str = ""
tax_category: str = ""
tax_code: str = ""
net_amount: Decimal = ZERO
tax_amount: Decimal = ZERO
invoicing_policy: str = "ordered"
billing_kind: str = ""
billing_plan: str = ""
quote_ref: str = ""
odoo_line_id: str = ""
pricing_quantity: int = 1
pricing_unit_code: str = ""
pricing_numerator: int = 1
pricing_denominator: int = 1
stocked: bool = False # v3: inventory's reservation answer took it (a service line is never stocked)
reserved_base: Decimal = ZERO # what inventory holds for the line, in the base unit, and what it could not
short_base: Decimal = ZERO
category_code: str = ""
product_type_code: str = ""
usage: str = "normal"
item_category_code: str = ""
gross_weight: Optional[Decimal] = None
weight_unit_code: str = ""
measure_valid_from: Optional[date] = None
procedural: bool = False
ref_sales_order_id: str = ""
ref_line_no: int = 0
ref_delivery_ref: str = ""
ref_invoice_id: str = ""
ref_invoice_line_no: int = 0
instalments: int = 0
statistical_value: Decimal = ZERO # a free-of-charge line: what it would have sold for, shown and never billed (net_amount is 0.00)
purchase_request_id: str = "" # a third-party line: purchasing's request, cancelled with the order
@property
def milestone(self) -> bool:
return self.billing_plan == "milestone"
@property
def quantity_priced(self) -> Decimal:
"""The quantity in the PRICING unit (1 500 C62 for 3 boxes of 500), exact."""
return priced_quantity(self.quantity, self.unit_code, self.unit_numerator, self.unit_denominator, self.pricing_unit_code,
self.pricing_numerator, self.pricing_denominator)
@classmethod
def from_spec(cls, sales_order_id: str, line_no: int, s: LineSpec) -> "OrderLine":
net = s.net()
ln = cls(sales_order_id, line_no, s.product_id, s.description, s.unit_code, s.quantity, s.unit_numerator,
s.unit_denominator, s.list_price, s.unit_price, s.discount_pct, s.priced_on, s.priced_from,
s.tax_category, s.tax_code, net, s.tax_amount if s.tax_amount is not None else ZERO,
s.invoicing_policy, s.billing_kind, s.billing_plan, s.quote_ref, s.odoo_line_id, s.pricing_quantity or 1,
s.pricing_unit_code, s.pricing_numerator or 1, s.pricing_denominator or 1)
ln.category_code, ln.product_type_code, ln.usage, ln.item_category_code = s.category_code, s.product_type_code, s.usage, s.item_category_code
ln.gross_weight, ln.weight_unit_code, ln.measure_valid_from, ln.procedural = s.gross_weight, s.weight_unit_code, s.measure_valid_from, s.procedural
ln.ref_sales_order_id, ln.ref_line_no, ln.ref_delivery_ref = s.ref_sales_order_id, s.ref_line_no, s.ref_delivery_ref
ln.ref_invoice_id, ln.ref_invoice_line_no, ln.instalments = s.ref_invoice_id, s.ref_invoice_line_no, s.instalments
return ln
@property
def repeating(self) -> bool:
return self.billing_kind in REPEATING
def billed_price(self) -> tuple:
"""(unit_price, discount_pct) as invoicing bills it: invoicing computes a line's net as quantity x unit_price x
(100 - discount_pct) / 100, so a discounted line sends its LIST price with the discount — sending the net price with
the discount would take the discount twice. Only where list x (100 - discount) / 100 IS the agreed unit price exactly;
otherwise (no discount, or a rounded quote discount such as 16.6667%) the agreed unit price is sent with no discount,
so invoicing's net is always the order's net."""
if self.discount_pct and self.list_price and self.list_price * (100 - self.discount_pct) / 100 == self.unit_price:
return self.list_price, self.discount_pct
return self.unit_price, ZERO
@dataclass
class LineCondition:
"""One step of a line's price (SO-30): BASE from product, then each condition the procedure found, each naming the record it came from.
Appended per revision, frozen at confirmation; the row in force is the latest revision of a step. A header condition's SHARE of the line
is a row too (header_share_of names it), so invoicing bills and taxes it with the goods."""
sales_order_id: str
line_no: int
step: int
revision: int
condition_type_code: str = ""
rate: Decimal = ZERO
amount: Decimal = ZERO
basis: Decimal = ZERO
source: str = ""
statistical: bool = False
header_share_of: str = ""
@dataclass
class HeaderCondition:
"""An order-wide condition (a header discount, freight): its amount, before it is spread over the lines."""
sales_order_id: str
condition_type_code: str
revision: int
step: int = 0
rate: Decimal = ZERO
amount: Decimal = ZERO
source: str = ""
@dataclass
class Delivery:
"""Some quantity of a line delivered or done. Append-only: a mistake is a correction row against the original."""
sales_order_id: str
line_no: int
delivery_ref: str
quantity: Decimal = ZERO
kind: str = "delivery"
corrects_ref: str = ""
delivered_on: Optional[date] = None
recorded_by: str = ""
source: str = "manual"
quantity_base: Decimal = ZERO # v3: exact, in the product's base unit (negative on a correction)
@dataclass
class InvoiceRequest:
"""What the order asked invoicing to bill. The request log, not the truth of what was billed."""
sales_order_id: str
request_no: int
kind: str = "regular"
lines_json: str = "[]"
amount: Decimal = ZERO # net of this request (the deposit's amount, or the lines' net less the deduction)
deducted: Decimal = ZERO # deposit used up by this request
invoice_id: str = ""
status: str = "requested" # pending (claimed here, invoicing not yet told) | requested | cancelled
requested_at: Optional[datetime] = None
requested_by: str = ""
deductions_json: str = "[]" # the deposits this request deducts: [{deposit_invoice_id, net_amount, tax_category}]
source_ref: str = "" # what the request is about (a credit's original invoice)
@property
def deductions(self) -> list:
return json.loads(self.deductions_json or "[]")
@property
def live(self) -> bool:
return self.status in ("requested", "pending")
@property
def lines(self) -> list:
return json.loads(self.lines_json or "[]")
@dataclass
class BillingMilestone:
"""One stage of a milestone line (SO-46). Released by its event; its amount is frozen at release, and a stage not yet released is re-spread
from what is left whenever the line's net changes. billed_request_no: the invoice request that billed it (0 = not yet)."""
sales_order_id: str
line_no: int
milestone_no: int
pct: Decimal = ZERO
event: str = "confirmation"
name: str = ""
due_on: Optional[date] = None
released_on: Optional[date] = None
released_by: str = ""
amount: Decimal = ZERO
billed_request_no: int = 0
@dataclass
class ReturnInspection:
"""What inventory found when it inspected a returned line (SO-49): appended once per (order, line, inspection_ref)."""
sales_order_id: str
line_no: int
inspection_ref: str
received_quantity: Decimal = ZERO
accepted_quantity: Decimal = ZERO
rejected_quantity: Decimal = ZERO
decision: str = "restock"
reason: str = ""
inspected_on: Optional[date] = None
recorded_at: Optional[datetime] = None
def _sub_of(by: str, identities: tuple) -> str:
"""The identity of the actor that is not `by` (by is email, else sub): what is stored beside it."""
return next((x for x in identities if x and x != by), "")
@dataclass
class Revision:
"""A change to a confirmed order, with why and who approved it."""
sales_order_id: str
revision: int
reason: str = ""
changed_lines_json: str = "[]"
status: str = "pending" # pending | applied | rejected
proposed_by: str = ""
proposed_at: Optional[datetime] = None
approved_by: str = ""
approved_at: Optional[datetime] = None
proposed_sub: str = "" # the proposer's sub, beside proposed_by (email else sub): either identity is the same person
@property
def changes(self) -> list:
return [{**c, "quantity": D(c["quantity"]) if c.get("quantity") is not None else None}
for c in json.loads(self.changed_lines_json or "[]")]
@dataclass
class Send:
"""A record that the confirmation went out."""
sales_order_id: str
sent_at: datetime
channel: str = "email"
address: str = ""
revision: int = 1
sent_by: str = ""
schedule_version: int = 0 # the promised dates the customer was sent (SO-36: a date told to the customer is never pushed earlier on its own)
@dataclass
class Suggestion:
"""What an AI step suggested on an order, and what a person did with it (the audit of an AI step)."""
sales_order_id: str
step: str
suggestion_no: int
model: str = ""
inputs_ref: str = ""
output_json: str = "{}"
decisions_json: str = "{}" # item -> {action, by, at}
decision: str = "open" # accepted | changed | rejected | open
decided_by: str = ""
decided_at: Optional[datetime] = None
@property
def output(self) -> dict:
return json.loads(self.output_json or "{}")
@property
def decisions(self) -> dict:
return json.loads(self.decisions_json or "{}")
def items(self) -> list:
out = self.output
items = [f"line:{i}" for i in range(len(out.get("lines", [])))]
if out.get("po_number"):
items.append("po_number")
items += [f"diff:{i}" for i in range(len(out.get("differences", [])))]
return items
@dataclass
class Resolution:
"""What a person's decisions on a suggestion come to — worked out before anything is priced or written."""
decisions: dict
add_lines: list # [{product_id, unit_code, quantity}] accepted or changed
po_number: Optional[str]
decision: str # accepted | changed | rejected | open
@dataclass
class OrderBlock:
"""A block, as an appended row: who set it and why, and (a later row) who released it. Nothing is edited or deleted."""
sales_order_id: str
block_no: int
kind: str = "delivery" # delivery | billing
reason_code: str = ""
event: str = "set" # set | release
note: str = ""
by_email: str = ""
by_sub: str = ""
at: Optional[datetime] = None
@dataclass
class ScheduleLine:
"""One promised delivery of a line: a quantity for a date, appended under a schedule version (a re-promise is a NEW version, never an
edit; the latest version of a line is in force). confirmed_on None: nothing confirms this quantity yet."""
sales_order_id: str
line_no: int
schedule_version: int
seq: int
requested_on: Optional[date] = None
requested_quantity: Decimal = ZERO
confirmed_on: Optional[date] = None
material_available_on: Optional[date] = None
confirmed_quantity: Decimal = ZERO
confirmed_base: Decimal = ZERO
reason: str = ""
at: Optional[datetime] = None
trigger_key: str = ""
@dataclass
class SalesOrder:
sales_order_id: str
order_no: str = ""
bp_id: int = 0
owner: str = ""
currency: str = ""
price_list_id: str = ""
lifecycle: str = "draft"
revision: int = 0
customer_po_ref: str = ""
ordered_on: Optional[date] = None
confirmed_at: Optional[datetime] = None
confirmed_on: Optional[date] = None
delivery_terms: str = ""
payment_terms_id: str = ""
deposit_pct: Decimal = ZERO
subtotal: Decimal = ZERO
tax_total: Decimal = ZERO
total: Decimal = ZERO
source_kind: str = "hand"
source_id: str = ""
credit_override_by: str = ""
credit_override_reason: str = ""
cancel_reason: str = ""
order_type_code: str = "STANDARD"
ship_to_bp_id: int = 0
ship_to_country: str = ""
price_group: str = ""
ref_sales_order_id: str = ""
ref_line_no: int = 0
ref_invoice_id: str = ""
warnings_json: str = "{}" # {line_no: {gate, met, reason, fix}}: the delivery-multiple warnings shown on the order
freight_unknown: str = "" # the lines whose weight is unknown while FREIGHT applies (gate freight_computable)
ship_to_address_lines: str = "" # v8: the ship-to as the address object (shared.yaml conventions.address): lines as a JSON array,
ship_to_address_city: str = "" # then the parts; the country is ship_to_country
ship_to_address_postcode: str = ""
ship_to_address_region: str = ""
bill_to_bp_id: int = 0 # SO-40: the partners, copied when the order was raised (0 = the sold-to)
payer_bp_id: int = 0
requested_delivery_on: Optional[date] = None
priority: str = "normal" # normal | rush
partial_delivery: str = "allowed" # allowed | complete_line | complete_order (copied from the customer's policy at confirmation)
max_partials: int = 0 # part deliveries per line; 0 = no limit
under_tolerance_pct: Decimal = ZERO
over_tolerance_pct: Decimal = ZERO
combine_allowed: bool = False
min_shelf_life_days: int = 0
return_reason: str = ""
deposit_deduction_mode: str = "proportional"
owner_sub: str = "" # the requester's sign-in subject beside owner (email else sub): an approver differs from BOTH
approved_by_email: str = ""
approved_by_sub: str = ""
approved_at: Optional[datetime] = None
lines: list = field(default_factory=list)
deliveries: list = field(default_factory=list)
invoice_requests: list = field(default_factory=list)
revisions: list = field(default_factory=list)
sends: list = field(default_factory=list)
suggestions: list = field(default_factory=list)
conditions: list = field(default_factory=list)
header_conditions: list = field(default_factory=list)
blocks: list = field(default_factory=list)
schedule: list = field(default_factory=list) # ScheduleLine rows, appended
milestones: list = field(default_factory=list) # BillingMilestone rows
inspections: list = field(default_factory=list) # ReturnInspection rows, appended
events: list = field(default_factory=list, repr=False)
# ── making one ────────────────────────────────────────────────────────────────────────────
@classmethod
def draft(cls, sales_order_id: str, *, owner: str, bp_id: int, currency: str, ordered_on: Optional[date], today: date,
source_kind: str = "hand", source_id: str = "", price_list_id: str = "", customer_po_ref: str = "",
delivery_terms: str = "", payment_terms_id: str = "", deposit_pct: Decimal = ZERO,
history: bool = False, order_type_code: str = "STANDARD", ship_to_bp_id: int = 0, ship_to_country: str = "",
price_group: str = "", ref_sales_order_id: str = "", ref_line_no: int = 0, ref_invoice_id: str = "",
bill_to_bp_id: int = 0, payer_bp_id: int = 0, requested_delivery_on: Optional[date] = None, return_reason: str = "") -> "SalesOrder":
"""A draft dated today in business_timezone. Nothing dated is back-dated; the migration is the one writer
allowed a past date (`history`)."""
on = ordered_on or today
if on != today and not history:
raise RuleBroken(f"an order is dated today ({today}): it cannot be dated {on}")
o = cls(sales_order_id, bp_id=int(bp_id), owner=owner, currency=currency, ordered_on=on, source_kind=source_kind,
source_id=source_id or sales_order_id, price_list_id=price_list_id, customer_po_ref=customer_po_ref,
delivery_terms=delivery_terms, payment_terms_id=payment_terms_id, deposit_pct=deposit_pct,
order_type_code=order_type_code, ship_to_bp_id=int(ship_to_bp_id or bp_id), ship_to_country=ship_to_country,
price_group=price_group, ref_sales_order_id=ref_sales_order_id, ref_line_no=int(ref_line_no or 0), ref_invoice_id=ref_invoice_id,
bill_to_bp_id=int(bill_to_bp_id or bp_id), payer_bp_id=int(payer_bp_id or bp_id), requested_delivery_on=requested_delivery_on,
return_reason=return_reason)
o.events.append(ev.OrderDrafted(sales_order_id, source_kind, o.source_id))
return o
@classmethod
def migrated(cls, sales_order_id: str, *, order_no: str, owner: str, bp_id: int, currency: str, price_list_id: str,
lifecycle: str, customer_po_ref: str, ordered_on: date, confirmed_at: datetime, delivery_terms: str,
payment_terms_id: str, specs: list, openings: dict, cut_off: date, deposits: list) -> "SalesOrder":
"""An order as the one-off Odoo migration writes it (SO-24): the number, dates, prices and amounts exactly as Odoo
stored them — never re-priced or re-taxed — and read-only once done or cancelled. `openings`: {line_no: (qty,
odoo_ref)} become one opening delivery each; `deposits`: down payments on an OPEN order become deposit requests
(they are never order lines), so later invoices still deduct what is unused. Asks invoicing for nothing."""
o = cls(sales_order_id, order_no=order_no, bp_id=int(bp_id), owner=owner, currency=currency, price_list_id=price_list_id,
customer_po_ref=customer_po_ref, ordered_on=ordered_on, delivery_terms=delivery_terms, payment_terms_id=payment_terms_id,
source_kind="odoo", source_id=sales_order_id)
for sp in specs:
o.add_line(sp)
o.lifecycle, o.confirmed_at, o.confirmed_on, o.revision = lifecycle, confirmed_at, ordered_on, 1
if lifecycle != "cancelled":
for line_no, (qty, ref) in sorted(openings.items()):
o.deliveries.append(Delivery(sales_order_id, line_no, ref, qty, "opening", "", cut_off, "migration", "migration"))
if lifecycle == "confirmed":
for d in deposits:
o.invoice_requests.append(InvoiceRequest(sales_order_id, o.next_request_no(), "deposit", json.dumps(d["detail"]), d["amount"],
ZERO, d["invoice_id"], "requested", confirmed_at, "migration"))
o.events.append(ev.OrderDrafted(sales_order_id, "odoo", sales_order_id))
return o
# ── reading ───────────────────────────────────────────────────────────────────────────────
def line(self, line_no: int) -> Optional[OrderLine]:
return next((ln for ln in self.lines if ln.line_no == int(line_no)), None)
def _need_line(self, line_no: int) -> OrderLine:
ln = self.line(line_no)
if ln is None:
raise RuleBroken(f"the order has no line {line_no}")
return ln
def delivered(self, line_no: int) -> Decimal:
return sum((d.quantity for d in self.deliveries if d.line_no == int(line_no)), ZERO)
def requested_qty(self, line_no: int) -> Decimal:
"""Quantity of a line already asked of invoicing (cancelled requests excluded)."""
total = ZERO
for r in self.invoice_requests:
if not r.live or r.kind not in ("regular", "final"):
continue
for x in r.lines:
if x.get("line_no") == int(line_no):
total += D(x["quantity"])
return total
def requested_net(self, line_no: int) -> Decimal:
"""Net already asked of invoicing for a line (cancelled requests excluded): a conditioned line bills its net in parts, the last
part taking the remainder, so the parts add up to the line's net exactly."""
total = ZERO
for r in self.invoice_requests:
if not r.live or r.kind not in ("regular", "final"):
continue
for x in r.lines:
if x.get("line_no") == int(line_no):
total += D(x.get("net_amount") or 0)
return total
def invoiced_qty(self, line_no: int, issued: dict) -> Decimal:
"""Invoiced = the larger of what invoicing says it issued and what we already asked it for: never twice."""
return max(D(issued.get(int(line_no), 0)), self.requested_qty(line_no))
def due_qty(self, line_no: int, issued: dict) -> Decimal:
"""Quantity invoiceable now, per the line's policy (SO-12). Repeating lines bill from their schedule."""
ln = self._need_line(line_no)
if ln.repeating or ln.milestone or self.lifecycle != "confirmed":
return ZERO
basis = ln.quantity if ln.invoicing_policy == "ordered" and not self.closed(ln) else self.delivered(ln.line_no)
return max(ZERO, min(basis, ln.quantity) - self.invoiced_qty(ln.line_no, issued))
def deposit_total(self) -> Decimal:
return sum((r.amount for r in self.invoice_requests if r.kind == "deposit" and r.live), ZERO)
def unused_deposit(self) -> Decimal:
used = sum((r.deducted for r in self.invoice_requests if r.kind in ("regular", "final") and r.live), ZERO)
return self.deposit_total() - used
@property
def payer(self) -> int:
return self.payer_bp_id or self.bp_id
@property
def bill_to(self) -> int:
return self.bill_to_bp_id or self.bp_id
def closed(self, ln: OrderLine) -> bool:
"""A line is closed when it is delivered in full or within its under-delivery tolerance (98 of 100 at 2%): nothing is back-ordered."""
got = self.delivered(ln.line_no)
return got > 0 and got >= ln.quantity * (100 - self.under_tolerance_pct) / 100
def target(self, ln: OrderLine) -> Decimal:
"""What the line is expected to deliver and bill: the ordered quantity, or what was delivered once it closed within tolerance."""
return min(self.delivered(ln.line_no), ln.quantity) if self.closed(ln) else ln.quantity
def fulfilment_status(self) -> str:
billable = [ln for ln in self.lines]
if not billable or all(self.delivered(ln.line_no) <= 0 for ln in billable):
return "not started"
return "delivered" if all(self.closed(ln) for ln in billable) else "partly delivered"
def part_deliveries(self, line_no: int) -> int:
return len({d.delivery_ref for d in self.deliveries if d.line_no == int(line_no) and d.kind == "delivery" and d.quantity > 0})
def active_block(self, kind: str) -> Optional[OrderBlock]:
rows = [b for b in self.blocks if b.kind == kind]
return rows[-1] if rows and rows[-1].event == "set" else None
def schedule_version(self) -> int:
return max((r.schedule_version for r in self.schedule), default=0)
def in_force(self, line_no: int) -> list:
"""The schedule lines of a line in force: those of its highest schedule version, in sequence."""
rows = [r for r in self.schedule if r.line_no == int(line_no)]
top = max((r.schedule_version for r in rows), default=0)
return sorted((r for r in rows if r.schedule_version == top), key=lambda r: r.seq)
def append_schedule(self, version: int, line_no: int, chunks: list, *, requested_on: Optional[date], reason: str, at: datetime, trigger_key: str = "") -> None:
"""Append a line's schedule lines under `version`: it must be above every version so far (they only increase) and the lines must not
confirm more than the line's quantity."""
if version <= max((r.schedule_version for r in self.schedule if r.line_no == int(line_no)), default=0):
raise RuleBroken(f"line {line_no}: schedule version {version} is not above the one in force: versions only increase")
total = sum((c.quantity for c in chunks if c.on), ZERO)
if total > self._need_line(line_no).quantity:
raise RuleBroken(f"line {line_no}: the schedule lines confirm {total} of {self._need_line(line_no).quantity} ordered")
for n, c in enumerate(chunks, 1):
self.schedule.append(ScheduleLine(self.sales_order_id, int(line_no), version, n, requested_on, c.quantity, c.on, c.material_on,
c.quantity if c.on else ZERO, c.base if c.on else ZERO, reason, at, trigger_key))
def _next_block_no(self) -> int:
highest = max((b.block_no for b in self.blocks), default=0) # a child number inside one aggregate: its version guards it
return highest + 1
def set_block(self, kind: str, reason_code: str, note: str, *, by_email: str, by_sub: str, at: datetime) -> OrderBlock:
if self.lifecycle != "confirmed":
raise RuleBroken(f"the order is {self.lifecycle}: only a confirmed order is blocked")
if self.active_block(kind) is not None:
raise RuleBroken(f"the {kind} is already blocked ({self.active_block(kind).reason_code}): release it first")
b = OrderBlock(self.sales_order_id, self._next_block_no(), kind, reason_code, "set", note, by_email, by_sub, at)
self.blocks.append(b)
return b
def release_block(self, kind: str, note: str, *, by_email: str, by_sub: str, at: datetime, approval: bool = False) -> OrderBlock:
cur = self.active_block(kind)
if cur is None:
raise RuleBroken(f"the {kind} is not blocked")
if cur.reason_code == "APPROVAL" and not approval:
raise GateUnmet([no("approval_block", "an APPROVAL block is released only by its approval", "the approver releases it")])
b = OrderBlock(self.sales_order_id, self._next_block_no(), kind, cur.reason_code, "release", note, by_email, by_sub, at)
self.blocks.append(b)
return b
def default_block(self, code: str, at: datetime) -> None:
"""An order type's default block (APPROVAL) starts on confirmation, on delivery and billing both."""
if code and code != "none":
for kind in ("delivery", "billing"):
self.blocks.append(OrderBlock(self.sales_order_id, self._next_block_no(), kind, code, "set", "the type's default block", "system", "system", at))
def invoice_status(self, issued: dict, skip=frozenset()) -> str:
"""`skip`: the lines that are never billed (a free-of-charge line, a consignment fill, a return: its credit is not an invoice)."""
lines = [ln for ln in self.lines if not ln.repeating and ln.line_no not in skip]
if not lines or self.lifecycle in ("draft", "cancelled"):
return "nothing to invoice"
plain = [ln for ln in lines if not ln.milestone]
invoiced = [self.invoiced_qty(ln.line_no, issued) for ln in plain]
stages = [m for ln in lines if ln.milestone for m in self.stages(ln.line_no)]
if all(i >= self.target(ln) for i, ln in zip(invoiced, plain)) and all(m.billed_request_no for m in stages):
return "invoiced"
any_billed = any(i > 0 for i in invoiced) or any(m.billed_request_no for m in stages)
if any(self.due_qty(ln.line_no, issued) > 0 for ln in plain) or any(m.released_on and not m.billed_request_no for m in stages):
return "partly invoiced" if any_billed else "to invoice"
return "partly invoiced" if any_billed else "nothing to invoice"
def has_open_suggestion(self) -> bool:
return any(s.decision == "open" for s in self.suggestions)
# ── conditions (SO-30, SO-31) ─────────────────────────────────────────────────────────────
def _put_base(self, ln: OrderLine, revision: int) -> None:
"""The BASE row of a line: product's price (find_price), or the quote's / Odoo's, as the line was written."""
self.conditions = [c for c in self.conditions if not (c.line_no == ln.line_no and c.step == 10 and c.revision == revision)]
self.conditions.append(LineCondition(self.sales_order_id, ln.line_no, 10, revision, pr.BASE, ln.unit_price, ln.net_amount, ZERO, ln.priced_from))
def current_conditions(self, line_no: int, below: Optional[int] = None) -> list:
"""The rows in force for a line: each step's latest revision (below `below` when asked), in step order."""
by: dict = {}
for c in sorted((c for c in self.conditions if c.line_no == int(line_no) and (below is None or c.revision < below)),
key=lambda c: c.revision):
by[c.step] = c
return [by[k] for k in sorted(by)]
def current_header_conditions(self, below: Optional[int] = None) -> list:
by: dict = {}
for h in sorted((h for h in self.header_conditions if below is None or h.revision < below), key=lambda h: h.revision):
by[h.condition_type_code] = h
return [by[k] for k in sorted(by, key=lambda k: by[k].step)]
def conditioned(self, line_no: int) -> bool:
"""True when anything beyond the BASE price is on the line: its net is then its conditions' sum, not quantity x price."""
return any(c.condition_type_code != pr.BASE and c.amount != 0 for c in self.current_conditions(line_no))
def set_manual_condition(self, line_no: int, ctype, step: int, rate: Decimal, by: str) -> None:
"""A condition typed by hand on a DRAFT line, within its type's limit (gate manual_within_limit); recorded as manual with who."""
self._editable()
self._need_line(line_no)
if ctype.level != "item" or not ctype.manual_allowed or ctype.calculation == "base":
raise RuleBroken(f"{ctype.condition_type_code} cannot be typed on a line by hand")
if ctype.calculation == "percent" and ctype.manual_limit_pct is not None and rate > ctype.manual_limit_pct:
raise GateUnmet([no("manual_within_limit", f"{ctype.condition_type_code} may be changed by hand up to {num(ctype.manual_limit_pct)}%",
f"type {num(ctype.manual_limit_pct)}% or less, or ask for a condition record")])
self.conditions = [c for c in self.conditions if not (c.line_no == int(line_no) and c.step == step and c.revision == self.revision)]
self.conditions.append(LineCondition(self.sales_order_id, int(line_no), step, self.revision, ctype.condition_type_code, rate,
ZERO, ZERO, f"manual:{by}"))
def set_header_condition(self, ctype, step: int, rate: Decimal, by: str) -> None:
"""A header condition typed by hand on a DRAFT order (a header discount), within its type's limit."""
self._editable()
if ctype.level != "header" or not ctype.manual_allowed:
raise RuleBroken(f"{ctype.condition_type_code} cannot be typed on an order by hand")
if ctype.calculation == "percent" and ctype.manual_limit_pct is not None and rate > ctype.manual_limit_pct:
raise GateUnmet([no("manual_within_limit", f"{ctype.condition_type_code} may be changed by hand up to {num(ctype.manual_limit_pct)}%",
f"type {num(ctype.manual_limit_pct)}% or less")])
self.header_conditions = [h for h in self.header_conditions if not (h.condition_type_code == ctype.condition_type_code and h.revision == self.revision)]
self.header_conditions.append(HeaderCondition(self.sales_order_id, ctype.condition_type_code, self.revision, step, rate, ZERO, f"manual:{by}"))
def reprice(self, *, only, types: dict, steps: list, access: list, records: list, on: date, revision: int, price_group: str = "",
statistical=frozenset()) -> set:
"""Price the lines in `only` through the procedure (each condition the first record its access sequence finds, or the typed one),
then compute the header conditions and spread them over ALL lines. Rows land at `revision` (a draft rewrites its own; an amendment
appends, and a step that no longer applies gets a zero row so the old one is never read as current). Each line's net is its rows'
sum, exactly. Returns the lines whose net changed (tax is asked again for them)."""
before = {ln.line_no: ln.net_amount for ln in self.lines}
only = {int(n) for n in only}
item_net: dict = {}
for ln in self.lines:
if not self.current_conditions(ln.line_no): # an order written before conditions existed: its price is its BASE
self._put_base(ln, revision)
earlier = {c.step: c for c in self.current_conditions(ln.line_no, below=revision) if not c.header_share_of}
if ln.line_no in only:
qty_base = Fraction(ln.quantity) * ln.unit_numerator / ln.unit_denominator
base = (Fraction(ln.quantity_priced) * Fraction(ln.unit_price) / ln.pricing_quantity if ln.procedural
else Fraction(next((c.amount for c in self.current_conditions(ln.line_no) if c.step == 10), ln.net_amount)))
chosen = {c.condition_type_code: (c.rate, c.source, Fraction(ln.quantity)) for c in self.current_conditions(ln.line_no)
if c.source.startswith("manual:") and c.step != 10 and not c.header_share_of}
ctx = pr.Ctx(self.bp_id, price_group or self.price_group, ln.product_id, ln.category_code, self.ship_to_country, qty_base, on)
if ln.procedural:
for st in steps:
t = types.get(st.condition_type_code)
if t is None or t.level != "item" or t.calculation == "base" or t.condition_type_code in chosen:
continue
rec, via = pr.find_record(t, access, records, ctx)
if rec is not None:
chosen[t.condition_type_code] = (rec.rate_for(qty_base), f"record {t.condition_type_code} {via.key_fields} {rec.key_value} from {rec.valid_from}",
Fraction(ln.quantity))
rows, _net = pr.item_rows(base, ln.unit_price, ln.priced_from, steps, types, chosen)
self.conditions = [c for c in self.conditions if not (c.line_no == ln.line_no and c.revision == revision and not c.header_share_of)]
have = set()
for r in rows:
have.add(r.step)
self.conditions.append(LineCondition(self.sales_order_id, ln.line_no, r.step, revision, r.condition_type_code, r.rate, r.amount,
r.basis, r.source, r.statistical or ln.line_no in statistical))
for step, old in earlier.items():
if step not in have:
self.conditions.append(LineCondition(self.sales_order_id, ln.line_no, step, revision, old.condition_type_code, ZERO, ZERO, ZERO,
"no longer applies"))
item_net[ln.line_no] = sum((c.amount for c in self.current_conditions(ln.line_no) if not c.header_share_of and not c.statistical), ZERO)
# the header conditions: typed ones (a header discount) and found ones (freight), then their shares on every line
typed = {h.condition_type_code: h for h in self.current_header_conditions() if h.source.startswith("manual:")}
earlier_h = {h.condition_type_code: h for h in self.current_header_conditions(below=revision)}
self.header_conditions = [h for h in self.header_conditions if h.revision != revision]
self.conditions = [c for c in self.conditions if not (c.revision == revision and c.header_share_of)]
unknown, shares, new_h = [], {}, []
for st in sorted(steps, key=lambda s: s.step):
t = types.get(st.condition_type_code)
if t is None or t.level != "header":
continue
weights = {n: Fraction(v) for n, v in item_net.items()}
minimum = None
if t.condition_type_code in typed:
rate, src, base = typed[t.condition_type_code].rate, typed[t.condition_type_code].source, sum(weights.values(), Fraction(0))
else:
rec, via = pr.find_record(t, access, records, pr.Ctx(self.bp_id, price_group or self.price_group, "", "", self.ship_to_country, Fraction(0), on))
if rec is None:
continue
rate, minimum = rec.rate_for(Fraction(0)), rec.minimum
src = f"record {t.condition_type_code} {via.key_fields} {rec.key_value} from {rec.valid_from}"
if t.calculation == "per_weight":
kg = {ln.line_no: pr.line_weight_kg(ln.quantity, ln.gross_weight, ln.weight_unit_code) for ln in self.lines}
unknown = [n for n, w in kg.items() if w is None]
if unknown:
continue
weights, base = kg, sum(kg.values(), Fraction(0))
else:
base = sum(weights.values(), Fraction(0))
amount = pr.header_amount(t, rate, base, minimum)
new_h.append(HeaderCondition(self.sales_order_id, t.condition_type_code, revision, st.step, rate, amount, src))
shares[t.condition_type_code] = (st.step, pr.spread(amount, weights if t.header_spread == "weight" or t.calculation == "per_weight"
else {n: Fraction(v) for n, v in item_net.items()}), rate, src)
self.freight_unknown = ",".join(str(n) for n in unknown)
self.header_conditions += new_h
for code, old in earlier_h.items():
if code not in shares:
self.header_conditions.append(HeaderCondition(self.sales_order_id, code, revision, old.step, ZERO, ZERO, "no longer applies"))
for ln in self.lines:
for code, (step, per_line, rate, src) in shares.items():
self.conditions.append(LineCondition(self.sales_order_id, ln.line_no, step, revision, code, rate, per_line.get(ln.line_no, ZERO), ZERO, src,
False, code))
have_share = {st for st, *_ in shares.values()}
for c in self.current_conditions(ln.line_no, below=revision):
if c.header_share_of and c.step not in have_share:
self.conditions.append(LineCondition(self.sales_order_id, ln.line_no, c.step, revision, c.condition_type_code, ZERO, ZERO, ZERO,
"no longer applies", False, c.header_share_of))
ln.net_amount = sum((c.amount for c in self.current_conditions(ln.line_no) if not c.statistical), ZERO)
if ln.line_no in statistical: # priced for statistics only: what it would have sold for is shown, nothing is billed
ln.statistical_value = sum((c.amount for c in self.current_conditions(ln.line_no) if not c.header_share_of), ZERO)
self._recompute()
return {ln.line_no for ln in self.lines if ln.net_amount != before.get(ln.line_no)}
# ── the draft ─────────────────────────────────────────────────────────────────────────────
def _editable(self) -> None:
if self.lifecycle == "draft":
return
raise RuleBroken({"confirmed": "confirmed: its lines and prices are frozen; amend it with a reason",
"done": "done: raise a new order", "cancelled": "cancelled: raise a new order"}[self.lifecycle])
def _recompute(self) -> None:
self.subtotal = sum((ln.net_amount for ln in self.lines), ZERO)
self.tax_total = sum((ln.tax_amount for ln in self.lines), ZERO)
self.total = self.subtotal + self.tax_total
def add_line(self, spec: LineSpec) -> OrderLine:
self._editable()
if spec.quantity <= 0:
raise RuleBroken("a quantity is more than zero")
ln = OrderLine.from_spec(self.sales_order_id, max((x.line_no for x in self.lines), default=0) + 1, spec)
self.lines.append(ln)
self._put_base(ln, self.revision)
self._recompute()
return ln
def remove_line(self, line_no: int) -> None:
self._editable()
self.lines.remove(self._need_line(line_no))
self.conditions = [c for c in self.conditions if c.line_no != int(line_no)]
self._recompute()
def retax(self, line_no: int, tax_code: str, tax_amount: Decimal) -> None:
"""The tax module's answer for a line of a CONFIRMED order that an amendment (or a header share) re-priced: the new revision's copy."""
ln = self._need_line(line_no)
ln.tax_code, ln.tax_amount = tax_code, money(tax_amount)
self._recompute()
def set_tax(self, line_no: int, tax_code: str, tax_amount: Decimal) -> None:
"""The tax module's answer for a line, stored as the order's own copy. Draft only: confirmation freezes it."""
self._editable()
ln = self._need_line(line_no)
ln.tax_code, ln.tax_amount = tax_code, money(tax_amount)
self._recompute()
def record_customer_po(self, ref: str) -> None:
self._editable()
self.customer_po_ref = ref.strip()
def override_credit(self, reason: str, by: str) -> None:
self._editable()
if not reason.strip():
raise RuleBroken("overriding a credit hold needs a reason")
self.credit_override_by, self.credit_override_reason = by, reason.strip()
def cancel_draft(self) -> bool:
if self.lifecycle == "cancelled":
return False
if self.lifecycle != "draft":
raise RuleBroken(f"the order is {self.lifecycle}: only a draft is cancelled this way")
self.lifecycle = "cancelled"
self.events.append(ev.OrderCancelled(self.sales_order_id, "draft cancelled"))
return True
# ── confirm ───────────────────────────────────────────────────────────────────────────────
def confirm(self, *, gates: list, order_no: str, now: datetime, today: date) -> None:
if self.lifecycle != "draft":
raise RuleBroken(f"the order is {self.lifecycle}, not a draft")
require(gates)
self.record_warnings(gates)
self.order_no, self.lifecycle, self.confirmed_at, self.confirmed_on = order_no, "confirmed", now, today
self.revision = 1
self.events.append(ev.OrderConfirmed(self.sales_order_id, order_no))
def record_warnings(self, gates: list) -> None:
"""A gate met WITH a reason in warn mode (a line off its delivery multiple) is shown and recorded on the order: nothing blocks, nothing hides."""
kept = json.loads(self.warnings_json or "{}")
for g in gates:
if g.gate == "delivery_multiple":
kept.pop(g.gate, None)
if g.met and g.reason:
kept[g.gate] = g.as_dict()
self.warnings_json = json.dumps(kept, sort_keys=True)
def confirm_migrated(self, *, order_no: str, at: datetime, on: date) -> None:
"""An open Odoo order arrives already confirmed, under its own number; the migration is the one writer."""
self.order_no, self.lifecycle, self.confirmed_at, self.confirmed_on, self.revision = order_no, "confirmed", at, on, 1
# ── deliveries ────────────────────────────────────────────────────────────────────────────
def record_delivery(self, line_no: int, quantity: Decimal, delivery_ref: str, delivered_on: date, *, today: date,
by: str, issued: dict, kind: str = "delivery", corrects_ref: str = "",
source: str = "manual", quantity_base: Decimal = ZERO, inventory: bool = False, hold: tuple = ()) -> Optional[Delivery]:
"""Append a delivery (or a correction against an earlier one). Recording the same reference twice records it
once (returns None). The migration's opening delivery is the one past-dated write."""
if self.lifecycle != "confirmed":
raise RuleBroken(f"the order is {self.lifecycle}: deliveries are recorded on a confirmed order")
ln = self._need_line(line_no)
ref = delivery_ref.strip()
if not ref:
raise RuleBroken("a delivery needs its reference (the delivery note)")
if any(d.line_no == ln.line_no and d.delivery_ref == ref for d in self.deliveries):
return None
require([*hold, block_gate(self, "delivery")] if kind != "opening" else [])
third = ln.item_category_code == "THIRD_PARTY"
if kind != "opening" and source == "timesheets" and ln.product_type_code != "SERVICE": # SO-51: approved hours are a service's delivery
raise GateAnswer([no("service_line", f"line {ln.line_no} ({ln.product_id}) is not a service line: timesheets delivers hours, not goods",
"pick the order's service line (in HUR or DAY), or record the delivery by hand")])
if kind != "opening" and source == "purchasing" and not third:
raise GateUnmet([no("supplier_delivery", f"{ln.product_id} is not a third-party line: purchasing does not deliver it",
"record it by hand (source manual), or by inventory where it is stocked")])
if kind != "opening" and third and source != "purchasing":
raise GateUnmet([no("third_party_delivery", "a third-party line is delivered by the supplier",
"purchasing records the supplier's shipment (source purchasing, delivery_ref pur:<shipping notice>)")])
if source == "inventory" and not ln.stocked:
raise GateUnmet([no("stocked_from_inventory", f"{ln.product_id} is not stocked: its deliveries are recorded by hand, not by inventory",
f"record it by hand (source manual) on line {ln.line_no}")])
if inventory and ln.stocked and source == "manual" and kind != "opening":
raise GateUnmet([no("stocked_from_inventory", f"{ln.product_id} is stocked: its deliveries come from inventory",
"validate the delivery in Inventory; it records here")])
if kind != "opening":
if delivered_on > today:
raise RuleBroken(f"a delivery is never dated in the future (today is {today})")
if self.confirmed_on and delivered_on < self.confirmed_on:
raise RuleBroken(f"a delivery cannot be dated before the confirmation ({self.confirmed_on})")
total = self.delivered(ln.line_no) + quantity
if kind == "correction":
orig = next((d for d in self.deliveries if d.line_no == ln.line_no and d.delivery_ref == corrects_ref
and d.kind != "correction"), None)
if orig is None:
raise RuleBroken(f"a correction names the delivery it corrects; {corrects_ref or '(none)'} is not one on line {ln.line_no}")
if quantity >= 0:
raise RuleBroken("a correction is a negative quantity against the original")
net = orig.quantity + sum((d.quantity for d in self.deliveries if d.corrects_ref == corrects_ref and d.line_no == ln.line_no), ZERO)
if net + quantity < 0:
raise RuleBroken(f"{corrects_ref} delivered {num(net)}: it cannot be corrected by {num(quantity)}")
floor = self.invoiced_qty(ln.line_no, issued) if ln.invoicing_policy == "delivered" else ZERO
if total < floor and source == "timesheets": # SO-51: the hours were already invoiced
raise GateAnswer([no("not_below_invoiced", f"line {ln.line_no}: a correction cannot take delivered ({num(total)}) below what is already invoiced ({num(floor)})",
"credit the invoice in invoicing")])
if total < floor:
raise RuleBroken(f"line {ln.line_no}: a correction cannot take delivered ({num(total)}) below what is already invoiced ({num(floor)})")
else:
if quantity <= 0:
raise RuleBroken("a delivery is more than zero; a mistake is a correction")
cap = ln.quantity * (100 + self.over_tolerance_pct) / 100
if total > cap:
raise (GateAnswer if source == "timesheets" else GateUnmet)([no("within_ordered" if source == "timesheets" else "within_tolerance", f"line {ln.line_no}: delivering {num(quantity)} would make {num(total)} of {num(ln.quantity)} ordered"
+ (f" (over-delivery tolerance {num(self.over_tolerance_pct)}%)" if self.over_tolerance_pct else ""),
f"deliver at most {num(cap - self.delivered(ln.line_no))}")])
if self.max_partials and self.part_deliveries(ln.line_no) >= self.max_partials:
raise GateUnmet([no("max_partials", f"line {ln.line_no} already has {self.part_deliveries(ln.line_no)} part deliveries: the customer allows at most {self.max_partials}",
"ask an order admin to change the order's delivery rule, or correct an earlier delivery")])
if self.partial_delivery == "complete_line" and total < ln.quantity * (100 - self.under_tolerance_pct) / 100:
raise GateUnmet([no("complete_line", f"this customer takes a line complete: line {ln.line_no} is {num(total)} of {num(ln.quantity)}",
f"deliver {num(ln.quantity - self.delivered(ln.line_no))} at once")])
d = Delivery(self.sales_order_id, ln.line_no, ref, quantity, kind, corrects_ref, delivered_on, by, source, quantity_base)
self.deliveries.append(d)
self.events.append(ev.DeliveryRecorded(self.sales_order_id, ln.line_no, quantity, ref))
return d
# ── invoicing requests ────────────────────────────────────────────────────────────────────
def plan_invoice(self, issued: dict, *, skip=frozenset(), deposits: Optional[list] = None, manual: Optional[dict] = None) -> dict:
"""What to ask invoicing for now: the lines due (a milestone line by the value of its released stages), the deposits deducted (per
deposit, oldest first: `deposits` are the ISSUED ones with what each has left) and whether this finishes the billing."""
gates = gate_rules.invoice_gates(self, gate_rules.Facts(issued=issued), skip=skip)
require(gates)
lines, net, stage_nos = [], ZERO, []
for ln in self.lines:
if ln.line_no in skip:
continue
if ln.milestone:
amounts = bill.milestone_amounts(ln.net_amount, self.stages(ln.line_no))
stages = self.stages(ln.line_no)
for m in stages:
if m.released_on and not m.billed_request_no:
label = m.name or m.event
lines.append({"line_no": ln.line_no, "product_id": ln.product_id, "unit_code": ln.unit_code, "quantity": "1",
"description": f"Milestone {m.milestone_no} of {len(stages)}: {label} ({num(m.pct)}%)",
"unit_price": num(amounts[m.milestone_no], 2), "discount_pct": "0", "tax_category": ln.tax_category,
"tax_code": ln.tax_code, "net_amount": num(amounts[m.milestone_no], 2), "milestone_no": m.milestone_no})
net += amounts[m.milestone_no]
stage_nos.append((ln.line_no, m.milestone_no))
continue
q = self.due_qty(ln.line_no, issued)
if q <= 0:
continue
price, pct = ln.billed_price()
entry = {"line_no": ln.line_no, "product_id": ln.product_id, "description": ln.description,
"unit_code": ln.unit_code, "quantity": num(q), "unit_price": num(price, 2),
"discount_pct": num(pct), "tax_category": ln.tax_category, "tax_code": ln.tax_code}
if self.conditioned(ln.line_no): # the line's net is its conditions' sum: bill that net in parts, the last taking the remainder
last = self.requested_qty(ln.line_no) + q >= ln.quantity
amount = ln.net_amount - self.requested_net(ln.line_no) if last else money(ln.net_amount * q / ln.quantity)
entry["billed_net"] = num(amount, 2)
else:
amount = money(priced_quantity(q, ln.unit_code, ln.unit_numerator, ln.unit_denominator, ln.pricing_unit_code,
ln.pricing_numerator, ln.pricing_denominator) * price / ln.pricing_quantity * (100 - pct) / 100)
if ln.pricing_quantity != 1 or (ln.pricing_unit_code and ln.pricing_unit_code != ln.unit_code):
entry["pricing_quantity"], entry["pricing_unit_code"] = ln.pricing_quantity, ln.pricing_unit_code or ln.unit_code
entry["quantity_priced"] = num(priced_quantity(q, ln.unit_code, ln.unit_numerator, ln.unit_denominator, ln.pricing_unit_code,
ln.pricing_numerator, ln.pricing_denominator)) # v8: ar's net = quantity_priced x price / pricing_quantity
net += amount
lines.append({**entry, "net_amount": num(amount, 2)})
plain = [ln for ln in self.lines if not ln.repeating and not ln.milestone and ln.line_no not in skip]
mine = {n for n in stage_nos}
final = (all(self.invoiced_qty(ln.line_no, issued) + self.due_qty(ln.line_no, issued) >= self.target(ln) for ln in plain)
and all(m.billed_request_no or (m.line_no, m.milestone_no) in mine
for ln in self.lines if ln.milestone and ln.line_no not in skip for m in self.stages(ln.line_no)))
picks = bill.plan_deductions(self.deposit_deduction_mode, net=net, order_net=self.subtotal, final=final, deposits=deposits or [],
manual=manual or {}) if deposits else []
return {"kind": "final" if final else "regular", "lines": lines, "net": net, "deduct": sum((a for _, a in picks), ZERO),
"deductions": [{"deposit_invoice_id": d["invoice_id"], "net_amount": num(a, 2), "tax_category": d["tax_category"],
"request_no": d["request_no"]} for d, a in picks], "stages": stage_nos}
def add_invoice_request(self, kind: str, lines: list, amount: Decimal, deducted: Decimal, *, by: str,
now: datetime, invoice_id: str = "", status: str = "requested", deductions: Optional[list] = None,
source_ref: str = "") -> InvoiceRequest:
n = self.next_request_no()
r = InvoiceRequest(self.sales_order_id, n, kind, json.dumps(lines), amount, deducted, invoice_id, status, now, by,
json.dumps(deductions or []), source_ref)
self.invoice_requests.append(r)
for line_no, no_ in (self._stage_keys(lines) if kind in ("regular", "final") else []):
next(m for m in self.milestones if (m.line_no, m.milestone_no) == (line_no, no_)).billed_request_no = n
self.events.append(ev.InvoiceRequested(self.sales_order_id, n, kind))
return r
def next_request_no(self) -> int:
"""The next request number: above the highest this ORDER has (a child number inside one aggregate, never a global counter).
Two writers racing for it both load the same order version, so the second save loses the version claim and is a 409."""
highest = max((r.request_no for r in self.invoice_requests), default=0)
return highest + 1
def plan_deposit(self, amount: Optional[Decimal] = None) -> Decimal:
gates = gate_rules.deposit_gates(self, amount)
require(gates)
return amount if amount is not None else money(self.subtotal * self.deposit_pct / 100)
# ── amend ─────────────────────────────────────────────────────────────────────────────────
def propose_amendment(self, changes: list, reason: str, *, by: str, now: datetime, issued: dict, identities: tuple = (),
multiples: Optional[dict] = None, delivery_mode: str = "warn") -> Revision:
if not reason.strip():
raise RuleBroken("an amendment needs a reason")
if not changes:
raise RuleBroken("an amendment changes at least one line")
for ch in changes:
ln = self._need_line(ch["line_no"])
if ch.get("quantity") is not None and ch["quantity"] <= 0:
raise RuleBroken(f"line {ln.line_no}: a quantity is more than zero; reduce to what is delivered or cancel the order")
if ch.get("quantity") is None and not ch.get("unit_code"):
raise RuleBroken(f"line {ln.line_no}: say what changes (quantity or unit_code)")
require(gate_rules.amend_gates(self, gate_rules.Facts(issued=issued, multiples=multiples, delivery_mode=delivery_mode), changes))
rev = Revision(self.sales_order_id, self.revision + 1, reason.strip(),
json.dumps([{**c, "quantity": num(c["quantity"]) if c.get("quantity") is not None else None} for c in changes]),
"pending", by, now, proposed_sub=_sub_of(by, identities))
self.revisions.append(rev)
return rev
def apply_revision(self, rev_no: int, new_lines: dict, *, by: str, now: datetime, issued: dict,
needs_approval: bool, identities: tuple = (), multiples: Optional[dict] = None, delivery_mode: str = "warn") -> Revision:
"""Apply a pending amendment: only the changed lines take the new (re-priced) spec; the rest keep their price
and source. Never approved by the person who proposed it where the control asks for approval."""
rev = next((r for r in self.revisions if r.revision == rev_no), None)
if rev is None or rev.status != "pending":
raise RuleBroken(f"there is no pending revision {rev_no}")
gates = gate_rules.amend_gates(self, gate_rules.Facts(issued=issued, multiples=multiples, delivery_mode=delivery_mode), rev.changes,
rev.proposed_by, by, needs_approval, proposer_ids=(rev.proposed_sub,), approver_ids=identities)
require(gates)
self.record_warnings(gates)
for line_no, spec in new_lines.items():
old = self._need_line(line_no)
fresh = OrderLine.from_spec(self.sales_order_id, old.line_no, spec)
fresh.quote_ref = old.quote_ref
fresh.purchase_request_id = old.purchase_request_id # purchasing's request stays on the line until the amended one replaces it
for k in ("ref_sales_order_id", "ref_line_no", "ref_delivery_ref", "ref_invoice_id", "ref_invoice_line_no"):
if not getattr(fresh, k):
setattr(fresh, k, getattr(old, k)) # what the line references is not re-priced away
self.lines[self.lines.index(old)] = fresh
self._put_base(fresh, rev.revision)
self._recompute()
if self.deposit_total() > self.subtotal:
require([no("within_total", f"deposits of {num(self.deposit_total(), 2)} would be more than the amended order total {num(self.subtotal, 2)}, which the final "
"invoice could not take and nothing would credit", "credit the unused deposit first, or amend to at least the deposits")])
self.revision = rev.revision
rev.status, rev.approved_by, rev.approved_at = "applied", ("" if not needs_approval else by), now
self.events.append(ev.OrderAmended(self.sales_order_id, rev.revision))
return rev
def reject_amendment(self, rev_no: int, *, by: str, now: datetime) -> Revision:
rev = next((r for r in self.revisions if r.revision == rev_no and r.status == "pending"), None)
if rev is None:
raise RuleBroken(f"there is no pending revision {rev_no}")
rev.status, rev.approved_by, rev.approved_at = "rejected", by, now
return rev
# ── cancel, finish ────────────────────────────────────────────────────────────────────────
def cancel_order(self, reason: str, *, by: str, issued: dict, issued_deposits: tuple = ()) -> list:
"""Cancel a confirmed order that nothing has happened to. Returns the invoicing drafts it asked for, so the
use case can have invoicing cancel them."""
if not reason.strip():
raise RuleBroken("cancelling a confirmed order needs a reason")
require(gate_rules.cancel_gates(self, gate_rules.Facts(issued=issued, issued_deposits=list(issued_deposits))))
drafts = []
for r in self.invoice_requests:
if r.status in ("requested", "pending") and r.kind not in ("credit", "debit") \
and f"so:{self.sales_order_id}:{r.request_no}" not in issued_deposits: # an ISSUED deposit is credited, never cancelled
r.status = "cancelled"
drafts.append(f"so:{self.sales_order_id}:{r.request_no}")
for rv in self.revisions:
if rv.status == "pending":
rv.status = "rejected"
self.lifecycle, self.cancel_reason = "cancelled", reason.strip()
self.events.append(ev.OrderCancelled(self.sales_order_id, reason.strip()))
return drafts
def can_finish(self, issued: dict, skip=frozenset()) -> bool:
"""Every line delivered (or closed within tolerance) and invoiced in full; a line in `skip` is never billed (done when delivered), a
milestone line is done when every stage is billed. A return closes through close_return, a memo request through finish_memo."""
if self.order_type_code in RETURN_TYPES or self.order_type_code in MEMO_TYPES:
return False
lines = [ln for ln in self.lines if not ln.repeating]
if not lines:
return False
for ln in lines:
if ln.milestone:
if not all(m.billed_request_no for m in self.stages(ln.line_no)):
return False
continue
if not self.closed(ln) or (ln.line_no not in skip and D(issued.get(ln.line_no, 0)) < self.target(ln)):
return False
return True
def finish(self, issued: dict, skip=frozenset()) -> bool:
"""Lock a fully delivered and invoiced order. Runs once however often it is triggered."""
if self.lifecycle == "done":
return False
if self.lifecycle != "confirmed":
raise RuleBroken(f"the order is {self.lifecycle}, not confirmed")
if not self.can_finish(issued, skip):
raise RuleBroken("not every line is delivered and invoiced in full")
self.lifecycle = "done"
self.events.append(ev.OrderDone(self.sales_order_id))
return True
def record_send(self, *, channel: str, address: str, by: str, at: datetime) -> Send:
if self.lifecycle not in ("confirmed", "done"):
raise RuleBroken("a confirmation is sent for a confirmed order")
s = Send(self.sales_order_id, at, channel, address, self.revision, by, self.schedule_version())
self.sends.append(s)
return s
# ── milestones (SO-46) ────────────────────────────────────────────────────────────────────
def stages(self, line_no: int) -> list:
return sorted((m for m in self.milestones if m.line_no == int(line_no)), key=lambda m: m.milestone_no)
@staticmethod
def _stage_keys(lines: list) -> list:
return [(x["line_no"], x["milestone_no"]) for x in lines if x.get("milestone_no")]
def set_milestones(self, line_no: int, items: list) -> None:
"""The stages of a DRAFT line (percent, event, name): the line becomes a milestone line, billed by value."""
self._editable()
ln = self._need_line(line_no)
ln.billing_plan = "milestone"
self.milestones = [m for m in self.milestones if m.line_no != ln.line_no]
for n, it in enumerate(items, 1):
self.milestones.append(BillingMilestone(self.sales_order_id, ln.line_no, n, it["pct"], it.get("event") or "confirmation",
it.get("name") or "", it.get("due_on")))
def plan_complete(self) -> Optional[str]:
"""None when every milestone line's stages total 100%; else why not."""
for ln in self.lines:
if ln.milestone:
total = sum((m.pct for m in self.stages(ln.line_no)), ZERO)
if total != 100:
return f"line {ln.line_no}: its milestones total {num(total)}%, not 100%"
return None
def _freeze(self, m: BillingMilestone, on: date, by: str) -> None:
ln = self._need_line(m.line_no)
m.amount = bill.milestone_amounts(ln.net_amount, self.stages(m.line_no))[m.milestone_no]
m.released_on, m.released_by = on, by
def release_on_confirmation(self, on: date) -> None:
for m in sorted(self.milestones, key=lambda m: (m.line_no, m.milestone_no)):
if m.event == "confirmation" and not m.released_on:
self._freeze(m, on, "system")
def release_milestone(self, line_no: int, milestone_no: int, *, today: date, by: str) -> BillingMilestone:
"""The fulfiller says the stage's event happened (gate: not released before; a stage with a date is released from that day)."""
if self.lifecycle != "confirmed":
raise RuleBroken(f"the order is {self.lifecycle}: milestones are released on a confirmed order")
m = next((x for x in self.milestones if (x.line_no, x.milestone_no) == (int(line_no), int(milestone_no))), None)
if m is None:
raise RuleBroken(f"line {line_no} has no milestone {milestone_no}")
if m.released_on:
raise GateUnmet([no("milestone_open", f"milestone {m.milestone_no} of line {m.line_no} was released on {m.released_on}", "nothing more to release")])
if m.due_on and m.due_on > today:
raise GateUnmet([no("milestone_event", f"milestone {m.milestone_no} is due on {m.due_on}", f"release it from {m.due_on}")])
self._freeze(m, today, by)
return m
# ── returns (SO-41, SO-49) ────────────────────────────────────────────────────────────────
def inspected(self, line_no: int) -> tuple:
rows = [i for i in self.inspections if i.line_no == int(line_no)]
return sum((i.accepted_quantity for i in rows), ZERO), sum((i.rejected_quantity for i in rows), ZERO)
def credited_qty(self, line_no: int) -> Decimal:
total = ZERO
for r in self.invoice_requests:
if r.kind == "credit" and r.live:
total += sum((D(x["quantity"]) for x in r.lines if x.get("line_no") == int(line_no)), ZERO)
return total
def add_inspection(self, line_no: int, inspection_ref: str, received: Decimal, accepted: Decimal, rejected: Decimal, decision: str, reason: str,
on: date, now: datetime) -> Optional[ReturnInspection]:
"""Append what inventory found. A repeat of the same inspection_ref records once (returns None). accepted + rejected, over every
inspection of the line, never exceed what was received (gate within_received)."""
if self.order_type_code not in RETURN_TYPES:
raise RuleBroken(f"the order is {self.order_type_code}: only a return is inspected")
ln = self._need_line(line_no)
ref = inspection_ref.strip()
if not ref:
raise RuleBroken("an inspection needs its reference (inv:<return number>:<line>)")
if any(i.line_no == ln.line_no and i.inspection_ref == ref for i in self.inspections):
return None # a repeat, even after the return closed, records nothing more
if self.lifecycle not in ("confirmed", "done"):
raise RuleBroken(f"the return is {self.lifecycle}: only an authorised return is inspected")
if decision not in ("restock", "scrap", "to_supplier"):
raise RuleBroken("the decision is restock, scrap or to_supplier")
if accepted < 0 or rejected < 0:
raise RuleBroken("an accepted or rejected quantity is not negative")
got = self.delivered(ln.line_no)
have = sum(self.inspected(ln.line_no), ZERO)
if have + accepted + rejected > got:
raise GateUnmet([no("within_received", f"line {ln.line_no}: {num(got)} received, {num(have)} already inspected: {num(accepted + rejected)} more is too many",
f"inspect at most {num(max(ZERO, got - have))}")])
row = ReturnInspection(self.sales_order_id, ln.line_no, ref, received, accepted, rejected, decision, reason.strip(), on, now)
self.inspections.append(row)
return row
def cancel_return(self, reason: str) -> None:
"""A return that is a draft or authorised, with nothing received, is cancelled (with a reason)."""
if not reason.strip():
raise RuleBroken("cancelling a return needs a reason")
if self.lifecycle not in ("draft", "confirmed"):
raise RuleBroken(f"the return is {self.lifecycle}: only a draft or an authorised return is cancelled")
if any(self.delivered(ln.line_no) > 0 for ln in self.lines):
raise GateUnmet([no("nothing_received", "inventory has already received some of it", "inspect and credit what came back")])
self.lifecycle, self.cancel_reason = "cancelled", reason.strip()
self.events.append(ev.OrderCancelled(self.sales_order_id, reason.strip()))
def close_return(self) -> bool:
"""Every line inspected and credited or rejected: the return is closed (edits refused). Runs once."""
if self.lifecycle == "done":
return False
if self.lifecycle != "confirmed" or self.order_type_code not in RETURN_TYPES:
raise RuleBroken(f"the order is {self.order_type_code} {self.lifecycle}, not an authorised return")
self.lifecycle = "done"
self.events.append(ev.OrderDone(self.sales_order_id))
return True
# ── memo requests and free of charge (SO-42, SO-43) ───────────────────────────────────────
def approve_by(self, ids: tuple, email: str, sub: str, now: datetime) -> None:
"""A second person approves what the first asked for: neither the requester's email nor their sign-in subject (gate approver_not_requester)."""
if self.order_type_code not in (*MEMO_TYPES, "FREE_OF_CHARGE") or self.lifecycle != "confirmed":
raise RuleBroken(f"the order is {self.order_type_code} {self.lifecycle}: only a confirmed memo request or free-of-charge order is approved")
if gate_rules._same_person((self.owner, self.owner_sub), ids):
raise GateUnmet([no("approver_not_requester", "the person who asked never approves", "ask another holder of so:memo:approve (another email AND another sign-in)")])
self.approved_by_email, self.approved_by_sub, self.approved_at = email, sub, now
def finish_memo(self) -> bool:
if self.lifecycle == "done":
return False
if self.lifecycle != "confirmed" or self.order_type_code not in MEMO_TYPES:
raise RuleBroken(f"the order is {self.order_type_code} {self.lifecycle}, not a confirmed memo request")
self.lifecycle = "done"
self.events.append(ev.OrderDone(self.sales_order_id))
return True
# ── AI suggestions: audit rows; nothing reaches the order until a person accepts ──────────
def open_suggestion(self, step: str, *, model: str, inputs_ref: str, output: dict) -> Suggestion:
"""Store what a step suggested. An open suggestion of the same step is replaced; an accepted one never is."""
if self.lifecycle != "draft":
raise RuleBroken("a suggestion belongs on a draft")
out = json.dumps(output, sort_keys=True, default=str)
for s in self.suggestions:
if s.step == step and s.decision == "open":
s.model, s.inputs_ref, s.output_json, s.decisions_json = model, inputs_ref, out, "{}"
return s
s = Suggestion(self.sales_order_id, step, max((x.suggestion_no for x in self.suggestions if x.step == step), default=0) + 1,
model, inputs_ref, out)
self.suggestions.append(s)
return s
def resolve_suggestion(self, step: str, suggestion_no: int, decisions: list, *, by: str, now: datetime) -> Resolution:
"""Work out what the person's decisions come to (nothing is written). decisions: [{item, action, ...}]."""
s = next((x for x in self.suggestions if x.step == step and x.suggestion_no == suggestion_no), None)
if s is None:
raise RuleBroken(f"no suggestion {suggestion_no} for {step}")
if s.decision != "open":
raise RuleBroken(f"suggestion {suggestion_no} is already {s.decision}")
valid = set(s.items())
out, done, add, po = s.output, dict(s.decisions), [], None
for d in decisions:
item, action = d["item"], d["action"]
if item not in valid:
raise RuleBroken(f"the suggestion has no item {item}")
if item in done:
raise RuleBroken(f"{item} is already decided")
if action not in ("accept", "change", "reject"):
raise RuleBroken("decide with accept, change or reject")
if action == "change" and not item.startswith("line:"):
raise RuleBroken(f"{item} can only be accepted or rejected")
done[item] = {"action": {"accept": "accepted", "change": "changed", "reject": "rejected"}[action], "by": by, "at": now.isoformat()}
if item.startswith("line:") and action != "reject":
base = out["lines"][int(item[5:])]
add.append({"product_id": d.get("product_id") or base["product_id"], "unit_code": d.get("unit_code") or base.get("unit_code"),
"quantity": D(str(d.get("quantity") if d.get("quantity") is not None else base["quantity"]))})
if item == "po_number" and action == "accept":
po = out["po_number"]
if set(done) >= valid:
acts = {v["action"] for v in done.values()}
decision = "rejected" if acts == {"rejected"} else "changed" if "changed" in acts else "accepted"
else:
decision = "open"
return Resolution(done, add, po, decision)
def apply_suggestion(self, step: str, suggestion_no: int, res: Resolution, specs: list, *, by: str, now: datetime) -> None:
s = next(x for x in self.suggestions if x.step == step and x.suggestion_no == suggestion_no)
for spec in specs:
self.add_line(spec)
if res.po_number:
self.customer_po_ref = res.po_number
s.decisions_json = json.dumps(res.decisions, sort_keys=True)
s.decision = res.decision
if res.decision != "open":
s.decided_by, s.decided_at = by, now
self.events.append(ev.SuggestionDecided(self.sales_order_id, step, suggestion_no, res.decision))