dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/remittances.py
"""Early-settlement discounts (PAY-16) and customers' remittance advice (PAY-17).
Both end in the SAME allocation table every other settlement uses: the discount as invoicing's credit (allocated by ar through
POST /allocations/from-source) or, under invoice_wording, as an allocation with source_kind settlement_discount; a remittance as
a match suggestion (model "remittance", not AI) that a person accepts, changes or rejects exactly like PAY-14.
"""
import re
from decimal import ROUND_HALF_UP, Decimal
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, text
from brick_payments.application import use_cases as uc
from brick_payments.application.inputs import Code, LineIn, Money, Ref
from brick_payments.domain import remittance as rem
from brick_payments.domain import settlement as stl
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO
from brick_payments.domain.readiness import fmt, refuse, unmet
from brick_payments.domain.receipt import Receipt
from brick_payments.infrastructure import seams
PAY = uc.PAY
REMIT = "pay:remittance:import"
# ═══ PAY-16: the early-settlement discount ═══════════════════════════════════════════════════════════════
@PAY.use_case(kind="command", route="POST /receipts/{payment_id}/settlement-discount", permission=uc.ALLOC, idempotent=True)
def apply_settlement_discount(caller: dict, payment_id: Ref, invoice_id: Code, discount_amount: Optional[Money] = None) -> dict:
"""Take the early-settlement discount on an invoice a receipt has paid at the discounted amount (2/10 net 30: 2% if received
within 10 days). Two gates, in the customer's words: within_discount_window - the receipt's received_on (the BANK's date, not today)
is on or before the invoice's settlement_discount_until; discount_matches_terms - the discount is exactly the terms' percentage of
the gross outstanding at receipt, and the receipt paid the invoice the rest. Then, by settlement_discount_vat: credit_note
(default) - invoicing raises the discount credit at the invoice's own rates (ar AR-38) and allocates it, so the invoice is paid;
invoice_wording - the discount is an allocation with source_kind settlement_discount, which tax reads for the return. A receipt
after the window gets no discount: the difference stays outstanding (or goes to the tolerance). One discount per invoice."""
with uc.repository() as repo, uc._decide():
r = uc._receipt(repo, payment_id)
if not r.live or r.bp_id is None:
raise RuleBroken(f"Receipt {payment_id} is returned or not on a customer. A discount needs a live receipt on the customer")
doc = seams.invoice(invoice_id, caller)
if doc is None:
raise Refused(404, f"no settleable invoice {invoice_id}")
if doc.bp_id != r.bp_id or doc.currency != r.currency:
raise RuleBroken(f"Invoice {doc.number} is not this receipt's customer and currency. A discount settles the receipt's own invoice")
who, now = uc._who(caller), uc._now()
if doc.discount_pct is None or doc.discount_until is None:
refuse(unmet("within_discount_window", f"Invoice {doc.number} carries no early-settlement terms", "There is no discount to take"))
if r.received_on > doc.discount_until:
refuse(unmet("within_discount_window", f"Received on {r.received_on}, after the discount ended on {doc.discount_until}",
"None: the short payment stays outstanding or goes to the tolerance (PAY-04)"))
taken = repo.claims.ref_of("discount", invoice_id)
if taken:
if taken == payment_id:
return {"payment_id": payment_id, "invoice_id": invoice_id, "unchanged": True}
raise RuleBroken(f"Invoice {doc.number} already took its settlement discount (receipt {taken}). One discount per invoice")
rows = repo.allocations.for_invoices([invoice_id])
gross_at_receipt = stl.outstanding(doc.total, rows, r.received_on)
expected = (gross_at_receipt * doc.discount_pct / Decimal(100)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
asked = discount_amount if discount_amount is not None else expected
if asked != expected:
refuse(unmet("discount_matches_terms", f"{fmt(asked)} is not {doc.discount_pct}% of the gross outstanding {fmt(gross_at_receipt)} at receipt ({fmt(expected)})",
"Allocate the receipt; the difference stays outstanding or goes to the tolerance"))
paid = sum((a.amount for a in rows if a.source_kind == "receipt" and a.source_id == payment_id), ZERO)
if paid != gross_at_receipt - expected:
refuse(unmet("discount_matches_terms", f"The receipt paid {fmt(paid)} of this invoice; the discounted amount is {fmt(gross_at_receipt - expected)}",
"Allocate the receipt to this invoice first, at the discounted amount"))
if stl.outstanding(doc.total, rows) < expected:
refuse(unmet("discount_matches_terms", f"Only {fmt(stl.outstanding(doc.total, rows))} is outstanding, less than the discount {fmt(expected)}",
"The invoice has been settled another way"))
repo.claims.take("discount", invoice_id, now, who, payment_id) # two receipts racing for one discount collide here
mode = repo.settings.current().settlement_discount_vat
if mode == "invoice_wording":
done, _ = uc._allocate_core(repo, caller, source_kind="settlement_discount", source_id=payment_id, source_currency=r.currency,
source_bp=r.bp_id, source_remainder=expected, lines=[LineIn(invoice_id=invoice_id, amount=expected)],
decided_how="manual", by=who, what="settlement discount")
return {"payment_id": payment_id, "invoice_id": invoice_id, "discount": f"{expected:.2f}", "mode": mode,
"allocations": done["allocations"], "invoices": done["invoices"]}
try:
got = seams.settlement_discount(invoice_id, payment_id, r.received_on, expected, who)
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
if not got["ok"]:
raise RuleBroken(f"Invoicing refused the discount ({got['gate']}): {got['reason']}. {got['fix']}".strip())
c = got["credit"]
return {"payment_id": payment_id, "invoice_id": invoice_id, "discount": f"{expected:.2f}", "mode": mode,
"credit": {k: c.get(k) for k in ("credit_id", "credit_no", "net", "tax", "total", "approved_by")}}
# ═══ PAY-17: remittance advice ═══════════════════════════════════════════════════════════════════════════
def _num(s: str) -> str:
return re.sub(r"[^a-z0-9]", "", s.lower())
def _payer_ok(a: str, b: str) -> bool:
x, y = _num(a), _num(b)
return not x or not y or x in y or y in x
def _link(repo, f: rem.RemitFile, payment_id: Optional[str], caller: dict) -> Optional[Receipt]:
"""The receipt this remittance explains: by amount, date and payer - exactly one candidate with its whole amount unallocated."""
cash = f.cash
if payment_id:
r = uc._receipt(repo, payment_id)
if r.amount != cash:
raise RuleBroken(f"Receipt {payment_id} is {fmt(r.amount)}; the remittance adds up to {fmt(cash)} (amounts less deductions). "
"They are not the same payment")
return r
out = []
for r in repo.receipts.search(status="received", limit=100000):
if r.amount != cash or (f.currency and r.currency != f.currency) or r.bp_id is None:
continue
if f.paid_on and abs((r.received_on - f.paid_on).days) > 5:
continue
if not _payer_ok(f.payer, r.remittance_text):
continue
if uc._remainder(r, repo.allocations.for_bp(r.bp_id)) != r.amount:
continue
out.append(r)
return out[0] if len(out) == 1 else None
def _import_one(repo, caller: dict, f: rem.RemitFile, payment_id: Optional[str]) -> dict:
rid, now, who = f.file_ref, uc._now(), uc._who(caller)
if repo.remittances.lines(rid):
return {"remittance_id": rid, "imported": False, "reason": "this remittance file was imported before: nothing was added"}
r = _link(repo, f, payment_id, caller)
by_ref, unknown, disputes = {}, [], []
if r is not None:
for d in seams.invoices(bp_id=r.bp_id, caller=caller):
if d.currency == r.currency:
by_ref[d.number.lower()] = d
by_ref[d.invoice_id.lower()] = d
proposed, rows = [], []
for n, ln in enumerate(f.lines, start=1):
d = by_ref.get(ln.invoice_ref.lower())
status = "imported"
if r is None:
status = "unlinked"
elif d is None:
status = "unknown_invoice"
unknown.append(ln.invoice_ref)
rows.append({"remittance_id": rid, "line_no": n, "file_ref": rid, "format": f.format, "payer": f.payer, "payment_id": r.payment_id if r else "",
"bp_id": r.bp_id if r else None, "invoice_ref": ln.invoice_ref, "invoice_id": d.invoice_id if d else "", "amount": ln.amount,
"currency": r.currency if r else f.currency, "deduction_reason": ln.reason, "status": status, "suggestion_id": "", "imported_by": who})
if d is not None and ln.net > ZERO:
proposed.append((n, d, ln))
sug_view = None
if r is not None and proposed:
by_inv = uc._group(repo.allocations.for_invoices([d.invoice_id for _, d, _ in proposed]))
lines = []
for n, d, ln in proposed:
if stl.outstanding(d.total, by_inv.get(d.invoice_id, [])) >= ln.net:
lines.append({"invoice_id": d.invoice_id, "amount": uc._s(ln.net)})
if lines:
cur = repo.suggestions.open_of(r.payment_id)
if cur:
repo.suggestions.move(cur["suggestion_id"], status="superseded")
sid = f"{r.payment_id}#{len(repo.suggestions.of_receipt(r.payment_id)) + 1}"
sug_view = {"suggestion_id": sid, "payment_id": r.payment_id, "proposed_lines": lines, "exact": False, "status": "open",
"confidence": "1.00" if not unknown else "0.90", "model": "remittance",
"evidence": {"remittance_id": rid, "format": f.format, "payer": f.payer, "invoices_named": len(f.lines),
"deductions": [{"invoice": x.invoice_ref, "amount": uc._s(x.deduction), "reason": x.reason} for x in f.lines if x.deduction],
"unknown_invoices": unknown}}
repo.suggestions.add(sug_view, now)
for row in rows:
if sug_view and any(pl["invoice_id"] == row["invoice_id"] for pl in sug_view["proposed_lines"]):
row["suggestion_id"] = sug_view["suggestion_id"]
repo.remittances.add(row, now)
if r is not None:
for n, d, ln in proposed: # a deduction is the customer's dispute with their reason - never a write-off
if ln.deduction > ZERO and uc.open_dispute(repo, r.bp_id, d.invoice_id,
f"{ln.reason} (remittance {rid}: {fmt(ln.deduction)} held back)", who) is not None:
disputes.append({"invoice_id": d.invoice_id, "amount": uc._s(ln.deduction), "reason": ln.reason})
return {"remittance_id": rid, "imported": True, "lines": len(rows), "payment_id": r.payment_id if r else "", "linked": r is not None,
"suggestion": sug_view, "unknown_invoices": unknown, "disputes_opened": disputes,
"note": "" if r is not None else "no single receipt matches this remittance's amount, date and payer: name it with payment_id"}
@PAY.use_case(kind="command", route="POST /remittances/import", permission=REMIT)
def import_remittance_file(caller: dict, format: Literal["csv", "camt054", "bai2"], content: text(5_000_000, min_length=1),
remittance_ref: Optional[Ref] = None, payer: Optional[text(200)] = None, payment_id: Optional[Ref] = None,
bank_account_id: Optional[Code] = None, bp_id: Optional[BigId] = None) -> dict:
"""Import a customer's remittance advice - CSV (invoice_ref, amount, deduction_amount, deduction_reason), ISO 20022 camt.054 with
structured remittance, or a BAI2 lockbox file. The lines are kept (keyed remittance_id + line_no, the id being the file's own
reference, so the same file again adds nothing) and linked to the receipt by amount, date and payer; a proposed allocation per
named invoice opens as a match suggestion with model "remittance" (not AI), which you accept, change or reject as in PAY-14; a
deduction becomes a dispute on its invoice with the customer's reason, never a write-off; an invoice number we do not know
is listed, not guessed. A BAI2 lockbox file's items are the bank's own record, so they create the receipts themselves
(payment_id lockbox:<account>:<batch>:<item>; bank_account_id required, bp_id optional)."""
with uc.repository() as repo, uc._decide():
if format == "bai2":
return _lockbox(repo, caller, content, bank_account_id, bp_id)
files = [rem.parse_csv(content, (remittance_ref or "").strip())] if format == "csv" else rem.parse_camt054(content)
if format == "csv":
if not files[0].file_ref:
raise RuleBroken("A CSV remittance needs its own reference (remittance_ref), so the same file again is recognised")
files[0].payer = payer or files[0].payer
out = [_import_one(repo, caller, f, payment_id if len(files) == 1 else None) for f in files]
return out[0] if len(out) == 1 else {"files": out}
def _lockbox(repo, caller: dict, content: str, bank_account_id: Optional[str], bp_id: Optional[int]) -> dict:
if not bank_account_id or repo.config.bank(bank_account_id) is None:
raise Refused(422, "A lockbox file needs bank_account_id: the account the cheques were deposited into")
batch, asof, items = rem.parse_bai2(content)
bank, today, now, who = repo.config.bank(bank_account_id), uc._today(repo), uc._now(), uc._who(caller)
settings = repo.settings.current()
ids = {it.item_ref: f"lockbox:{bank_account_id}:{batch}:{it.item_ref}" for it in items}
have = repo.receipts.existing(ids.values())
made, skipped = [], 0
for it in items:
pid = ids[it.item_ref]
if pid in have:
skipped += 1
continue
bp = bp_id or (settings.suspense_bp_id or None)
r = Receipt.record(payment_id=pid, bp_id=bp, currency=bank.get("currency") or "GBP", amount=it.amount, received_on=min(asof or today, today),
today=today, tz=settings.business_timezone, by=who, at=now, method="lockbox", bank_account_id=bank_account_id,
bank_reference=it.item_ref, remittance_text=it.text, origin="lockbox:bai2")
repo.receipts.add(r)
made.append(r)
posted = [uc._post_receipt(repo, r, actor=who) for r in made]
return {"imported": bool(made), "batch": batch, "created": len(made), "skipped_duplicate": skipped, "receipt_ids": [r.payment_id for r in made],
"unassigned": sum(1 for r in made if r.bp_id is None or r.bp_id == settings.suspense_bp_id),
"ledger": {"posted": sum(1 for p in posted if p.get("posted"))}}
@PAY.use_case(kind="command", route="POST /remittances/{remittance_id}/apply", permission=uc.ALLOC, idempotent=True)
def apply_remittance(caller: dict, remittance_id: Ref) -> dict:
"""Apply a remittance in one step: accept its suggestion as proposed (each line's invoice must still be open for at least the
amount), so the allocations are made with decided_how remittance and cite the suggestion and you. The deductions were already
opened as disputes at import. Applied once: asking again changes nothing."""
with uc.repository() as repo, uc._decide():
lines = repo.remittances.lines(remittance_id)
if not lines:
raise Refused(404, f"no remittance {remittance_id}")
pids = sorted({ln["payment_id"] for ln in lines if ln["payment_id"]})
if not pids:
raise RuleBroken(f"Remittance {remittance_id} is not linked to a receipt. Import it again naming the payment_id")
if all(ln["status"] != "imported" for ln in lines):
if any(ln["status"] == "rejected" for ln in lines):
raise RuleBroken(f"Remittance {remittance_id} was rejected and has no open suggestion. Allocate by hand")
return {"remittance_id": remittance_id, "status": lines[0]["status"], "unchanged": True}
r = uc._receipt(repo, pids[0])
sug = next((s for s in repo.suggestions.of_receipt(r.payment_id) if s["model"] == "remittance" and s["status"] == "open"
and s["evidence"].get("remittance_id") == remittance_id), None)
if sug is None:
raise RuleBroken(f"Remittance {remittance_id} has no open suggestion any more (it was decided or superseded). Allocate by hand")
done = uc._decide_core(repo, caller, r, sug, "accept", None)
return {"remittance_id": remittance_id, "status": "applied", **done}