dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/bills.py
"""Bills: the supplier's claim on us, matched against what we ordered and received, approved, posted (PUR-10 to PUR-12, PUR-14, PUR-17, PUR-18,
PUR-20, PUR-27, PUR-28, PUR-36, PUR-52, PUR-56, PUR-63, PUR-64).
The supplier's invoice number is THEIRS: the natural key is (bp_id, supplier_invoice_no, invoice_date), so recording the same invoice again returns the
existing bill. The match is DERIVED from order, receipt and bill rows each time it is read, within the tolerance in force on the bill's date for the
supplier's tolerance group; a failing line is accepted by a SECOND person with a reason. Approving claims the billed quantity per order line with a plain
INSERT, so two bills approved at once cannot both claim the same received goods."""
import json
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import BigId, IsoDate, Limit, text
from brick_purchasing.application import base, rules
from brick_purchasing.application.base import (
AI_CHECK,
BILL_APPROVE,
BILL_POST,
BILL_READ,
BILL_WRITE,
PURCHASING,
_decide,
_same_version,
_stamp,
_sub,
_today,
_who,
stamp_number,
unit,
view,
)
from brick_purchasing.application.inputs import AddressIn, BillLineIn, Code, Currency, Money, Offset, Reason, Ref, Version, fit_money
from brick_purchasing.application.master import profile_on
from brick_purchasing.application.receipts import _unit_factor, planned_costs
from brick_purchasing.domain import bill as bl
from brick_purchasing.domain import matching
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.payment import Allocation
from brick_purchasing.domain.readiness import no, ok, require, views
from brick_purchasing.domain.values import ZERO, add_months, as_dec, base_of, fr, money_round
from brick_purchasing.infrastructure import ai_steps
from brick_purchasing.infrastructure.seams import Missing, ports
repository = lambda: base.repository() # noqa: E731
def _load(repo, caller: dict, bill_id: str) -> tuple:
found = repo.bills.get(bill_id)
if found is None or not (auth.has_permission(caller, BILL_READ) or auth.has_permission(caller, BILL_WRITE)):
raise Refused(404, f"no bill {bill_id}")
return found
# ═══ outstanding, derived ═════════════════════════════════════════════════════════════════════════
def outstanding(repo, b: bl.SupplierBill, as_of: Optional[date] = None) -> Decimal:
"""total minus allocations, always: a bill's by what settled it (payments, credits, down payments); a credit's by what it settled."""
if b.kind == "credit":
used = sum((a.amount for a in repo.allocations.of_source("credit", b.bill_id) if as_of is None or a.allocated_on <= as_of), ZERO)
return b.total - used
return b.total - repo.allocations.sums([b.bill_id], as_of).get(b.bill_id, ZERO)
# ═══ the match, derived ═════════════════════════════════════════════════════════════════════════════
def default_tolerances(s: base.Settings) -> dict:
return {"PP": matching.Tol(upper_pct=s["price_tolerance_pct"], upper_amount=s["price_tolerance_amount"] or None),
"DQ": matching.Tol(upper_pct=s["qty_tolerance_pct"]), "TX": matching.Tol(upper_amount=s["tax_tolerance_amount"]),
"AN": matching.Tol(), "BD": matching.Tol(), "KW": matching.Tol()}
def tolerances_for(repo, bp_id: int, on: date) -> tuple:
prof = profile_on(repo, bp_id, on)
return matching.tolerance_set(repo.tolerances.all(), prof.get("tolerance_group") or "STANDARD", on, default_tolerances(base.Settings(repo, on))), prof
def bill_match(repo, b: bl.SupplierBill, order=None) -> dict:
"""The match of a bill, derived from rows now: per line price, quantity and tax (and amount, delivery cost) with the difference as an exact
fraction, the tolerance in force on the invoice date and its key. Nothing is stored."""
on = b.invoice_date
tol, prof = tolerances_for(repo, b.bp_id, on)
s = base.Settings(repo, on)
o = order
if o is None and b.purchase_order_id:
got = repo.orders.get(b.purchase_order_id)
o = got[0] if got else None
kind = "credit" if b.kind == "credit" else "bill"
recv = rules.received_by_line(repo, b.purchase_order_id) if o else {}
returned = rules.returned_by_line(repo, b.purchase_order_id) if o else {}
claimed = repo.bills.claimed_by_line(b.purchase_order_id, kind, exclude_bill=b.bill_id) if o else {}
results = []
for ln in ([] if (b.kind == "down_payment_invoice" or b.source_kind == "intercompany") else b.lines): # an intercompany bill IS ar's lines: two-way equal by construction # a VAT invoice for a down payment is checked against the down payment, not the order's lines
x = o.line(ln.order_line_no) if (o and ln.order_line_no) else None
if ln.withdrawal_id:
amount = None
w = repo.withdrawals.get(ln.withdrawal_id)
if w and (w["status"] == "open" or w.get("settled_bill_id") == b.bill_id): # a settled withdrawal is billed once
amount = fr(w["amount"])
results.append(matching.match_withdrawals(ln, amount))
elif ln.delivery_cost_ref and o:
planned = sum((fr(a) for xl in o.lines for k, a in planned_costs(o, xl, Fraction(1)) if k == ln.delivery_cost_ref), Fraction(0)) or None
results.append(matching.match_delivery_cost(ln, planned, tol))
elif x is None:
results.append(matching.match_unordered(ln, tol))
else:
if b.kind == "credit":
basis = returned.get(x.line_no, Fraction(0))
else:
mode = s["match_mode_goods"] if x.kind == "goods" else s["match_mode_services"]
basis = recv.get(x.line_no, Fraction(0)) if mode == "three_way" else x.exact_base
results.append(matching.match_line(ln, x, basis_base=basis, claimed_other=claimed.get(x.line_no, Fraction(0)), tol=tol,
credit=b.kind == "credit"))
tx = matching.match_tax(b.supplier_tax_total, b.tax_total, tol) if b.tax_frozen is False or True else None
if tx:
results.append(tx)
small = matching.small_difference(results, tol)
fails = matching.failures(results)
pending = []
for ln_no, kind_, key in fails:
if b.accepted(ln_no, kind_):
continue
if small is not None and kind_ == "price":
continue
pending.append({"line_no": ln_no, "kind": kind_, "key": key})
return {"lines": results, "failures": [{"line_no": a, "kind": k, "key": ky} for a, k, ky in fails], "unaccepted": pending,
"passed": not fails, "approvable": not pending, "small_difference": None if small is None else str(as_dec(small, 2)),
"tolerance_group": prof.get("tolerance_group") or "STANDARD", "as_of": on}
# ═══ the view ═════════════════════════════════════════════════════════════════════════════════════
def bill_view(repo, caller: dict, b: bl.SupplierBill, version: int, *, today: Optional[date] = None, match: bool = True) -> dict:
out = view(b, drop=("claims",))
t = today or _today(repo)
out["version"] = version
out["outstanding"] = outstanding(repo, b)
if b.due_date:
out["days_to_due"] = (b.due_date - t).days
if match and b.status in ("recorded", "disputed", "approved") and b.source_kind != "odoo":
m = bill_match(repo, b)
out["match"] = {**m, "lines": matching.view(m["lines"])}
if b.status == "recorded":
out["readiness"] = views(approve_gates(repo, b, t, local_only=True))
out["claims"] = [view(c) for c in b.claims]
out["findings"] = repo.findings.all(subject_ref=f"pur_bill:{b.bill_id}")
return out
# ═══ recording ═══════════════════════════════════════════════════════════════════════════════════
def terms_due(repo, terms_id: str, invoice_date: date) -> date:
t = repo.terms.get(terms_id or "NET30") or {"days": 30, "end_of_month": False}
if t.get("end_of_month"):
eom = add_months(invoice_date.replace(day=1), 1)
eom = date.fromordinal(eom.toordinal() - 1)
return date.fromordinal(eom.toordinal() + int(t.get("days") or 0))
return date.fromordinal(invoice_date.toordinal() + int(t.get("days") or 30))
def _tax_lines(b: bl.SupplierBill, o) -> list:
return [{"line_no": ln.line_no, "description": ln.description, "category_code": ln.tax_category or None,
"supply": "services" if (ln.order_line_no and o and o.line(ln.order_line_no) and o.line(ln.order_line_no).kind == "service") else "goods",
"amount": str(ln.net_amount), "quantity": str(ln.quantity), "unit_price": str(ln.unit_price)} for ln in b.lines]
def compute_tax(repo, caller: dict, b: bl.SupplierBill, o, reclaims: bool) -> None:
"""Ask the tax module for the VAT on the bill (direction purchase): per line the code and rate, and the amount; the supplier's stated VAT
stays beside it. Without the tax module the supplier's stated VAT is the bill's tax."""
if not ports.installed("tax"):
b.tax_total = b.supplier_tax_total or ZERO
b.total = b.subtotal + b.tax_total
return
body = {"document_ref": f"pur_bill:{b.bill_id}", "direction": "purchase", "tax_point": b.invoice_date.isoformat(), "bp_id": str(b.bp_id),
"currency": b.currency, "self_billed": bool(b.self_billed), "lines": _tax_lines(b, o)}
try:
got = ports.tax.compute(body)
except Missing as m:
raise base._refuse(m, "tax") from None
by = {int(x["line_no"]): x for x in got.get("lines") or []}
for ln in b.lines:
x = by.get(ln.line_no)
if x:
ln.tax_code, ln.tax_rate = x.get("tax_code") or "", Decimal(str(x.get("rate_pct") or 0))
ln.tax_amount = Decimal(str(x.get("vat") or 0))
ln.vat_reclaimable = Decimal(str(x.get("vat_reclaimable", ln.tax_amount))) if reclaims else ZERO
b.tax_ref = body["document_ref"]
b.recompute_totals()
def build_bill_lines(repo, caller: dict, o, rows: list, bill_id: str, bp_id: int, currency: str) -> list:
out = []
for n, x in enumerate(rows, 1):
ol = o.line(x.order_line_no) if (o and x.order_line_no) else None
if x.order_line_no and o and ol is None:
raise Refused(422, f"line {n}: the order has no line {x.order_line_no}")
if ol is not None and ol.item_category == "consignment" and not x.withdrawal_id:
raise Refused(422, f"line {n}: a consignment line is billed through its withdrawals (withdrawal_id), never against the receipt")
unit_code = x.unit_code or (ol.pricing_unit_code if ol else "")
if not unit_code:
raise Refused(422, f"line {n}: name the unit the supplier billed in")
pu = x.pricing_unit_code or unit_code
if ol is not None:
nu, de = _unit_factor(caller, o, ol, unit_code)
pn, pd = _unit_factor(caller, o, ol, pu)
else:
nu = de = pn = pd = 1
pq = x.pricing_quantity or 1
qb = base_of(x.quantity, nu, de)
qp = qb * Fraction(pd, pn)
net = x.net_amount if x.net_amount is not None else money_round(qp * fr(x.unit_price) / pq, currency)
fit_money(net, f"line {n}'s net")
out.append(bl.BillLine(bill_id, n, x.order_line_no, x.product_id or (ol.product_id if ol else ""), x.description or (ol.description if ol else ""),
x.quantity, unit_code, nu, de, as_dec(qb), pu, pn, pd, pq, x.unit_price, net,
x.tax_category or (ol.tax_category if ol else ""), x.tax_code or "", ZERO, ZERO, ZERO,
x.expense_category or (ol.expense_category if ol else "general"), x.cis_part or "none", x.withdrawal_id or "",
x.delivery_cost_ref or "", "", ol.asset_id if ol else "", ol.cost_centre if ol else "", ol.project_id if ol else "",
ol.account_assignment if ol else "", ol.item_category if ol else "", "",
_posting_kind(ol, x)))
return out
def _posting_kind(ol, x) -> str:
if x.withdrawal_id:
return "consignment"
if x.delivery_cost_ref:
return "delivery_cost"
if ol is not None and ol.account_assignment == "asset":
return "asset"
return "expense"
def _party_vat(bp_id: int) -> str:
"""v8 (PUR-28): the supplier's VAT number as tax holds it, the default for what the bill shows (the clerk edits it to what the paper says)."""
if ports.installed("tax"):
try:
return str((ports.tax.party(int(bp_id)).get("current") or {}).get("vat_number") or "")
except Missing:
return ""
return ""
def record_core(repo, caller: dict, *, bp_id: int, supplier_invoice_no: str, invoice_date: date, currency: str, lines: list, purchase_order_id: str = "",
due_date: Optional[date] = None, kind: str = "bill", supplier_tax_total=None, corrects_bill_id: str = "", company_code: str = "",
source_kind: str = "person", source_id: str = "", self_billed: bool = False, partner_company: str = "", ic_reference: str = "",
down_payment_id: str = "", recorded_by: Optional[tuple] = None, tax_lines_prebuilt: bool = False,
supplier_vat_number: str = "", supplier_address: Optional[dict] = None):
"""Record a bill: the one place a bill is built (a person's record_bill, an intercompany bill, an ERS invoice). Returns (bill, created)."""
today = _today(repo)
bid = bl.bill_id_of(bp_id, supplier_invoice_no, invoice_date)
have = repo.bills.get(bid)
if have:
return have[0], have[1], False
if invoice_date > today:
raise RuleBroken(f"not_future: a supplier's invoice cannot be dated after today ({today})")
o = None
if purchase_order_id:
g = repo.orders.get(purchase_order_id)
if g is None:
raise Refused(422, f"there is no order {purchase_order_id}")
o = g[0]
if o.status == "cancelled":
raise RuleBroken("order_open: the order it names is cancelled; fix: ask the supplier, or record it without an order and dispute it")
if o.bp_id != int(bp_id) and int(bp_id) not in {c.payee_bp_id for c in o.delivery_costs if c.payee_bp_id}:
raise Refused(422, f"order {purchase_order_id} is with supplier {o.bp_id}, not {bp_id}")
prof = profile_on(repo, bp_id, invoice_date)
email, sub = recorded_by or (_who(caller), _sub(caller))
b = bl.SupplierBill(bid, int(bp_id), supplier_invoice_no.strip(), invoice_date, kind, self_billed, corrects_bill_id, purchase_order_id, company_code or (o.company_code if o else ""),
currency, ZERO, ZERO, ZERO, supplier_tax_total, due_date or terms_due(repo, prof["terms_id"], invoice_date), prof["terms_id"],
"recorded", "", email, sub, today, source_kind=source_kind, source_id=source_id, partner_company=partner_company,
ic_reference=ic_reference, down_payment_id=down_payment_id)
b.supplier_vat_number = (supplier_vat_number or "").strip() or _party_vat(bp_id)
b.supplier_address = json.dumps(supplier_address) if supplier_address else ""
b.lines = lines
for ln in b.lines:
if ic_reference:
ln.ic_reference = ic_reference
b.recompute_totals()
if not tax_lines_prebuilt:
compute_tax(repo, caller, b, o, bool(base.Settings(repo, invoice_date)["reclaim_vat"] and prof.get("reclaims_vat", True)))
return b, 0, True
@PURCHASING.use_case(kind="command", route="POST /bills", permission=BILL_WRITE, idempotent=True)
def record_bill(caller: dict, bp_id: BigId, supplier_invoice_no: text(100, min_length=1), invoice_date: IsoDate, currency: Currency, lines: list[BillLineIn],
purchase_order_id: Optional[Ref] = None, due_date: Optional[IsoDate] = None, kind: str = "bill", supplier_tax_total: Optional[Money] = None,
corrects_bill_id: Optional[Ref] = None, company_code: Optional[text(4)] = None, supplier_vat_number: Optional[text(40)] = None,
supplier_address: Optional[AddressIn] = None) -> dict:
"""Record a supplier's invoice (or credit note, kind credit) as what it is: THEIR claim. It is keyed on (supplier, their number, their date), so
recording it again returns the existing bill. The supplier's VAT number (defaulted from tax) and postal address are kept as the paper shows
them: tax's content check reads them at approval. Lines name the order lines they bill (none: a line with no order, which the match flags). The VAT is
asked of the tax module (direction purchase) and the supplier's stated VAT is kept beside it. Returns the bill with its derived match."""
if kind not in ("bill", "credit", "down_payment_invoice"):
raise Refused(422, "kind is bill, credit or down_payment_invoice")
if not lines:
raise Refused(422, "a bill has at least one line")
cur = base._currency(currency)
def run():
with repository() as repo, _decide():
idate = base._day(invoice_date, "invoice_date")
o = None
if purchase_order_id:
g = repo.orders.get(purchase_order_id)
o = g[0] if g else None
built = build_bill_lines(repo, caller, o, lines, bl.bill_id_of(bp_id, supplier_invoice_no, idate), bp_id, cur)
b, v, created = record_core(repo, caller, bp_id=bp_id, supplier_invoice_no=supplier_invoice_no, invoice_date=idate, currency=cur, lines=built,
purchase_order_id=purchase_order_id or "", due_date=base._day(due_date, "due_date"), kind=kind,
supplier_tax_total=supplier_tax_total, corrects_bill_id=corrects_bill_id or "", company_code=company_code or "",
supplier_vat_number=supplier_vat_number or "",
supplier_address=supplier_address.model_dump(exclude_none=True) if supplier_address else None)
if created:
v = repo.bills.save(b, 0)
_ai_check(repo, b)
return bill_view(repo, caller, b, v)
return unit(None, run)
def _ai_check(repo, b: bl.SupplierBill) -> None:
"""The baseline duplicate check (always shown); findings are suggestions for the clerk, never a block."""
today = b.invoice_date
if base.Settings(repo, _today(repo)).mode("ai_check_bill") == "off":
return
others = [{"bill_id": x.bill_id, "supplier_invoice_no": x.supplier_invoice_no, "total": x.total, "invoice_date": x.invoice_date}
for x in repo.bills.repo.all_heads() if x.bp_id == b.bp_id]
found = ai_steps.likely_duplicates({"bill_id": b.bill_id, "supplier_invoice_no": b.supplier_invoice_no, "total": b.total, "invoice_date": today}, others)
for n, f in enumerate(found, 1):
repo.findings.put({"subject_ref": f"pur_bill:{b.bill_id}", "step": "ai_check_bill", "run_no": 1, "finding_no": n, "model": "baseline",
"inputs_digest": "", "finding": json.dumps(f), "evidence": f["why"]})
# ═══ accepting a difference, disputing, cancelling ═════════════════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/accept-variance", permission=BILL_APPROVE)
def accept_variance(caller: dict, bill_id: Ref, line_no: int, kind: str, reason: Reason, version: Optional[Version] = None) -> dict:
"""Accept a match difference on a bill line (kind price, quantity, tax, amount or delivery_cost) with a reason, so a real price rise is not a
permanent blockage. The approver is not the clerk who recorded the bill (same email OR same sub). The order is not rewritten: its line still
says what was ordered."""
def run():
with repository() as repo, _decide():
b, v = _load(repo, caller, bill_id)
_same_version("the bill", v, version)
m = bill_match(repo, b)
hit = next((c for r in m["lines"] if r["line_no"] == line_no for c in r["checks"] if c["kind"] == kind and not c["passed"]), None)
if hit is None:
raise RuleBroken(f"line {line_no} does not fail on {kind}: there is nothing to accept")
b.accept_variance(line_no, kind, email=caller.get("email"), sub=caller.get("sub"), reason=reason, today=_today(repo),
difference=hit["difference"], tolerance=json.dumps(hit["tolerance"]), key=hit["key"])
return bill_view(repo, caller, b, repo.bills.save(b, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/dispute", permission=BILL_WRITE)
def dispute_bill(caller: dict, bill_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Put a bill on hold as disputed, with a reason: it is held out of every payment run until the dispute is resolved. Who raised it is recorded."""
def run():
with repository() as repo, _decide():
b, v = _load(repo, caller, bill_id)
_same_version("the bill", v, version)
b.dispute(reason, _who(caller))
return bill_view(repo, caller, b, repo.bills.save(b, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/resolve-dispute", permission=BILL_WRITE)
def resolve_bill_dispute(caller: dict, bill_id: Ref, outcome: Reason, version: Optional[Version] = None) -> dict:
"""Resolve a dispute with its outcome: the bill returns to what it was (recorded, or approved if it had been)."""
def run():
with repository() as repo, _decide():
b, v = _load(repo, caller, bill_id)
_same_version("the bill", v, version)
b.resolve(outcome, was_approved=bool(b.bill_no))
return bill_view(repo, caller, b, repo.bills.save(b, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/cancel", permission=BILL_APPROVE, idempotent=True, destructive=True)
def cancel_bill(caller: dict, bill_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Cancel a bill that was never approved and has nothing allocated, with a reason."""
def run():
with repository() as repo, _decide():
b, v = _load(repo, caller, bill_id)
if b.status == "cancelled":
return bill_view(repo, caller, b, v, match=False)
_same_version("the bill", v, version)
b.cancel(reason, bool(repo.allocations.of_bill(b.bill_id)))
return bill_view(repo, caller, b, repo.bills.save(b, v), match=False)
return unit(version, run)
# ═══ approving ═════════════════════════════════════════════════════════════════════════════════════
def _company_name(b: bl.SupplierBill) -> str:
"""Our name as the customer on the supplier's invoice: gl's company (the bill's, else the only company the caller's ledger has)."""
if not ports.installed("gl"):
return ""
try:
if b.company_code:
return str(ports.gl.company(b.company_code).get("name") or "")
rows = ports.gl.companies()
return str(rows[0].get("name") or "") if len(rows) == 1 else ""
except Missing:
return ""
def _supplier_name(repo, b: bl.SupplierBill, caller) -> str:
"""The supplier as the paper names it: purchasing's profile, else the partner record in business-partner."""
name = profile_on(repo, b.bp_id, b.invoice_date).get("legal_name")
if name or not ports.installed("bp"):
return name or ""
try:
return str(ports.bp.get(caller, b.bp_id).get("legal_name") or "")
except Missing:
return ""
def _contents(repo, b: bl.SupplierBill, caller=None) -> dict:
"""What tax's check-contents and freeze take (TAX-24, v8): the bill's particulars as the supplier printed them. An unknown VAT number or
address is sent ABSENT, never empty, so tax names its own gate."""
out = {"invoice_number": b.supplier_invoice_no, "issue_date": b.invoice_date.isoformat(),
"supplier_name": _supplier_name(repo, b, caller)}
if b.supplier_address:
out["supplier_address"] = json.loads(b.supplier_address)
if b.supplier_vat_number:
out["supplier_vat_number"] = b.supplier_vat_number
name = _company_name(b)
if name:
out["customer_name"] = name
return out
def _cost_centre_gate(b: bl.SupplierBill):
"""v8 gate cost_centre_open: every cost centre on the bill's lines is known to gl, of the bill's company and open on the invoice date."""
bad = []
for cc in sorted({x.cost_centre for x in b.lines if x.cost_centre}):
try:
r = ports.gl.cost_centre(cc)
except Missing as mm:
if mm.status == 404:
bad.append(f"{cc} is unknown to the ledger")
continue
return no("cost_centre_open", f"the ledger: {mm.reason}", "try again")
if r.get("status") == "closed":
bad.append(f"{cc} is closed on {b.invoice_date}")
elif b.company_code and r.get("company_code") and r["company_code"] != b.company_code:
bad.append(f"{cc} belongs to company {r['company_code']}, not {b.company_code}")
if not bad:
return ok("cost_centre_open")
lines = {x.cost_centre: x.line_no for x in b.lines if x.cost_centre}
return no("cost_centre_open", "; ".join(f"line {lines[c.split()[0]]}'s cost centre {c}" for c in bad), "pick an open cost centre (gl), or ask the controller to open it")
def approve_gates(repo, b: bl.SupplierBill, today: date, *, local_only: bool = False, caller=None) -> list:
"""Every gate of approve_bill as {gate, met, reason, fix}: the match (every failure accepted), not over-billed (what is already claimed plus
this bill fits under what was received), tax content (tax's own answer), the period open (the ledger's own answer), the account map."""
gates = []
m = bill_match(repo, b)
gates.append(ok("match") if m["approvable"] else
no("match", "a line fails " + ", ".join(sorted({f"{p['kind']} ({p['key']}) on line {p['line_no']}" for p in m["unaccepted"]})) + " and no acceptance is recorded",
"accept the difference with a reason (PUR-12), or dispute it"))
over = _over_billed(repo, b)
gates.append(ok("not_over_billed") if not over else no("not_over_billed", over, "wait for the receipt, or dispute"))
if local_only:
return gates
if ports.installed("tax") and b.tax_ref and b.kind != "credit":
try:
r = ports.tax.check_contents(b.tax_ref, _contents(repo, b, caller))
bad = [g for g in r.get("gates") or [] if not g.get("met")]
gates.append(ok("tax_content") if not bad else no("tax_content", "; ".join(g.get("reason", "") for g in bad),
"ask the supplier for a valid VAT invoice, or record the VAT as not reclaimable"))
except Missing as mm:
gates.append(no("tax_content", f"tax: {mm.reason}", "try again"))
if ports.installed("gl"):
try:
p = ports.gl.period_open(b.invoice_date, b.company_code or None)
gates.append(ok("period_open") if p.get("open", True) else no("period_open", p.get("reason") or "the invoice date falls in a closed ledger period",
p.get("fix") or "ask the accountant to reopen, or post in the next open period"))
except Missing as mm:
gates.append(no("period_open", f"the ledger: {mm.reason}", "try again"))
gates.append(_cost_centre_gate(b))
missing = missing_accounts(repo, b)
gates.append(ok("account_map") if not missing else no("account_map", "no account in force on the invoice date for " + ", ".join(missing), "Settings: account map"))
return gates
def _over_billed(repo, b: bl.SupplierBill) -> str:
if not b.purchase_order_id or b.kind == "down_payment_invoice":
return ""
got = repo.orders.get(b.purchase_order_id)
if not got:
return ""
o = got[0]
s = base.Settings(repo, b.invoice_date)
kind = "credit" if b.kind == "credit" else "bill"
claimed = repo.bills.claimed_by_line(b.purchase_order_id, kind, exclude_bill=b.bill_id)
recv = rules.received_by_line(repo, b.purchase_order_id) if kind == "bill" else rules.returned_by_line(repo, b.purchase_order_id)
this = {}
for ln in b.lines:
if ln.order_line_no and not ln.withdrawal_id and not ln.delivery_cost_ref:
this[ln.order_line_no] = this.get(ln.order_line_no, Fraction(0)) + base_of(ln.quantity, ln.unit_numerator, ln.unit_denominator)
for no_, q in this.items():
x = o.line(no_)
if x is None:
continue
mode = s["match_mode_goods"] if x.kind == "goods" else s["match_mode_services"]
basis = recv.get(no_, Fraction(0)) if (kind == "credit" or mode == "three_way") else x.exact_base
tol = tolerances_for(repo, b.bp_id, b.invoice_date)[0].get("DQ", matching.Tol())
cap = basis * (1 + (fr(tol.upper_pct) / 100 if tol.upper_pct else 0))
if claimed.get(no_, Fraction(0)) + q > cap:
return (f"line {no_}: {as_dec(claimed.get(no_, Fraction(0)))} of {as_dec(basis)} is already claimed by another bill; this bill's {as_dec(q)} does "
"not fit (the order line has already been billed for what was received)")
return ""
def missing_accounts(repo, b: bl.SupplierBill) -> list:
need = {("supplier_down_payment", "*")} if b.kind == "down_payment_invoice" else {("payables", "*")}
for ln in b.lines:
kind = ln.posting_kind
if kind == "expense" and b.kind != "down_payment_invoice":
need.add(("expense", ln.expense_category or "general"))
if ln.tax_amount:
need.add(("input_vat", ln.tax_code or "*"))
out = []
for purpose, cat in sorted(need):
if account_for(repo, purpose, cat, b.invoice_date) is None:
out.append(f"{purpose}{'' if cat == '*' else ' ' + cat}")
return out
def account_for(repo, purpose: str, category: str, on: date) -> Optional[str]:
"""The ledger account in force on a date for a purpose and category (else the every-category row)."""
for cat in (category, "*"):
rows = [r for r in repo.account_map.all(purpose=purpose, category=cat) if r["valid_from"] <= on]
if rows:
return max(rows, key=lambda r: r["valid_from"])["account_id"]
return None
def approve_core(repo, caller: dict, b: bl.SupplierBill, v: int, version: Optional[int] = None):
"""The approval itself, inside the caller's transaction (a person's approve_bill, and the ERS run which approves the invoice it just raised)."""
if b.status == "approved":
return bill_view(repo, caller, b, v)
if b.status != "recorded":
raise RuleBroken(f"the bill is {b.status}: only a recorded bill is approved")
_same_version("the bill", v, version)
today = _today(repo)
gates = approve_gates(repo, b, today, caller=caller)
require(gates)
# small differences (BD): the system accepts them, recorded
m = bill_match(repo, b)
if m["small_difference"] is not None:
for p in m["failures"]:
if p["kind"] == "price" and not b.accepted(p["line_no"], "small_difference") and not b.accepted(p["line_no"], "price"):
b.accept_variance(p["line_no"], "small_difference", email="system", sub="system", reason="within the small-difference tolerance (BD)",
today=today, key="BD", difference=m["small_difference"], by_system=True)
b.bill_no = stamp_number(repo, "bill", "bill_number_format", today, b.bill_id)
b.status, b.approved_on = "approved", today
b.approved_by_email, b.approved_by_sub = _who(caller), _sub(caller)
if ports.installed("tax") and b.tax_ref and b.kind != "credit":
try:
ports.tax.freeze(b.tax_ref, _contents(repo, b, caller))
b.tax_frozen = True
except Missing as mm:
raise Refused(409, f"tax_content: {'; '.join(g.get('reason', '') for g in mm.gates) or mm.reason}") from None
_claim_lines(repo, b)
for ln in b.lines:
w = repo.withdrawals.get(ln.withdrawal_id) if ln.withdrawal_id else None
if w and w["status"] == "open":
repo.withdrawals.put({**w, "status": "settled", "settled_bill_id": b.bill_id})
_auto_allocate_credit(repo, b, caller, today)
v2 = repo.bills.save(b, v)
return bill_view(repo, caller, b, v2), b.bill_id
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/approve", permission=BILL_APPROVE, idempotent=True)
def approve_bill(caller: dict, bill_id: Ref, version: Optional[Version] = None) -> dict:
"""Approve a bill once it matches (or every difference is accepted by a second person), its tax contents pass, the ledger period is open and the
billed quantity still fits under what was received: our reference is stamped (BILL-2026-00017), the tax is frozen in the tax module, the billed
quantity is claimed on each order line (two bills for the same goods cannot both be approved), the bill is posted to the ledger, and any
asset line is handed to the asset register. A second approval changes nothing."""
def run():
with repository() as repo, _decide():
b, v = _load(repo, caller, bill_id)
return approve_core(repo, caller, b, v, version)
from brick_purchasing.infrastructure.tables import request_memo
with request_memo(): # the four transactions below read the same settings, account map and supplier rows: once
got = unit(version, run)
if isinstance(got, dict):
return got
out, bid = got
post_bill(caller, bid)
from brick_purchasing.application import assets
assets.after_approve(caller, bid)
with repository() as repo:
b, v = repo.bills.get(bid)
return bill_view(repo, caller, b, v)
def _claim_lines(repo, b: bl.SupplierBill) -> None:
"""Claim the billed quantity per order line: a plain INSERT keyed (purchase_order_id, line_no, bill_id), and the line's claim sequence, so two
bills approved at once for the same goods collide and the loser re-reads (and is refused not_over_billed)."""
if not b.purchase_order_id:
return
for ln in b.lines:
if ln.withdrawal_id: # one withdrawal, one bill: two approvals at once collide on this claim
repo.claims.take(f"withdrawn:{ln.withdrawal_id}", b.bill_id)
per = {}
for ln in b.lines:
if ln.order_line_no and not ln.withdrawal_id and not ln.delivery_cost_ref:
per[ln.order_line_no] = per.get(ln.order_line_no, Fraction(0)) + base_of(ln.quantity, ln.unit_numerator, ln.unit_denominator)
for line_no, q in per.items():
repo.claims.take(f"billed:{b.purchase_order_id}:{line_no}", b.bill_id)
b.claims.append(bl.BilledClaim(b.purchase_order_id, line_no, b.bill_id, as_dec(q), _stamp()))
def _auto_allocate_credit(repo, b: bl.SupplierBill, caller: dict, today: date) -> None:
"""An approved credit that names the bill it corrects reduces that bill's outstanding through an allocation of source_kind credit."""
if b.kind != "credit" or not b.corrects_bill_id:
return
got = repo.bills.get(b.corrects_bill_id)
if not got or got[0].status != "approved":
return
target = got[0]
amount = min(b.total, outstanding(repo, target))
if amount > 0:
seq = repo.claims.take(f"alloc:{target.bill_id}", b.bill_id)
repo.allocations.put(Allocation("credit", b.bill_id, target.bill_id, seq, amount, today, _who(caller)))
# ═══ posting to the ledger ══════════════════════════════════════════════════════════════════════════
def journal_for_bill(repo, b: bl.SupplierBill) -> dict:
"""The balanced journal of an approved bill: expense (or the clearing purpose of the line's kind) and input VAT debited, payables credited —
reversed for a credit. Keyed journal_id = derived_id(bill, 'approve'), so a retry posts once. An account comes from the dated account map."""
on, sign = b.invoice_date, (1 if b.kind != "credit" else -1)
s = base.Settings(repo, on)
lines = []
prof = profile_on(repo, b.bp_id, on)
reclaim = bool(s["reclaim_vat"] and prof.get("reclaims_vat", True))
small = {a.line_no: Decimal(a.difference) for a in b.acceptances if a.kind == "small_difference" and a.difference}
if b.kind == "down_payment_invoice":
return _dp_invoice_journal(repo, b, reclaim, on)
def add(side: str, amount: Decimal, memo: str, *, account=None, purpose=None, **extra):
if amount == 0:
return
e = {"debit" if side == "d" else "credit": str(abs(amount)), "memo": memo, "bp_id": str(b.bp_id), **{k: v for k, v in extra.items() if v}}
if purpose:
e["purpose"] = purpose
else:
e["account_id"] = account
if (amount < 0) != (sign < 0) and sign > 0:
e["debit" if "debit" in e else "credit"] = e.pop("debit" if "debit" in e else "credit")
lines.append(e)
for ln in b.lines:
cost = ln.net_amount + (ln.tax_amount - ln.vat_reclaimable if reclaim else ln.tax_amount)
extra = {"cost_centre": ln.cost_centre, "asset_id": ln.asset_id if ln.posting_kind == "asset" else "", "partner_company": b.partner_company,
"ic_reference": b.ic_reference or ln.ic_reference}
memo = f"{b.bill_no or b.supplier_invoice_no} line {ln.line_no}" + (f" project {ln.project_id}" if ln.project_id else "")
diff = small.get(ln.line_no, ZERO)
side = "d" if sign > 0 else "c"
if diff:
_push(lines, side, abs(diff), "small difference", purpose="small_difference", bp_id=b.bp_id, sign=sign)
cost -= diff
k = ln.posting_kind
if k == "asset":
_push(lines, side, cost, memo, purpose="asset_clearing", bp_id=b.bp_id, sign=sign, **extra)
elif k == "consignment":
_push(lines, side, cost, memo, purpose="consignment_clearing", bp_id=b.bp_id, sign=sign, **extra)
elif k == "delivery_cost":
_push(lines, side, cost, memo, purpose="delivery_cost_clearing", bp_id=b.bp_id, sign=sign, **extra)
else:
grni = account_for(repo, "grni", "*", on) if (ln.order_line_no and ln.account_assignment in ("stock", "") and ln.item_category in ("", "standard")
and ports.installed("inventory") and _is_goods(repo, b, ln)) else None
acct = grni or account_for(repo, "expense", ln.expense_category or "general", on)
_push(lines, side, cost, memo, account=acct, bp_id=b.bp_id, sign=sign, **extra)
if reclaim and ln.vat_reclaimable:
_push(lines, side, ln.vat_reclaimable, f"input VAT {ln.tax_code}", account=account_for(repo, "input_vat", ln.tax_code or "*", on), bp_id=b.bp_id,
sign=sign, partner_company=b.partner_company, ic_reference=b.ic_reference or ln.ic_reference)
_push(lines, "c" if sign > 0 else "d", b.total, f"payables {b.supplier_invoice_no}", account=account_for(repo, "payables", "*", on), bp_id=b.bp_id, sign=sign,
partner_company=b.partner_company, ic_reference=b.ic_reference)
return {"journal_id": str(derived_id(b.bill_id, "approve")), "source_module": "purchasing", "source_document": b.bill_id, "event": "approve",
"company_code": b.company_code or None, "journal_date": on.isoformat(), "currency": b.currency,
"memo": f"{'supplier credit' if b.kind == 'credit' else 'supplier bill'} {b.bill_no} {b.supplier_invoice_no}", "lines": lines}
def _dp_invoice_journal(repo, b: bl.SupplierBill, reclaim: bool, on: date) -> dict:
"""The supplier's VAT invoice for a down payment moves only the VAT: input VAT debited, supplier down payments credited. Nothing is payable (the
down payment was paid already) and without reclaim nothing moves."""
vat = sum((ln.vat_reclaimable for ln in b.lines), ZERO) if reclaim else ZERO
lines = []
if vat:
lines = [{"debit": str(vat), "account_id": account_for(repo, "input_vat", b.lines[0].tax_code or "*", on), "memo": f"input VAT on down payment {b.supplier_invoice_no}", "bp_id": str(b.bp_id)},
{"credit": str(vat), "account_id": account_for(repo, "supplier_down_payment", "*", on), "memo": f"VAT of down payment {b.supplier_invoice_no}", "bp_id": str(b.bp_id)}]
return {"journal_id": str(derived_id(b.bill_id, "approve")), "source_module": "purchasing", "source_document": b.bill_id, "event": "approve",
"company_code": b.company_code or None, "journal_date": on.isoformat(), "currency": b.currency,
"memo": f"VAT invoice for down payment {b.supplier_invoice_no}", "lines": lines}
def _is_goods(repo, b, ln) -> bool:
got = repo.orders.get(b.purchase_order_id)
x = got[0].line(ln.order_line_no) if got else None
return bool(x and x.kind == "goods")
def _push(lines: list, side: str, amount: Decimal, memo: str, *, account=None, purpose=None, bp_id=None, sign: int = 1, **extra) -> None:
if amount == 0:
return
e = {"debit" if side == "d" else "credit": str(abs(amount)), "memo": memo, "bp_id": str(bp_id)}
e.update({k: v for k, v in extra.items() if v})
if purpose:
e["purpose"] = purpose
else:
e["account_id"] = account
lines.append(e)
def post_bill(caller: dict, bill_id: str) -> dict:
"""Post an approved bill to the ledger (POST /journal-entry/journals), after the approval committed. A refusal keeps the bill approved and lists
the hand-over with its reason for a retry. Posting twice changes the ledger once (the journal id is derived)."""
with repository() as repo:
got = repo.bills.get(bill_id)
if not got or got[0].status != "approved":
raise Refused(404, f"no approved bill {bill_id}")
b = got[0]
if b.posted_journal_id or b.source_kind == "odoo" or not ports.installed("gl"):
return {"bill_id": bill_id, "posted": bool(b.posted_journal_id), "journal_id": b.posted_journal_id, "note": b.post_note or
("not posted: the ledger is not installed" if not ports.installed("gl") else "")}
body = journal_for_bill(repo, b)
if not body["lines"]:
return {"bill_id": bill_id, "posted": False, "journal_id": "", "note": "nothing to post: no VAT is reclaimed on it"}
note, jid = "", ""
try:
got = ports.gl.post_journal(body)
jid = str(got.get("journal_id") or body["journal_id"])
except Missing as m:
note = f"the ledger refused ({m.status}): " + ("; ".join(f"{g.get('gate')}: {g.get('reason')}" for g in m.gates) or m.reason)
with repository() as repo, _decide():
b, v = repo.bills.get(bill_id)
b.posted_journal_id, b.post_note = jid, note
repo.bills.save(b, v)
return {"bill_id": bill_id, "posted": bool(jid), "journal_id": jid, "note": note}
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/post", permission=BILL_POST, idempotent=True)
def post_approved_bill(caller: dict, bill_id: Ref) -> dict:
"""Post (or re-post) an approved bill to the ledger: the clock worker retries the bills the ledger refused or did not answer. Idempotent: the
journal id is derived from the bill and the event."""
return post_bill(caller, bill_id)
# ═══ reading ═══════════════════════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="query", route="GET /bills/open", permission=BILL_READ, tool=False)
def list_open_bills(caller: dict, as_of: IsoDate, bp_id: Optional[BigId] = None) -> dict:
"""The payables open items as of a date: each approved bill (and credit, negative) with what is outstanding after allocations dated on or
before it. The ledger reconciles its payables control account against this; payments' cash forecast reads it; the migration's open-payables
check is this."""
with repository() as repo:
d = base._day(as_of, "as_of")
return {"as_of": d, "items": [{**i, "bp_id": str(i["bp_id"])} for i in open_items(repo, d, bp_id)]}
@PURCHASING.use_case(kind="query", route="GET /bills/{bill_id}", permission=BILL_READ, etag="version")
def get_bill(caller: dict, bill_id: Ref) -> dict:
"""One bill with its lines, the MATCH derived now (per line: price, quantity, tax, each with its tolerance key and the difference), the
acceptances, what is outstanding and the readiness gates of approval."""
with repository() as repo:
b, v = _load(repo, caller, bill_id)
return bill_view(repo, caller, b, v)
@PURCHASING.use_case(kind="query", route="GET /bills", permission=BILL_READ)
def list_bills(caller: dict, bp_id: Optional[BigId] = None, status: Optional[Code] = None, kind: Optional[Code] = None, limit: Limit = 50,
offset: Offset = 0) -> dict:
"""Bills and credits (heads) with what is outstanding and the days until (or past) due as of today in the business time zone — derived on
read, never stored."""
with repository() as repo:
today = _today(repo)
rows, total = repo.bills.heads(bp_id=bp_id, status=status, kind=kind, limit=limit, offset=offset)
sums = repo.allocations.sums([b.bill_id for b in rows])
out = []
for b in rows:
out.append({"bill_id": b.bill_id, "bill_no": b.bill_no, "bp_id": b.bp_id, "supplier_invoice_no": b.supplier_invoice_no, "kind": b.kind,
"invoice_date": b.invoice_date, "currency": b.currency, "total": b.total, "status": b.status, "due_date": b.due_date,
"outstanding": b.total - sums.get(b.bill_id, ZERO) if b.kind != "credit" else outstanding(repo, b),
"days_to_due": (b.due_date - today).days if b.due_date else None})
return {"total": total, "as_of": today, "bills": out}
def open_items(repo, as_of: date, bp_id: Optional[int] = None) -> list:
rows = [b for b in repo.bills.approved() if (bp_id is None or b.bp_id == int(bp_id)) and (b.approved_on or b.invoice_date) <= as_of]
sums = repo.allocations.sums([b.bill_id for b in rows], as_of)
out = []
for b in sorted(rows, key=lambda x: (x.due_date or x.invoice_date, x.bill_id)):
o = outstanding(repo, b, as_of) if b.kind == "credit" else b.total - sums.get(b.bill_id, ZERO)
sign = -1 if b.kind == "credit" else 1
if o != 0:
out.append({"bill_id": b.bill_id, "bill_no": b.bill_no, "bp_id": b.bp_id, "currency": b.currency, "total": sign * b.total, "outstanding": sign * o,
"due_date": b.due_date})
return out
# ═══ AI steps (suggest only) ═════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/ai-check", permission=AI_CHECK)
def ai_check_bill(caller: dict, bill_id: Ref) -> dict:
"""AI step (suggest): list likely duplicates (the same total near the date, another number) and an unusual amount for this supplier, each with
its evidence, for the clerk to accept or dismiss. It never changes a bill; approval never waits on it."""
with repository() as repo:
b, v = _load(repo, caller, bill_id)
mode = base.Settings(repo, _today(repo)).mode("ai_check_bill")
_ai_check(repo, b)
history = [x.total for x in repo.bills.repo.all_heads() if x.bp_id == b.bp_id and x.bill_id != b.bill_id]
unusual = ai_steps.unusual_amount(b.total, history)
if unusual:
repo.findings.put({"subject_ref": f"pur_bill:{b.bill_id}", "step": "ai_check_bill", "run_no": 1, "finding_no": 99, "model": "baseline",
"inputs_digest": "", "finding": json.dumps(unusual), "evidence": unusual["why"]})
ans = ai_steps.ask("ai_check_bill", mode, f"pur_bill:{b.bill_id}", v, {"total": str(b.total)})
return {"mode": mode, "answer": ans.as_dict(), "findings": repo.findings.all(subject_ref=f"pur_bill:{b.bill_id}")}
@PURCHASING.use_case(kind="command", route="POST /bills/{bill_id}/findings/{finding_no}/decide", permission=BILL_WRITE)
def decide_ai_finding(caller: dict, bill_id: Ref, finding_no: int, decision: str) -> dict:
"""The clerk accepts, corrects or dismisses an AI finding on a bill; the decision, who and when are recorded in pur_ai_finding."""
if decision not in ("accepted", "corrected", "dismissed"):
raise Refused(422, "decision is accepted, corrected or dismissed")
with repository() as repo:
rows = [r for r in repo.findings.all(subject_ref=f"pur_bill:{bill_id}") if r["finding_no"] == finding_no]
if not rows:
raise Refused(404, f"no finding {finding_no} on bill {bill_id}")
row = rows[0]
row.update(decision=decision, decided_by_email=_who(caller), decided_at=_stamp())
repo.findings.put(row)
return row