dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/application/rebates.py
"""Rebate accrual and settlement (SO-33, SO-34).
A rebate is earned over a year and paid back once. While the year runs the order brick ACCRUES it: each night, for the previous business
day, the invoice and credit-note lines in the agreement's basis (read from invoicing's GET /billed-lines, never from ar_ tables) add to the
agreement's accrual, and the ledger is asked for the journal (rebate expense / accrued rebates). When the agreement ends, the tier reached
on the whole year is settled: asked by one person, approved by another, credited ONCE in invoicing (against accrued rebates, not revenue
again), the accrual trued up in the ledger so the agreement's accrual account nets to exactly 0.00.
Order of things (a retry must never double anything): the accrual rows, the pending posting and the agreement's claim are committed FIRST,
then the ledger and invoicing are asked with ids derived from the real event, so a retry asks the same question. A journal the ledger
refuses stays pending with its reason in so_posting and is retried on the next run.
"""
import json
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.repository import retry_on_conflict
from erp_platform.seam import derived_id
from brick_sales_order.application import use_cases as uc
from brick_sales_order.application.inputs import Day, LineNo, Reason, Ref, day
from brick_sales_order.domain.values import ZERO, num
from brick_sales_order.infrastructure.seams import Missing, ports
SO, ACCRUE, APPROVE_REBATE, REBATE = uc.SO, "so:rebate:accrue", "so:rebate:approve", uc.REBATE
D = Decimal
def _mode(repo, today: date) -> str:
"""post | report: unset = post when the ledger is installed, else a monthly report for the accountant."""
return repo.settings.value_on("rebate_accrual_mode", today) or ("post" if ports.gl_installed() else "report")
def _accounts(repo, today: date) -> dict:
raw = repo.settings.value_on("rebate_accounts", today)
try:
return json.loads(raw) if raw else {}
except ValueError:
return {}
def _line(acc: dict, purpose: str, side: str, amount: Decimal, bp_id: int) -> dict:
who = {"account_id": int(acc[purpose])} if acc.get(purpose) else {"purpose": purpose}
return {**who, side: num(amount, 2), "bp_id": int(bp_id)}
def _payload(repo, a, source_document: str, event: str, amount: Decimal, on: date, memo: str, today: date) -> dict:
"""The journal the ledger is asked for: expense debit / accrued credit for money earned, the reverse for a credit note or a negative
true-up. journal_id is derived from (source document, event): the same event never names a second journal."""
acc = _accounts(repo, today)
big, small = ("debit", "credit") if amount > 0 else ("credit", "debit")
lines = [_line(acc, "rebate_expense", big, abs(amount), a.bp_id), _line(acc, "accrued_rebates", small, abs(amount), a.bp_id)]
return {"journal_id": str(derived_id(source_document, event)), "source_document": source_document, "event": event,
"company_code": str(acc.get("company_code") or ""), "journal_date": on.isoformat(), "currency": a.currency, "memo": memo, "lines": lines}
def _add_posting(repo, a, document_id: str, event: str, payload: dict, mode: str) -> str:
status = "reported" if mode == "report" else "pending"
repo.postings.add({"document_id": document_id, "event": event, "journal_id": payload["journal_id"], "status": status, "reason": "",
"attempts": 0, "rebate_agreement_id": a.rebate_agreement_id, "payload_json": json.dumps(payload)})
return status
def _basis(repo, a, lo: date, hi: date) -> list:
"""The billed lines of the agreement's customers in [lo, hi] that are in its basis, from invoicing."""
if a.bp_id:
bps = [a.bp_id]
else:
latest = {}
for p in repo.policies.all():
if p["valid_from"] <= hi:
latest[p["bp_id"]] = p["price_group"]
bps = sorted(b for b, g in latest.items() if g == a.price_group)
out = []
for bp in bps:
try:
got = ports.invoicing.billed_lines(bp_id=bp, issued_from=lo, issued_to=hi)
except Missing as m:
raise uc._refuse(m, "invoicing (billed lines)") from None
for x in got:
if a.basis == "products" and x.get("product_id") not in a.basis_values:
continue
if a.basis == "categories" and x.get("category_code") not in a.basis_values:
continue
out.append({**x, "issue_date": date.fromisoformat(str(x["issue_date"])[:10])})
return out
def _accrue_one(aid: str, lo: date, hi: date) -> dict:
"""Accrue one agreement for [lo, hi]: rows, pending postings and the claim in ONE commit; nothing is asked of the ledger yet."""
def unit():
with uc.repository() as repo, uc._decide():
a, v = uc._agreement(repo, aid, None)
lo2, hi2 = max(lo, a.valid_from), min(hi, a.valid_to)
if a.lifecycle not in ("active", "ended") or hi2 < lo2:
return {"rebate_agreement_id": aid, "accrued": 0, "amount": "0.00"}
today = uc._today(repo)
new = a.accrue(_basis(repo, a, lo2, hi2), a.accrual_rate_pct)
if not new:
return {"rebate_agreement_id": aid, "accrued": 0, "amount": "0.00"}
mode = _mode(repo, today)
docs: dict = {}
for x in new:
docs.setdefault(x.document_id, []).append(x)
for doc, rows in docs.items():
amount = sum((r.accrual_amount for r in rows), ZERO)
src = f"rebate:{aid}:{doc}"
if amount == 0:
a.mark_posting(doc, "posted")
continue
payload = _payload(repo, a, src, "accrual", amount, rows[0].issue_date, f"Rebate accrual {aid} on {doc}", today)
status = _add_posting(repo, a, src, "accrual", payload, mode)
for r in rows:
r.journal_id, r.posting_status = payload["journal_id"], status
repo.rebates.claim(aid, "accrue")
repo.rebates.save(a, v)
return {"rebate_agreement_id": aid, "accrued": len(new), "amount": num(sum((r.accrual_amount for r in new), ZERO), 2)}
return retry_on_conflict(unit)
def _post_pending(aid: str) -> dict:
"""Ask the ledger for every pending journal of the agreement (derived ids: a retry is a replay). A refusal keeps it pending with the
ledger's reason."""
with uc.repository() as repo:
pend = repo.postings.pending(aid)
posted, kept = 0, 0
for p in pend:
payload = json.loads(p["payload_json"])
payload["journal_date"] = date.fromisoformat(payload["journal_date"])
try:
ports.gl.post_journal(**payload)
status, reason = "posted", ""
except Missing as m:
status, reason = "pending", m.reason
with uc.repository() as repo:
repo.postings.settle(p["document_id"], p["event"], status, reason, int(p["attempts"] or 0) + 1, p["journal_id"])
if p["event"] == "accrual":
doc = p["document_id"].split(":", 2)[2]
retry_on_conflict(lambda doc=doc, status=status: _mark(aid, doc, status))
posted, kept = posted + (status == "posted"), kept + (status == "pending")
return {"posted": posted, "pending": kept}
def _mark(aid: str, doc: str, status: str) -> None:
with uc.repository() as repo:
a, v = uc._agreement(repo, aid, None)
if any(x.document_id == doc and x.posting_status != status for x in a.accruals):
a.mark_posting(doc, status)
repo.rebates.save(a, v)
def _end_one(aid: str) -> bool:
"""Fix the final basis: accrue the whole validity period (idempotent: only what is missing), then end the agreement."""
def unit():
with uc.repository() as repo, uc._decide():
a, v = uc._agreement(repo, aid, None)
today = uc._today(repo)
if a.lifecycle != "active":
return False
if not a.valid_to < today:
a.end(today) # raises the gate: valid_to_passed
return True
if not retry_on_conflict(unit):
return False
_accrue_one(aid, date(1970, 1, 1), date(9999, 12, 31))
_post_pending(aid)
def close():
with uc.repository() as repo, uc._decide():
a, v = uc._agreement(repo, aid, None)
if a.end(uc._today(repo)):
repo.rebates.claim(aid, "end")
repo.rebates.save(a, v)
return True
return False
return retry_on_conflict(close)
@SO.use_case(kind="command", route="POST /rebate-agreements/accrue", permission=ACCRUE, idempotent=True)
def accrue_rebates(caller: dict, for_date: Optional[Day] = None) -> dict:
"""The nightly rebate accrual (a machine's): for one business day (default the previous one, counted in business_timezone, never the UTC
date) read the invoice and credit-note lines each active agreement's customer was billed (invoicing's /billed-lines) that are in the
agreement's basis, write one accrual row per (agreement, document, line) and ask the ledger for the journal (rebate expense debit /
accrued rebates credit; a credit note reverses it). Running it again for the same day writes nothing and posts nothing. A journal the
ledger refuses (a closed period) stays pending with its reason and is retried by the next run. Agreements whose last day has passed are
then ended with their final basis fixed."""
with uc.repository() as repo:
today = uc._today(repo)
try:
on = day(for_date, "for_date") or today - timedelta(days=1)
except ValueError as e:
raise Refused(422, str(e)) from None
if on > today:
raise Refused(422, f"the business date is {today}: a run is never for a day in the future ({on})")
every = repo.rebates.all()
out = [_accrue_one(a.rebate_agreement_id, on, on) for a in every
if a.lifecycle in ("active", "ended") and a.valid_from <= on <= a.valid_to]
with uc.repository() as repo:
due = [a.rebate_agreement_id for a in repo.rebates.all() if a.lifecycle == "active" and a.valid_to < today]
with uc._decide():
ended = [aid for aid in due if _end_one(aid)]
with uc.repository() as repo:
every = [a.rebate_agreement_id for a in repo.rebates.all() if a.lifecycle in ("active", "ended")]
posting = {aid: _post_pending(aid) for aid in every}
return uc._plain({"for_date": on, "agreements": out, "ended": ended,
"posted": sum(p["posted"] for p in posting.values()), "pending": sum(p["pending"] for p in posting.values())})
@SO.use_case(kind="command", route="POST /rebate-agreements/{rebate_agreement_id}/end", permission=ACCRUE, idempotent=True)
def end_rebate_agreement(caller: dict, rebate_agreement_id: Ref) -> dict:
"""End an agreement whose last day has passed (gate valid_to_passed): the whole validity period is accrued first (only what is missing is
written), so the final basis is fixed, and no further accrual follows. A repeat changes nothing."""
with uc.repository() as repo:
uc._agreement(repo, rebate_agreement_id, None)
with uc._decide():
_end_one(rebate_agreement_id)
with uc.repository() as repo:
a, v = uc._agreement(repo, rebate_agreement_id, None)
return uc._plain({**a.view(), "version": v})
# ── settlements ───────────────────────────────────────────────────────────────────────────────
def _request(caller: dict, aid: str, kind: str, reason: str) -> dict:
with uc.repository() as repo, uc._decide():
a, v = uc._agreement(repo, aid, None)
st = a.request_settlement(kind, reason, by_email=(caller.get("email") or "").strip().lower(),
by_sub=(caller.get("sub") or "").strip().lower(), today=uc._today(repo))
repo.rebates.claim(aid, f"request_{kind}_settlement")
new = repo.rebates.save(a, v)
return uc._plain({**a.view(), "version": new, "settlement_no": st.settlement_no})
@SO.use_case(kind="command", route="POST /rebate-agreements/{rebate_agreement_id}/partial-settlements", permission=REBATE)
def request_partial_settlement(caller: dict, rebate_agreement_id: Ref, reason: Reason) -> dict:
"""Ask for a partial settlement of an ACTIVE agreement: the rate the scale has reached on the basis so far, less what was already
settled. It is a draft until a second person approves it (approve_partial_settlement). One open request at a time."""
return _request(caller, rebate_agreement_id, "partial", reason)
@SO.use_case(kind="command", route="POST /rebate-agreements/{rebate_agreement_id}/final-settlement", permission=REBATE)
def request_final_settlement(caller: dict, rebate_agreement_id: Ref, reason: Reason) -> dict:
"""Ask for the final settlement of an ENDED agreement: the tier reached on the whole basis, retroactive, less what partial settlements
paid, and the true-up of the accrual (earned less accrued). A draft until a second person approves it."""
return _request(caller, rebate_agreement_id, "final", reason)
def _approve(caller: dict, aid: str, settlement_no: int, kind: str) -> dict:
def first():
with uc.repository() as repo, uc._decide():
a, v = uc._agreement(repo, aid, None)
today = uc._today(repo)
st, newly = a.approve_settlement(settlement_no, kind, ids=uc._ids(caller), email=(caller.get("email") or "").strip().lower(),
sub=(caller.get("sub") or "").strip().lower(), today=today)
if newly:
repo.rebates.claim(aid, f"approve_{kind}_settlement")
if st.true_up_amount != 0:
src = f"rebate:{aid}:{st.settlement_no}"
payload = _payload(repo, a, src, "true_up", st.true_up_amount, today, f"Rebate true-up {aid}", today)
_add_posting(repo, a, src, "true_up", payload, _mode(repo, today))
repo.rebates.save(a, v)
return {k: getattr(st, k) for k in ("status", "amount", "lines_json", "credit_id")} | {"currency": a.currency, "bp_id": a.bp_id}
snap = retry_on_conflict(first)
if snap["status"] == "credited":
pass # approving twice credits once
else:
pend = _post_pending(aid)
if pend["pending"]:
with uc.repository() as repo:
why = "; ".join(p["reason"] for p in repo.postings.pending(aid) if p["reason"]) or "the ledger has not taken the journal"
raise Refused(409, f"the ledger did not take the accrual journals ({why}): approve again once it can — nothing was credited yet")
credit_id = ""
if D(str(snap["amount"])) > 0:
lines = [{"description": f"Rebate {aid} settlement {settlement_no}" + (f" ({x['tax_category']})" if x["tax_category"] else ""),
"quantity": "1", "unit_price": x["net_amount"], "net_amount": x["net_amount"],
**({"tax_category": x["tax_category"]} if x["tax_category"] else {})} for x in json.loads(snap["lines_json"])]
try:
got = ports.invoicing.credit_note(source_id=f"rebate:{aid}:{settlement_no}", bp_id=snap["bp_id"], currency=snap["currency"],
reason=f"Rebate {aid}: settlement {settlement_no}", account_purpose="rebate", lines=lines)
except Missing as m:
raise uc._refuse(m, "invoicing (rebate credit note)") from None
credit_id = str(got.get("credit_id") or f"rebate:{aid}:{settlement_no}")
def done():
with uc.repository() as repo, uc._decide():
a, v = uc._agreement(repo, aid, None)
if a.settlement(settlement_no).status != "credited":
a.mark_credited(settlement_no, credit_id)
repo.rebates.save(a, v)
retry_on_conflict(done)
with uc.repository() as repo:
a, v = uc._agreement(repo, aid, None)
return uc._plain({**a.view(), "version": v})
@SO.use_case(kind="command", route="POST /rebate-agreements/{rebate_agreement_id}/partial-settlements/{settlement_no}/approve",
permission=APPROVE_REBATE, idempotent=True)
def approve_partial_settlement(caller: dict, rebate_agreement_id: Ref, settlement_no: LineNo) -> dict:
"""Approve a partial settlement as a SECOND person (never the one who asked for it or opened the agreement, by email or sign-in subject:
gate approver_not_requester). Invoicing is asked for one credit note keyed rebate:{agreement}:{settlement_no} against accrued rebates;
the accrual is released by its amount. Approving twice credits once."""
return _approve(caller, rebate_agreement_id, settlement_no, "partial")
@SO.use_case(kind="command", route="POST /rebate-agreements/{rebate_agreement_id}/final-settlement/{settlement_no}/approve",
permission=APPROVE_REBATE, idempotent=True)
def approve_final_settlement(caller: dict, rebate_agreement_id: Ref, settlement_no: LineNo) -> dict:
"""Approve the final settlement as a SECOND person (gate approver_not_requester). The accrual is trued up in the ledger (earned less
accrued), invoicing is asked for ONE credit note keyed rebate:{agreement}:{settlement_no} with a line per tax category of the basis
(against accrued rebates, not revenue again), and the agreement closes with its accrual account at exactly 0.00. Approving twice
credits once."""
return _approve(caller, rebate_agreement_id, settlement_no, "final")