dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/self_billing.py
"""Self-billing and ERS settlement (PUR-50, PUR-51): an agreement per supplier, approved by a second person and dated; receipts of order lines flagged
ers settled into invoices WE issue for the supplier, numbered from our SB series, taxed (self_billed true), approved by the run and posted.
An invoice has ONE order (a bill carries one purchase_order_id), so the run raises one per supplier, currency and order (deviation from 'per supplier'
where a supplier has several open orders; the report says so). Quantities come from the receipts, the price from the order line: no match failure
is possible. A return after settlement becomes a self-billed credit in the next run."""
import base64
import binascii
import hashlib
from datetime import date, timedelta
from fractions import Fraction
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, text
from brick_purchasing.application import base, bills, payments, rules
from brick_purchasing.application.base import (
BILL_READ,
ERS_RUN,
PURCHASING,
SB_APPROVE,
SB_WRITE,
_decide,
_ids,
_sub,
_today,
_who,
unit,
)
from brick_purchasing.application.inputs import BillLineIn
from brick_purchasing.domain import bill as bl
from brick_purchasing.domain.approval import identities, same_person
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.values import add_months, as_dec, fr
from brick_purchasing.infrastructure.seams import Missing, ports
repository = lambda: base.repository() # noqa: E731
RENEW_DAYS = 30
def _view(r: dict, today: date) -> dict:
out = dict(r)
out["bp_id"] = str(r["bp_id"])
out["in_force"] = r["status"] == "approved" and r["valid_from"] <= today and (r["valid_to"] is None or r["valid_to"] >= today)
out["renewal_due"] = bool(r["status"] == "approved" and r["valid_to"] and today <= r["valid_to"] <= today + timedelta(days=RENEW_DAYS))
return out
def in_force_on(repo, bp_id: int, on: date) -> Optional[dict]:
rows = [r for r in repo.self_billing.all(bp_id=int(bp_id)) if r["status"] == "approved" and r["valid_from"] <= on and (r["valid_to"] is None or r["valid_to"] >= on)]
return max(rows, key=lambda r: r["valid_from"]) if rows else None
def _vat_number(bp_id: int) -> tuple:
"""(vat number, registered) from the tax module's party record, else the partner's tax id."""
if ports.installed("tax"):
try:
p = ports.tax.party(bp_id)
cur = p.get("current") or {}
return str(cur.get("vat_number") or ""), cur.get("treatment_code") not in ("unregistered", "outside_scope")
except Missing:
pass
return "", True
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/self-billing", permission=SB_WRITE)
def propose_self_billing(caller: dict, bp_id: BigId, valid_from: IsoDate, valid_to: Optional[IsoDate], document_base64: text(14_000_000, min_length=8)) -> dict:
"""Record the supplier's signed self-billing agreement (a PDF kept as raw/purchasing/self-billing/<sha256>.pdf) for a period: it needs an end
date no more than the owner's maximum (12 months) after its start, and the supplier's VAT number where they are registered. A second person
approves it; 30 days before its end it is listed for renewal."""
def run():
with repository() as repo, _decide():
if valid_to is None:
raise RuleBroken("end_date: a self-billing agreement has an end date; fix: enter the date the supplier signed to")
start, end = base._day(valid_from, "valid_from"), base._day(valid_to, "valid_to")
mx = int(repo.settings.value_on("self_billing_max_months", _today(repo)))
if end <= start or end > add_months(start, mx):
raise RuleBroken(f"max_months: an agreement runs at most {mx} months from its start (and ends after it starts); fix: renew it for a shorter period")
vat, registered = _vat_number(int(bp_id))
if registered and not vat:
raise RuleBroken("supplier_vat_number: the supplier is VAT-registered and has no VAT number on file; fix: record it on the supplier first")
try:
data = base64.b64decode(document_base64, validate=True)
except (binascii.Error, ValueError):
raise Refused(422, "document_base64 is not base64") from None
if not data.startswith(b"%PDF"):
raise Refused(422, "the signed agreement is a PDF")
if repo.self_billing.get(int(bp_id), start) is not None:
raise Refused(409, f"there is already an agreement for this supplier from {start}")
key = f"raw/purchasing/self-billing/{hashlib.sha256(data).hexdigest()}.pdf"
payments.objects().put(key, data, "application/pdf")
row = {"bp_id": int(bp_id), "valid_from": start, "valid_to": end, "document_key": key, "supplier_vat_number": vat, "status": "proposed",
"entered_by_email": _who(caller), "entered_by_sub": _sub(caller), "approved_by_email": "", "approved_by_sub": "", "end_reason": ""}
repo.self_billing.put(row)
return _view(row, _today(repo))
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /suppliers/{bp_id}/self-billing/{valid_from}/approve", permission=SB_APPROVE, idempotent=True)
def approve_self_billing(caller: dict, bp_id: BigId, valid_from: IsoDate) -> dict:
"""Approve the agreement: not by whoever recorded it (same email OR same sign-in). From its start date ERS may settle that supplier's receipts."""
def run():
with repository() as repo, _decide():
row = repo.self_billing.get(int(bp_id), base._day(valid_from, "valid_from"))
if row is None:
raise Refused(404, f"no self-billing agreement for supplier {bp_id} from {valid_from}")
if row["status"] == "approved":
return _view(row, _today(repo))
if same_person(_ids(caller), identities(row["entered_by_email"], row["entered_by_sub"])) or not _ids(caller):
raise RuleBroken("second_person: you recorded this agreement (same email or same sign-in); fix: another approver")
row.update(status="approved", approved_by_email=_who(caller), approved_by_sub=_sub(caller))
repo.self_billing.put(row)
return _view(row, _today(repo))
return unit(None, run)
@PURCHASING.use_case(kind="query", route="GET /self-billing-agreements", permission=BILL_READ)
def list_self_billing(caller: dict, bp_id: Optional[BigId] = None, renewal_due: bool = False) -> dict:
"""The agreements, each with whether it is in force today; renewal_due lists those ending within 30 days."""
with repository() as repo:
today = _today(repo)
items = [_view(r, today) for r in (repo.self_billing.all(bp_id=int(bp_id)) if bp_id else repo.self_billing.all())]
return {"items": [i for i in items if i["renewal_due"] or not renewal_due]}
# ═══ PUR-51: the settlement ═══════════════════════════════════════════════════════════════════════
def _pieces(repo, po_id: str, line_no: int) -> list:
"""The unbilled receipt pieces of an ers order line, oldest first: [(received_on, receipt_id, base quantity)]. Billed quantity consumes the oldest."""
rows = sorted([r for r in repo.receipts.lines_of_order(po_id) if r["order_line_no"] == line_no and r["kind"] == "receipt"], key=lambda r: (r["received_on"], r["receipt_id"]))
billed = rules.billed_by_line(repo, po_id).get(line_no, Fraction(0))
net = rules.received_by_line(repo, po_id).get(line_no, Fraction(0))
out, skip = [], max(billed, Fraction(0))
for r in rows:
q = fr(r["quantity_base"])
take = max(q - skip, Fraction(0))
skip = max(skip - q, Fraction(0))
if take > 0:
out.append([r["received_on"], r["receipt_id"], take])
over = sum((p[2] for p in out), Fraction(0)) - max(net - billed, Fraction(0)) # a return since: the newest pieces are not owed
while over > 0 and out:
cut = min(over, out[-1][2])
out[-1][2] -= cut
over -= cut
if out[-1][2] == 0:
out.pop()
return [tuple(p) for p in out]
def _candidates(repo, up_to: date) -> list:
"""[(bp_id, currency, purchase_order_id)] with something to settle: unbilled ers receipts, or billed goods since returned."""
out = []
for h in repo.orders.repo.all_heads():
if h.status not in ("confirmed", "closed"):
continue
o = repo.orders.get(h.purchase_order_id)[0]
if any(x.ers for x in o.lines):
out.append((o.bp_id, o.currency, o.purchase_order_id))
return out
def _credit_base(repo, po_id: str, line_no: int) -> Fraction:
billed = rules.billed_by_line(repo, po_id).get(line_no, Fraction(0))
net = rules.received_by_line(repo, po_id).get(line_no, Fraction(0))
return max(billed - max(net, Fraction(0)), Fraction(0))
def _invoice_text(b: bl.SupplierBill, agreement: dict) -> str:
out = [f"SELF-BILLED INVOICE {b.bill_no or b.supplier_invoice_no}" + (" (CREDIT)" if b.kind == "credit" else ""),
f"Issued by the customer on behalf of supplier {b.bp_id} under the self-billing agreement {agreement['valid_from']} to {agreement['valid_to']} (document {agreement['document_key']})",
f"Supplier VAT number: {agreement['supplier_vat_number']} Tax point: {b.invoice_date} Currency: {b.currency}", ""]
out += [f"{ln.line_no}. {ln.description or ln.product_id} {ln.quantity} {ln.unit_code} x {ln.unit_price} net {ln.net_amount} VAT {ln.tax_code} {ln.tax_rate}% = {ln.tax_amount}" for ln in b.lines]
out += ["", f"Net {b.subtotal} VAT {b.tax_total} Total {b.total}", "The supplier accepts this invoice as the invoice for the supply and will not issue another."]
return "\n".join(out)
def _settle_order(caller: dict, bp_id: int, currency: str, po_id: str, up_to: date) -> dict:
refused, raised = [], []
with repository() as repo, _decide():
today = _today(repo)
o = repo.orders.get(po_id)[0]
on = min(up_to, today)
grouping = repo.settings.value_on("ers_grouping", today)
plan = {} # group -> [(order line, base qty, sign)]
agreement_of = {}
for ln in (x for x in o.lines if x.ers):
for received_on, rid, q in _pieces(repo, po_id, ln.line_no):
if received_on > up_to:
continue
ag = in_force_on(repo, bp_id, received_on)
if ag is None:
refused.append({"receipt_id": rid, "order_line_no": ln.line_no, "reason": f"no approved self-billing agreement in force on {received_on}; fix: record and approve one, or wait for the supplier's invoice"})
continue
agreement_of[rid] = ag
plan.setdefault(("bill", rid if grouping == "per_receipt" else ""), {}).setdefault(ln.line_no, Fraction(0))
plan[("bill", rid if grouping == "per_receipt" else "")][ln.line_no] += q
cr = _credit_base(repo, po_id, ln.line_no)
if cr > 0:
ag = in_force_on(repo, bp_id, on)
if ag is None:
refused.append({"order_line_no": ln.line_no, "reason": f"a return after settlement needs a self-billing agreement in force on {on}"})
else:
agreement_of["credit"] = ag
plan.setdefault(("credit", ""), {})[ln.line_no] = cr
for (kind, _g), per in plan.items():
lines_in = []
for line_no, qb in per.items():
ol = o.line(line_no)
qty = as_dec(qb * Fraction(ol.unit_denominator, ol.unit_numerator))
lines_in.append(BillLineIn(order_line_no=line_no, quantity=qty, unit_code=ol.unit_code, pricing_unit_code=ol.pricing_unit_code,
pricing_quantity=ol.pricing_quantity, unit_price=ol.unit_price))
k = 0
while True:
k += 1
number = base.stamp_number(repo, "self_billing", "self_billing_number_format", today, f"ers:{po_id}:{on}:{kind}:{k}")
break
bid = bl.bill_id_of(bp_id, number, on)
built = bills.build_bill_lines(repo, caller, o, lines_in, bid, bp_id, currency)
b, v, _created = bills.record_core(repo, caller, bp_id=bp_id, supplier_invoice_no=number, invoice_date=on, currency=currency, lines=built,
purchase_order_id=po_id, kind=kind, source_kind="ers", self_billed=True, company_code=o.company_code,
recorded_by=(_who(caller), _sub(caller)))
v = repo.bills.save(b, 0)
try:
bills.approve_core(repo, {"email": "system", "sub": "ers"}, b, v)
except (RuleBroken, Refused) as e:
raise RuleBroken(f"{number} could not be approved: {e}") from e
ag = agreement_of.get("credit" if kind == "credit" else next(iter(agreement_of)))
sent_key = f"raw/purchasing/self-billing/sent/{number}.txt"
payments.objects().put(sent_key, _invoice_text(b, ag).encode("utf-8"), "text/plain")
raised.append({"bill_id": bid, "bill_no": b.bill_no, "supplier_invoice_no": number, "kind": kind, "purchase_order_id": po_id,
"net": b.subtotal, "tax": b.tax_total, "total": b.total, "sent_to_supplier": sent_key})
return {"raised": raised, "refused": refused}
@PURCHASING.use_case(kind="command", route="POST /ers/settle", permission=ERS_RUN)
def settle_ers(caller: dict, up_to: IsoDate) -> dict:
"""Settle the receipts of ers-flagged order lines up to a date into self-billed invoices (recorded, tax computed with self_billed true and
frozen, approved by the run, posted, a copy kept to send to the supplier): one per supplier, currency and order, at the order price for the
quantity received and not yet billed. Receipts with no agreement in force on their date are listed, never billed. A second run settles nothing
new, a receipt the supplier already billed is skipped, and a return after settlement becomes a self-billed credit. The clerk who ran it cannot
approve a payment run containing its invoices."""
day = base._day(up_to, "up_to")
with repository() as repo:
cands = _candidates(repo, day)
raised, refused = [], []
for bp_id, cur, po in cands:
got = unit(None, lambda bp_id=bp_id, cur=cur, po=po: _settle_order(caller, bp_id, cur, po, day))
raised += got["raised"]
refused += got["refused"]
for r in raised:
bills.post_bill(caller, r["bill_id"])
return {"up_to": day, "raised": raised, "refused": refused, "settled_nothing_new": not raised}