dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/domain/invoice.py
"""The Invoice aggregate: a demand for payment, addressed to a customer. The rules, and nothing else.
No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored.
Four things an accountant checks first, all kept here:
- the invoice keeps ONE immutable id for life (derived from its source); the NUMBER is stamped at issue;
- issuing freezes the number, the dates, the contents, the tax and the bill-to details: an issued invoice refuses
every edit ("credit it instead");
- settlement (open, part paid, paid) is never stored here: payments derives it from its allocations;
- subtotal + tax = total always holds, and the per-rate summary sums to the tax.
"""
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Optional
from brick_ar.domain import events
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.money import ZERO, line_net, money
from brick_ar.domain.readiness import Gate, met, require, unmet
from brick_ar.domain.tax import TaxQuote
from brick_ar.domain.values import Decision, InvoiceKind, InvoiceState, SourceKind
@dataclass
class Line:
"""One billed item, with where it came from (the order line, the schedule, the accepted item, the Odoo line)."""
line_no: int
description: str = ""
quantity: Optional[Decimal] = None # None on a usage line until the biller fills it in
unit_price: Decimal = Decimal(0) # NUMERIC(19,4)
discount_pct: Decimal = Decimal(0)
unit_code: str = "" # the product's (UN/ECE Rec 20)
product_id: str = ""
tax_category: str = "" # what tax decides the code from; the code is tax's answer
tax_code: str = ""
net_amount: Decimal = ZERO
tax_amount: Decimal = ZERO
sales_order_id: str = "" # -> so_line (sales_order_id, so_line_no), when billed from an order
so_line_no: Optional[int] = None
service_from: Optional[date] = None
service_to: Optional[date] = None
schedule_id: str = ""
item_id: str = ""
deducts_invoice_id: str = "" # the deposit a final invoice's negative line uses up
odoo_line_id: str = ""
pinned_from: str = "" # the earlier document whose frozen tax this line takes (a deposit deduction, v4)
interest_on_invoice_id: str = "" # an interest line names the late invoice, the days and the rate (v4)
interest_days: Optional[int] = None
interest_rate_pct: Optional[Decimal] = None
usage: bool = False
cost_centre: str = "" # shared key; gl splits receivable and VAT by it (AR-47, v4)
pricing_quantity: Optional[Decimal] = None # v7: the count unit_price is per (2.4500 per 100 -> 100); None is 1
pricing_unit_code: str = "" # v6: the unit that price is per
quantity_priced: Optional[Decimal] = None # v7: the quantity in pricing_unit_code (exact)
net_given: bool = False # v7: net_amount is the caller's exact net (the order's conditions): kept, never recomputed
def price(self) -> None:
if self.net_given:
return
if self.pricing_unit_code and self.quantity_priced is not None:
self.net_amount = line_net(self.quantity_priced, self.unit_price, self.discount_pct, per=self.pricing_quantity or 1)
else:
self.net_amount = line_net(self.quantity, self.unit_price, self.discount_pct)
@dataclass
class TaxRow:
"""The tax summary the document must print, one row per code: a frozen copy of tax's compute()."""
tax_code: str
rate_pct: Decimal = Decimal(0)
net_amount: Decimal = ZERO
tax_amount: Decimal = ZERO
rounding_method: str = ""
@dataclass
class Send:
sent_at: datetime
channel: str = "email"
address: str = ""
format: str = "pdf"
result: str = "handed to the mail capability"
message_id: str = "" # the access point's id for a Peppol send (AR-43, v4)
statuses_json: str = "" # [{status, at, reason}] as the access point reported them (AR-44, v4)
@dataclass
class Finding:
"""One AI suggestion and the person's decision on it. It never edits the draft."""
step: str
run_no: int
model: str = ""
inputs_digest: str = ""
finding: str = ""
evidence: str = ""
decision: str = Decision.OPEN.value
reason: str = ""
decided_by: str = ""
decided_at: Optional[datetime] = None
@dataclass
class Invoice:
invoice_id: str
bp_id: int = 0
lifecycle: str = InvoiceState.DRAFT.value
kind: str = InvoiceKind.REGULAR.value
source_kind: str = SourceKind.MANUAL.value
source_id: str = ""
invoice_no: str = ""
series_id: str = ""
currency: str = ""
fx_rate: Optional[Decimal] = None
issue_date: Optional[date] = None
tax_point_date: Optional[date] = None
due_date: Optional[date] = None
terms_id: str = ""
customer_po_ref: str = ""
bill_to_json: str = ""
subtotal: Decimal = ZERO
tax_total: Decimal = ZERO
total: Decimal = ZERO
total_functional: Optional[Decimal] = None
export_batch_id: str = ""
migration_run_id: str = ""
company_code: str = "" # the issuing company (v4); "" = the installation's one company
trading_partner: str = "" # the partner's company_code on an intercompany invoice (v4)
settlement_discount_pct: Optional[Decimal] = None # frozen from the terms at issue (v4)
settlement_discount_until: Optional[date] = None
einvoice_message_id: str = "" # the Peppol message of the latest e-invoice send (v4)
sold_to_bp_id: Optional[int] = None # v6 partner functions (SO-40): bp_id stays the payer; the invoice is addressed from bill_to
ship_to_bp_id: Optional[int] = None
bill_to_bp_id: Optional[int] = None
created_by: str = ""
reviewed_by: str = ""
reviewed_at: Optional[datetime] = None
void_reason: str = ""
voided_on: Optional[date] = None
lines: list = field(default_factory=list)
tax: list = field(default_factory=list)
sends: list = field(default_factory=list)
findings: list = field(default_factory=list)
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
# ── creation ──────────────────────────────────────────────────────────────────────────────────
@classmethod
def raise_draft(cls, invoice_id: str, *, by: str, bp_id: int, currency: str, source_kind: str, source_id: str,
lines: list, kind: str = InvoiceKind.REGULAR.value, customer_po_ref: str = "",
tax_point_date: Optional[date] = None, fx_rate: Optional[Decimal] = None,
company_code: str = "", trading_partner: str = "", sold_to_bp_id: Optional[int] = None,
ship_to_bp_id: Optional[int] = None, bill_to_bp_id: Optional[int] = None) -> "Invoice":
"""A draft, from a source or by hand. Line totals are computed here, never typed; tax is applied by the use case
once tax has answered."""
if not currency or len(currency) != 3 or not currency.isalpha():
raise RuleBroken(f"{currency!r} is not a currency code such as GBP")
if kind not in {k.value for k in InvoiceKind}:
raise RuleBroken(f"an invoice is {', '.join(k.value for k in InvoiceKind)}, not {kind}")
if fx_rate is not None:
rate = Decimal(str(fx_rate))
if not rate.is_finite() or rate <= 0 or rate.as_tuple().exponent < -4 or rate >= Decimal("1e15"):
raise RuleBroken("an exchange rate is a number above zero with up to 4 decimals and fewer than 15 whole digits")
fx_rate = rate
inv = cls(invoice_id=invoice_id, bp_id=bp_id, currency=currency.upper(), source_kind=source_kind,
source_id=source_id or invoice_id, kind=kind, customer_po_ref=customer_po_ref,
tax_point_date=tax_point_date, fx_rate=fx_rate, created_by=by,
company_code=company_code, trading_partner=trading_partner, sold_to_bp_id=sold_to_bp_id,
ship_to_bp_id=ship_to_bp_id, bill_to_bp_id=bill_to_bp_id)
inv.lines = list(lines)
inv._renumber_and_price()
inv._record(events.InvoiceRaised(by=by, invoice_id=invoice_id, source_kind=source_kind, source_id=inv.source_id))
return inv
def _renumber_and_price(self) -> None:
seen = set()
for i, ln in enumerate(self.lines, start=1):
ln.line_no = ln.line_no or i
if ln.line_no in seen:
raise RuleBroken(f"line {ln.line_no} appears twice")
seen.add(ln.line_no)
ln.price()
self.subtotal = sum((ln.net_amount for ln in self.lines), ZERO)
self.tax_total = sum((t.tax_amount for t in self.tax), ZERO)
self.total = self.subtotal + self.tax_total
# ── editing: a draft only ───────────────────────────────────────────────────────────────────────
@property
def is_draft(self) -> bool:
return self.lifecycle == InvoiceState.DRAFT.value
def _assert_draft(self) -> None:
if self.lifecycle == InvoiceState.ISSUED.value:
raise RuleBroken(f"invoice {self.invoice_no or self.invoice_id} is issued and cannot be edited: credit it instead")
if self.lifecycle == InvoiceState.VOID.value:
raise RuleBroken(f"invoice {self.invoice_no or self.invoice_id} is void and cannot be edited")
if self.lifecycle == InvoiceState.CANCELLED.value:
raise RuleBroken(f"invoice {self.invoice_id} was cancelled and cannot be edited")
def edit(self, *, by: str, lines: Optional[list] = None, customer_po_ref: Optional[str] = None,
tax_point_date: Optional[date] = None, clear_tax_point: bool = False) -> None:
self._assert_draft()
if lines is not None:
self.lines = list(lines)
self.tax = [] # the tax was computed on other lines: ask tax again
if customer_po_ref is not None:
self.customer_po_ref = customer_po_ref
if tax_point_date is not None or clear_tax_point:
self.tax_point_date = tax_point_date
for ln in self.lines:
ln.tax_amount = ZERO
self._renumber_and_price()
def fill_usage(self, line_no: int, quantity: Decimal, *, by: str) -> None:
self._assert_draft()
ln = next((x for x in self.lines if x.line_no == line_no), None)
if ln is None:
raise RuleBroken(f"invoice {self.invoice_id} has no line {line_no}")
if not ln.usage:
raise RuleBroken(f"line {line_no} is not a usage line: its quantity is fixed by what was sold")
ln.quantity = quantity
self.tax = []
for x in self.lines:
x.tax_amount = ZERO
self._renumber_and_price()
def apply_tax(self, quote: TaxQuote) -> None:
"""Copy tax's answer onto the document. The invariants hold or nothing is stored."""
by_line = {t.line_no: t for t in quote.lines}
for ln in self.lines:
ln.price()
if ln.net_amount == ZERO and ln.quantity is None:
ln.tax_amount, ln.tax_code = ZERO, ln.tax_code
continue
t = by_line.get(ln.line_no)
if t is None:
raise RuleBroken(f"tax did not price line {ln.line_no}")
ln.tax_code, ln.tax_amount = t.tax_code, money(t.tax_amount)
rows = [TaxRow(r.tax_code, r.rate_pct, money(r.net_amount), money(r.tax_amount), quote.rounding_method)
for r in quote.summary]
net = sum((r.net_amount for r in rows), ZERO)
if rows and net != sum((ln.net_amount for ln in self.lines), ZERO):
raise RuleBroken("tax's summary does not cover the invoice's lines: its net amounts differ from the lines'")
self.tax = rows
self.subtotal = sum((ln.net_amount for ln in self.lines), ZERO)
self.tax_total = sum((r.tax_amount for r in rows), ZERO)
self.total = self.subtotal + self.tax_total
if self.fx_rate:
self.total_functional = money(self.total * Decimal(self.fx_rate))
# ── readiness ────────────────────────────────────────────────────────────────────────────────
def local_gates(self) -> list:
"""The issue gates this document answers by itself (the others belong to the customer, tax and the ledger)."""
gates = [met("has_lines") if self.lines else
unmet("has_lines", "the invoice has no lines", "add a line")]
blanks = [str(ln.line_no) for ln in self.lines if ln.usage and ln.quantity is None]
gates.append(met("usage_filled") if not blanks else
unmet("usage_filled", f"a usage line has no quantity (line {', '.join(blanks)})",
"enter the period's usage (AR-19)"))
return gates
# ── transitions ──────────────────────────────────────────────────────────────────────────────
def cancel(self, *, by: str) -> None:
"""A draft that should never have existed. No number is used."""
if self.lifecycle == InvoiceState.CANCELLED.value:
return
self._assert_draft()
self.lifecycle = InvoiceState.CANCELLED.value
self._record(events.InvoiceCancelled(by=by, invoice_id=self.invoice_id))
def review(self, *, by: str, at: datetime) -> None:
self._assert_draft()
self.reviewed_by, self.reviewed_at = by, at
@property
def settlement_discount_amount(self) -> Optional[Decimal]:
"""The amount that settles the invoice if paid by settlement_discount_until: total less the terms' percentage of the gross."""
if not self.settlement_discount_pct:
return None
return money(self.total - money(self.total * Decimal(self.settlement_discount_pct) / 100))
def issue(self, *, today: date, number: str, series_id: str, due_date: date, bill_to_json: str, terms_id: str,
quote: TaxQuote, by: str, discount_pct: Optional[Decimal] = None, discount_days: Optional[int] = None) -> None:
"""Fix the number, the dates, the contents, the tax and the bill-to details. The issue date is today in the
business time zone; it is never typed. The tax point is the supply date the source recorded, else today."""
self._assert_draft()
require("issue the invoice", self.local_gates())
self.apply_tax(quote)
self.invoice_no, self.series_id = number, series_id
self.issue_date = today
self.tax_point_date = self.tax_point_date or today
self.due_date, self.terms_id, self.bill_to_json = due_date, terms_id, bill_to_json
if discount_pct and discount_days is not None: # frozen from the terms (AR-37)
self.settlement_discount_pct = Decimal(discount_pct)
self.settlement_discount_until = today + timedelta(days=int(discount_days))
self.lifecycle = InvoiceState.ISSUED.value
self._record(events.InvoiceIssued(by=by, invoice_id=self.invoice_id, invoice_no=number, total=self.total))
def void(self, *, today: date, reason: str, by: str) -> None:
"""The number is kept; the invoice drops out of every balance. Whether it MAY be voided (no allocations, the
owner's void_policy) is asked of the use case's gates before this runs."""
if self.lifecycle != InvoiceState.ISSUED.value:
raise RuleBroken(f"only an issued invoice can be voided; this one is {self.lifecycle}")
if not (reason or "").strip():
raise RuleBroken("a void needs a reason")
self.lifecycle = InvoiceState.VOID.value
self.void_reason, self.voided_on = reason.strip(), today
self._record(events.InvoiceVoided(by=by, invoice_id=self.invoice_id, reason=self.void_reason))
def record_send(self, *, at: datetime, address: str, channel: str, by: str) -> Send:
"""Re-sending records a second send, never a second invoice."""
if self.lifecycle != InvoiceState.ISSUED.value:
raise RuleBroken(f"only an issued invoice is sent; this one is {self.lifecycle}")
if self.source_kind == SourceKind.ODOO.value:
raise RuleBroken("a migrated invoice is history: it was sent by Odoo and is never sent again")
s = Send(sent_at=at, channel=channel, address=address)
self.sends.append(s)
self._record(events.InvoiceSent(by=by, invoice_id=self.invoice_id, address=address))
return s
def mark_exported(self, batch_id: str) -> None:
if self.export_batch_id:
raise RuleBroken(f"already in batch {self.export_batch_id}: it is exported once")
self.export_batch_id = batch_id
# ── AI findings: written open by the system, decided only by a person ───────────────────────────
def add_finding(self, step: str, model: str, digest: str, finding: str, evidence: str) -> Finding:
self._assert_draft()
run_no = 1 + max((f.run_no for f in self.findings if f.step == step), default=0)
f = Finding(step=step, run_no=run_no, model=model, inputs_digest=digest, finding=finding, evidence=evidence)
self.findings.append(f)
return f
def decide_finding(self, step: str, run_no: int, decision: str, reason: str, *, by: str, at: datetime) -> Finding:
"""Accept or dismiss. Nothing in the draft changes; issue never waits on a finding."""
f = next((x for x in self.findings if x.step == step and x.run_no == run_no), None)
if f is None:
raise RuleBroken(f"invoice {self.invoice_id} has no finding {step} #{run_no}")
if decision not in (Decision.ACCEPTED.value, Decision.DISMISSED.value):
raise RuleBroken("a finding is accepted or dismissed")
if decision == Decision.DISMISSED.value and not (reason or "").strip():
raise RuleBroken("dismissing a finding needs a reason")
if f.decision != Decision.OPEN.value:
raise RuleBroken(f"finding {step} #{run_no} was already {f.decision} by {f.decided_by}")
f.decision, f.reason, f.decided_by, f.decided_at = decision, (reason or "").strip(), by, at
self._record(events.FindingDecided(by=by, invoice_id=self.invoice_id, step=step, decision=decision))
return f
def deposit_deduction_lines(deposit: "Invoice", first_line_no: int, amount: Optional[Decimal] = None) -> list:
"""The negative lines a final invoice carries for a deposit already invoiced: one per tax code, each PINNED to the deposit's
frozen code and rate (tax TAX-47), so the VAT falls only on the balance (SC-AR-01). `amount` is the net taken: the whole
deposit when left out, or a part (split over the codes pro rata, the last code taking the remainder). The deposit must
be issued."""
if deposit.lifecycle != InvoiceState.ISSUED.value:
raise RuleBroken(f"deposit invoice {deposit.invoice_id} is {deposit.lifecycle}: only an issued deposit is deducted")
if deposit.kind != InvoiceKind.DEPOSIT.value:
raise RuleBroken(f"invoice {deposit.invoice_id} is not a deposit invoice")
take = money(deposit.subtotal if amount is None else amount)
if take <= 0:
raise RuleBroken("a deposit deduction is an amount above zero")
rows = sorted(deposit.tax, key=lambda r: r.tax_code)
out, given = [], ZERO
for i, row in enumerate(rows):
net = take - given if i == len(rows) - 1 else money(take * row.net_amount / deposit.subtotal)
given += net
out.append(Line(line_no=first_line_no + i, description=f"Less deposit {deposit.invoice_no}", quantity=Decimal(1),
unit_price=-net, tax_code=row.tax_code, deducts_invoice_id=deposit.invoice_id,
pinned_from=deposit.invoice_id))
if not out: # a deposit with no tax rows: deduct its net as one line
out.append(Line(line_no=first_line_no, description=f"Less deposit {deposit.invoice_no}", quantity=Decimal(1),
unit_price=-take, deducts_invoice_id=deposit.invoice_id))
return out
def gate_for(met_: bool, gate: str, reason: str, fix: str) -> Gate:
return met(gate) if met_ else unmet(gate, reason, fix)