dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/directdebit.py
"""Direct debit (PAY-18, PAY-19, PAY-20): mandates, runs, and the bank's reports back.
Guard keys (pay_claim), written with a plain INSERT by EVERY operation that touches the thing, in its own transaction:
'mandate' <sun>:<mandate_ref> - record, lodge, cancel, reject, collect, ADDACS, ARUDD, DDICA, dormant: a mandate is cancelled
while a run collects it -> one of the two is refused (409)
'ddrun' <run_id> - prepare, approve, notice, submit, collect, report: a submitted run is never rebuilt and a run
cannot be approved while it is being submitted
'ddinv' <invoice_id> - an invoice in a run: two runs cannot both collect it
'dd_report' <ref>:<line_no> - every report line is applied once
and a collection's money is the RECEIPT's: its returns go through the same reversal (and the same source claim) as any returned payment.
States never move back: pending -> active -> cancelled | dormant; draft -> approved -> submitted -> collected; a late or repeated
report line is reported as already applied or not applicable, never undone.
"""
import hashlib
import json
from datetime import date
from decimal import 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, Ref
from brick_payments.domain import directdebit as dd
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 refuse
from brick_payments.domain.receipt import Receipt
from brick_payments.infrastructure import seams
PAY = uc.PAY
MANDATE, DD_WRITE, DD_APPROVE = "pay:mandate:write", "pay:dd_run:write", "pay:dd_run:approve"
DORMANT_AFTER_DAYS = 13 * 31 # 13 months
def _s(v) -> Optional[str]:
return None if v is None else f"{Decimal(v):.2f}"
def _clean(row: dict) -> dict:
return {k: (v.isoformat() if isinstance(v, date) else _s(v) if isinstance(v, Decimal) else v) for k, v in row.items()
if k not in ("created_at", "updated_at")}
def _mandate_view(m: dict) -> dict:
return _clean(m) # the full bank details are in the dated details table and never in a view
def _find_mandate(repo, mandate_ref: str, sun: Optional[str] = None) -> dict:
found = [m for m in repo.dd.mandates(sun=sun) if m["mandate_ref"] == mandate_ref]
if not found:
raise Refused(404, f"no mandate {mandate_ref}")
if len(found) > 1:
raise RuleBroken(f"Mandate reference {mandate_ref} exists under more than one Service User Number. Name the sun")
return found[0]
def _mclaim(repo, m: dict, who: str, what: str) -> int:
return repo.claims.take("mandate", f"{m['sun']}:{m['mandate_ref']}", uc._now(), who, what)
def _run_claim(repo, run_id: str, who: str, what: str) -> int:
return repo.claims.take("ddrun", run_id, uc._now(), who, what)
def _house(repo, bank_account_id: str) -> tuple:
bank = repo.config.bank(bank_account_id)
if bank is None:
raise Refused(422, f"there is no bank account {bank_account_id} - set it up in the settings first")
purposes = [p.strip() for p in (bank.get("purposes") or "").split(",")]
if "dd_collect" not in purposes:
raise RuleBroken(f"Bank account {bank_account_id} does not collect direct debits (its purposes are {bank.get('purposes') or 'none'}). Add dd_collect")
house = repo.bank.house_bank(bank.get("house_bank_id") or "")
if house is None:
raise RuleBroken(f"{bank_account_id} has no house bank. Settings > Bank accounts")
return bank, house
# ═══ PAY-18: mandates ════════════════════════════════════════════════════════════════════════════════════
@PAY.use_case(kind="command", route="POST /dd-mandates", permission=MANDATE)
def record_dd_mandate(caller: dict, bp_id: BigId, mandate_ref: Ref, bank_account_id: Code, payer_sort_code: text(10),
payer_account_number: text(40), payer_account_name: text(120), payer_confirmed: bool,
scheme: Literal["bacs", "sepa_core"] = "bacs", notice_days: Optional[int] = None,
ends_on: Optional[text(10)] = None) -> dict:
"""Record a Direct Debit Instruction a customer completed: the reference (6-18 characters, unique per Service User Number), their
bank details (stored masked - the full number is held only for the Bacs file) and our house bank. It is `pending` until the next
submission lodges it with Bacs as AUDDIS 0N; with no AUDDIS rejection it is active. The payer must have been sent the DDI
confirmation and the Direct Debit Guarantee (payer_confirmed). notice_days is the 3 agreed in the DDI terms, else the default."""
if notice_days is not None and not (1 <= notice_days <= 30):
raise Refused(422, "notice_days is 1 to 30 working days")
with uc.repository() as repo, uc._decide():
bank, house = _house(repo, bank_account_id)
sun = house["sun"] if scheme == "bacs" else house["creditor_id"]
if not sun:
raise RuleBroken(f"{bank_account_id} has no Service User Number. Settings > Bank accounts")
if not dd.valid_mandate_ref(mandate_ref):
raise Refused(422, "A mandate reference is 6 to 18 letters, digits or - / . &, like DOD-7311-01")
if repo.dd.mandate(sun, mandate_ref) is not None:
raise RuleBroken(f"Mandate reference {mandate_ref} is already used for SUN {sun}. Use the next reference")
if not payer_confirmed:
raise RuleBroken("The payer must have been sent the DDI confirmation and the Direct Debit Guarantee. Send them, then record the mandate")
known = seams.check_customer(bp_id, caller)
if known.status != "known":
raise Refused(known.http, known.note)
sort = dd.valid_sort_code(payer_sort_code) if scheme == "bacs" else ""
number = dd.valid_account_number(payer_account_number) if scheme == "bacs" else payer_account_number.replace(" ", "").upper()
today, now = uc._today(repo), uc._now()
ends = uc._date(ends_on, "ends_on")
row = {"sun": sun, "mandate_ref": mandate_ref, "bp_id": bp_id, "bank_account_id": bank_account_id, "payer_sort_code": sort,
"payer_account_masked": dd.mask(number), "payer_account_name": payer_account_name, "scheme": scheme, "lifecycle": "pending",
"lodged_on": None, "auddis_result": "", "cancelled_reason": "", "cancel_lodged_on": None, "last_collected_on": None,
"recorded_on": today, "recorded_by": uc._who(caller), "notice_days": notice_days, "ends_on": ends, "disputed_on": None}
repo.dd.add_mandate(row, now)
repo.dd.add_details({"sun": sun, "mandate_ref": mandate_ref, "valid_from": today, "sort_code": sort, "account_number": number,
"source": "ddi"}, now)
_mclaim(repo, row, uc._who(caller), "record")
return {**_mandate_view(row), "next": "pending until the next submission lodges it (AUDDIS 0N)"}
@PAY.use_case(kind="command", route="POST /dd-mandates/{mandate_ref}/cancel", permission=MANDATE, idempotent=True)
def cancel_dd_mandate(caller: dict, mandate_ref: Ref, reason: text(500, min_length=1), sun: Optional[text(35)] = None) -> dict:
"""Cancel a mandate (a pending or active one) with a reason: the next submission sends AUDDIS 0C to Bacs, and it is never
collected again. Cancelled is final."""
with uc.repository() as repo, uc._decide():
m = _find_mandate(repo, mandate_ref, sun)
if m["lifecycle"] == "cancelled":
return {**_mandate_view(m), "unchanged": True}
if m["lifecycle"] not in ("pending", "active"):
raise RuleBroken(f"Mandate {mandate_ref} is {m['lifecycle']}; only a pending or active one is cancelled")
who = uc._who(caller)
_mclaim(repo, m, who, "cancel")
never_lodged = m["lodged_on"] is None
new = {**m, "lifecycle": "cancelled", "cancelled_reason": reason, "cancel_lodged_on": uc._today(repo) if never_lodged else None}
repo.dd.save_mandate(new)
return {**_mandate_view(new), "next": "nothing to send to Bacs: it was never lodged" if never_lodged else "AUDDIS 0C goes in the next submission"}
@PAY.use_case(kind="command", route="POST /dd-mandates/lodge", permission=DD_WRITE)
def lodge_dd_mandates(caller: dict, bank_account_id: Code, submission_ref: Ref) -> dict:
"""Lodge the pending instructions of a bank account with Bacs (AUDDIS 0N) and send the cancellations (0C) in one AUDDIS-only file,
kept as an object. With no AUDDIS rejection a lodged mandate is active - and only an active mandate is collected, so a new
customer's instruction goes out here (or rides in the next run's file). SEPA Core mandates are not lodged with a scheme: they are
active at once."""
with uc.repository() as repo, uc._decide():
bank, house = _house(repo, bank_account_id)
today, now, who = uc._today(repo), uc._now(), uc._who(caller)
sepa = bank["currency"] == "EUR"
sun = house["creditor_id"] if sepa else house["sun"]
pend = [m for m in repo.dd.mandates(sun=sun, lifecycle="pending") if m["bank_account_id"] == bank_account_id]
gone = [m for m in repo.dd.mandates(sun=sun, lifecycle="cancelled") if m["bank_account_id"] == bank_account_id and m["cancel_lodged_on"] is None]
if not pend and not gone:
return {"bank_account_id": bank_account_id, "lodged_0N": [], "cancelled_0C": [], "unchanged": True}
key, size = "", 0
if not sepa:
lines = [dd.FileLine("0N" if m in pend else "0C", *(lambda d: (d.get("sort_code", ""), d.get("account_number", "")))(
repo.dd.details_at(m["sun"], m["mandate_ref"], today) or {}), m["payer_account_name"], ZERO, m["mandate_ref"]) for m in pend + gone]
origin_account = (bank.get("account_ref") or "")[-8:]
if not house["sort_code"] or len(origin_account) != 8:
raise RuleBroken(f"The house bank needs a sort code and {bank_account_id} an 8-digit account number for the Bacs file. Settings > Bank accounts")
body = dd.bacs_standard18(sun=house["sun"], service_user_name=house["name"], origin_sort=dd.valid_sort_code(house["sort_code"]),
origin_account=origin_account, processing_on=today, lines=lines)
key, size = f"dd/auddis:{bank_account_id}:{today.isoformat()}:{submission_ref}", len(body)
repo.objects.put(key, body, "text/plain", hashlib.sha256(body.encode()).hexdigest(), now)
for m in pend + gone:
_mclaim(repo, m, who, "lodge" if m in pend else "send-0C")
for m in pend:
repo.dd.save_mandate({**m, "lifecycle": "active", "lodged_on": today})
for m in gone:
repo.dd.save_mandate({**m, "cancel_lodged_on": today})
return {"bank_account_id": bank_account_id, "file_key": key, "bytes": size, "submission_ref": submission_ref,
"lodged_0N": [m["mandate_ref"] for m in pend], "cancelled_0C": [m["mandate_ref"] for m in gone]}
@PAY.use_case(kind="command", route="POST /dd-mandates/{mandate_ref}/reject", permission=DD_WRITE, idempotent=True)
def reject_dd_mandate(caller: dict, mandate_ref: Ref, reason: text(200, min_length=1), sun: Optional[text(35)] = None) -> dict:
"""Bacs rejected the lodged instruction (AUDDIS, e.g. 'no account'): the reason is recorded and the mandate is cancelled - unusable.
Only a mandate lodged and not yet collected can be rejected."""
with uc.repository() as repo, uc._decide():
m = _find_mandate(repo, mandate_ref, sun)
if m["lifecycle"] == "cancelled":
return {**_mandate_view(m), "unchanged": True}
if m["lifecycle"] != "active" or m["lodged_on"] is None or m["last_collected_on"] is not None:
raise RuleBroken(f"Mandate {mandate_ref} is {m['lifecycle']}; an AUDDIS rejection applies to one lodged and not yet collected")
_mclaim(repo, m, uc._who(caller), "reject")
new = {**m, "lifecycle": "cancelled", "auddis_result": reason, "cancelled_reason": f"AUDDIS rejection: {reason}", "cancel_lodged_on": uc._today(repo)}
repo.dd.save_mandate(new)
return _mandate_view(new)
@PAY.use_case(kind="command", route="POST /dd-mandates/mark-dormant", permission=uc.DD_COLLECT, idempotent=True, tool=False)
def mark_mandate_dormant(caller: dict) -> dict:
"""The daily pass: an active mandate with no collection for 13 months is dormant and needs a new instruction. Final."""
with uc.repository() as repo, uc._decide():
today, marked = uc._today(repo), []
for m in repo.dd.mandates(lifecycle="active"):
last = m["last_collected_on"] or m["lodged_on"] or m["recorded_on"]
if last and (today - last).days > DORMANT_AFTER_DAYS:
_mclaim(repo, m, "system", "dormant")
repo.dd.save_mandate({**m, "lifecycle": "dormant", "cancelled_reason": "no collection for 13 months"})
marked.append(m["mandate_ref"])
return {"dormant": marked}
@PAY.use_case(kind="query", route="GET /dd-mandates", permission=uc.READ)
def list_dd_mandates(caller: dict, bp_id: Optional[BigId] = None, lifecycle: Optional[Literal["pending", "active", "cancelled", "dormant"]] = None) -> dict:
"""Direct Debit Instructions, bank details masked."""
with uc.repository() as repo:
return {"mandates": [_mandate_view(m) for m in repo.dd.mandates(bp_id=bp_id, lifecycle=lifecycle)]}
# ═══ PAY-19: runs ════════════════════════════════════════════════════════════════════════════════════════
def _run_view(repo, run: dict) -> dict:
cols = repo.dd.collections(run["run_id"])
return {**_clean(run), "collections": [{**_clean(c), "invoice_ids": json.loads(c["invoice_ids"] or "[]")} for c in cols]}
def _totals(cols: list) -> tuple:
live = [c for c in cols if c["status"] != "deferred"]
return sum((c["amount"] for c in live), ZERO), len(live)
def _run(repo, run_id: str) -> dict:
run = repo.dd.run(run_id)
if run is None:
raise Refused(404, f"no direct-debit run {run_id}")
return run
@PAY.use_case(kind="command", route="POST /dd-runs", permission=DD_WRITE)
def prepare_dd_run(caller: dict, bank_account_id: Code, collection_date: text(10)) -> dict:
"""Prepare a direct-debit run for a collection date (a working day after today): one run keyed dd:<bank_account_id>:<collection_date>
naming, per active mandate, each of the customer's invoices due on or before that date and still outstanding (no invoice is in two
runs). The transaction code is 01 for a first collection, 17 regular, 18 for the one re-presentation of an ARUDD 'refer to payer', 19 a final one.
A run already prepared - or submitted - is never rebuilt."""
with uc.repository() as repo, uc._decide():
bank, house = _house(repo, bank_account_id)
today, now, who = uc._today(repo), uc._now(), uc._who(caller)
on = uc._date(collection_date, "collection_date")
refuse(dd.collection_day_ok(on, today))
scheme = "sepa_core" if bank["currency"] == "EUR" else "bacs"
sun = house["sun"] if scheme == "bacs" else house["creditor_id"]
if not sun:
raise RuleBroken(f"{bank_account_id} has no {'Service User Number' if scheme == 'bacs' else 'SEPA creditor identifier'}. Settings > Bank accounts")
run_id = f"dd:{bank_account_id}:{on.isoformat()}"
have = repo.dd.run(run_id)
if have is not None:
raise RuleBroken(f"Run {run_id} is already {have['lifecycle']}. A run is never rebuilt: choose another collection date")
settings = repo.settings.current()
in_flight = repo.dd.invoices_in_flight()
collections, skipped = [], []
for m in repo.dd.mandates(sun=sun, lifecycle="active"):
if m["scheme"] != scheme or m["bank_account_id"] != bank_account_id:
continue
last = m["last_collected_on"] or m["lodged_on"] or m["recorded_on"]
if last and (today - last).days > DORMANT_AFTER_DAYS:
skipped.append({"mandate_ref": m["mandate_ref"], "reason": "no collection for 13 months: it needs a new instruction"})
continue
docs = [d for d in seams.invoices(bp_id=m["bp_id"], caller=caller) if d.currency == bank["currency"] and d.due_date and d.due_date <= on]
by = uc._group(repo.allocations.for_invoices([d.invoice_id for d in docs]))
take = [(d, stl.outstanding(d.total, by.get(d.invoice_id, []))) for d in sorted(docs, key=lambda d: (d.due_date, d.invoice_id))]
take = [(d, o) for d, o in take if o > ZERO and d.invoice_id not in in_flight]
if not take:
continue
past = [c for c in repo.dd.collections_of_mandate(m["sun"], m["mandate_ref"])]
represent = [c for c in past if c["return_reason_code"] == "0" and c["represented"] != "Y"]
code = dd.transaction_code(first=not any(c["status"] in ("collected", "returned") for c in past), represent=bool(represent),
final=bool(m["ends_on"] and on >= m["ends_on"]))
collections.append({"run_id": run_id, "mandate_ref": m["mandate_ref"], "bp_id": m["bp_id"], "sun": sun,
"invoice_ids": json.dumps([d.invoice_id for d, _ in take]), "amount": sum((o for _, o in take), ZERO),
"transaction_code": code, "notice_sent_on": None, "status": "planned", "payment_id": "",
"return_reason_code": "", "returned_on": None, "represented": "", "_old": represent,
"_notice": m["notice_days"] or settings.dd_notice_days})
if not collections:
raise RuleBroken("Nothing to collect: no active mandate has an invoice due on or before that date. Check the mandates and the invoices")
repo.dd.add_run({"run_id": run_id, "bank_account_id": bank_account_id, "collection_date": on, "input_day": dd.input_day_of(on),
"lifecycle": "draft", "prepared_by": (caller.get("email") or "").strip().lower(),
"prepared_by_sub": (caller.get("sub") or "").strip().lower(), "approved_by": "", "approved_by_sub": "",
"file_key": "", "submission_ref": "", "total": sum((c["amount"] for c in collections), ZERO),
"count": len(collections), "currency": bank["currency"]}, now)
_run_claim(repo, run_id, who, "prepare")
for c in collections:
old = c.pop("_old")
c.pop("_notice")
repo.dd.add_collection(c, now)
for iid in json.loads(c["invoice_ids"]):
repo.claims.take("ddinv", iid, now, who, run_id) # two runs racing for one invoice collide here
for o in old:
repo.dd.save_collection({**o, "represented": "Y"})
return {"run_id": run_id, "lifecycle": "draft", "collection_date": on.isoformat(), "input_day": dd.input_day_of(on).isoformat(),
"total": _s(sum((c["amount"] for c in collections), ZERO)), "count": len(collections), "currency": bank["currency"],
"collections": [{"mandate_ref": c["mandate_ref"], "bp_id": c["bp_id"], "amount": _s(c["amount"]), "transaction_code": c["transaction_code"],
"invoice_ids": json.loads(c["invoice_ids"])} for c in collections], "skipped_mandates": skipped}
@PAY.use_case(kind="command", route="POST /dd-runs/{run_id}/approve", permission=DD_APPROVE, idempotent=True)
def approve_dd_run(caller: dict, run_id: Ref) -> dict:
"""Approve a prepared run. The approver may not be the preparer, under EITHER identity (email or sub)."""
with uc.repository() as repo, uc._decide():
run = _run(repo, run_id)
if run["lifecycle"] != "draft":
return {**_clean(run), "unchanged": True} if run["lifecycle"] in ("approved", "submitted", "collected") else {**_clean(run)}
refuse(dd.second_person(run["prepared_by"], run["prepared_by_sub"], uc._ids(caller)))
_run_claim(repo, run_id, uc._who(caller), "approve")
repo.dd.save_run({**run, "lifecycle": "approved", "approved_by": (caller.get("email") or "").strip().lower(),
"approved_by_sub": (caller.get("sub") or "").strip().lower()})
return {**_clean(run), "lifecycle": "approved", "approved_by": uc._who(caller)}
@PAY.use_case(kind="command", route="POST /dd-runs/{run_id}/notices", permission=DD_WRITE, idempotent=True)
def send_dd_notices(caller: dict, run_id: Ref) -> dict:
"""Record the advance notices of an approved run as sent today. Each must go no later than the mandate's notice period (10 working
days by default, or the 3 agreed in the DDI terms) before the collection date; a collection whose notice is late is DEFERRED and
its invoices wait for the next run. (The delivery itself is the customer-communication channel's; this records that it went.)"""
with uc.repository() as repo, uc._decide():
run = _run(repo, run_id)
if run["lifecycle"] != "approved":
raise RuleBroken(f"Run {run_id} is {run['lifecycle']}; notices go for an approved run")
today, settings, who = uc._today(repo), repo.settings.current(), uc._who(caller)
_run_claim(repo, run_id, who, "notices")
sent, deferred, cols = [], [], repo.dd.collections(run_id)
for c in cols:
if c["status"] != "planned" or c["notice_sent_on"] is not None:
continue
m = repo.dd.mandate(c["sun"], c["mandate_ref"])
days = (m or {}).get("notice_days") or settings.dd_notice_days
gate = dd.advance_notice(today, run["collection_date"], days)
if gate.met:
c2 = {**c, "notice_sent_on": today}
sent.append(c["mandate_ref"])
else:
c2 = {**c, "status": "deferred"}
deferred.append({"mandate_ref": c["mandate_ref"], "reason": gate.reason, "invoice_ids": json.loads(c["invoice_ids"] or "[]")})
repo.dd.save_collection(c2)
cols = [c2 if x["mandate_ref"] == c["mandate_ref"] else x for x in cols]
total, count = _totals(cols)
repo.dd.save_run({**run, "total": total, "count": count})
return {"run_id": run_id, "notices_sent": sent, "deferred_to_next_run": deferred, "total": _s(total), "count": count}
def _file(repo, run: dict, cols: list, bank: dict, house: dict, lodge: list, cancel: list) -> tuple:
on = run["collection_date"]
lines = []
for m in lodge + cancel:
det = repo.dd.details_at(m["sun"], m["mandate_ref"], uc._today(repo))
lines.append(dd.FileLine("0N" if m in lodge else "0C", (det or {}).get("sort_code", ""), (det or {}).get("account_number", ""),
m["payer_account_name"], ZERO, m["mandate_ref"]))
for c in cols:
m = repo.dd.mandate(c["sun"], c["mandate_ref"])
det = repo.dd.details_at(c["sun"], c["mandate_ref"], on) or {}
lines.append(dd.FileLine(c["transaction_code"], det.get("sort_code", ""), det.get("account_number", ""), m["payer_account_name"],
c["amount"], c["mandate_ref"]))
if bank["currency"] == "EUR":
body = dd.pain008(message_id=run["run_id"], creditor_name=house["name"], creditor_id=house["creditor_id"], creditor_iban=bank.get("account_ref") or "",
bic=house["bic"], collection_date=on, currency="EUR", lines=[ln for ln in lines if ln.code not in ("0N", "0C")])
return body, "xml", "application/xml"
origin_account = (bank.get("account_ref") or "")[-8:]
if not house["sort_code"] or len(origin_account) != 8:
raise RuleBroken(f"The house bank needs a sort code and {bank['bank_account_id']} an 8-digit account number for the Bacs file. Settings > Bank accounts")
body = dd.bacs_standard18(sun=house["sun"], service_user_name=house["name"], origin_sort=dd.valid_sort_code(house["sort_code"]),
origin_account=origin_account, processing_on=run["input_day"], lines=lines)
return body, "bacs", "text/plain"
@PAY.use_case(kind="command", route="POST /dd-runs/{run_id}/submit", permission=DD_WRITE)
def submit_dd_run(caller: dict, run_id: Ref, submission_ref: Ref) -> dict:
"""Submit an approved run: the file (Bacs Standard 18 with the transaction codes, or SEPA pain.008 for a EUR account) is built and kept
as an object, with the AUDDIS instructions riding in it - 0N for mandates still pending (they become active unless Bacs rejects them),
0C for those we cancelled - and the run is FROZEN. Every collection's notice must have gone, and the file must go by the input day
(two working days before the collection date). A run already submitted is never rebuilt."""
with uc.repository() as repo, uc._decide():
run = _run(repo, run_id)
if run["lifecycle"] != "approved":
raise RuleBroken(f"Run {run_id} is {run['lifecycle']}; only an approved run is submitted, and a submitted run is never rebuilt")
today, now, who = uc._today(repo), uc._now(), uc._who(caller)
if today > run["input_day"]:
raise RuleBroken(f"The input day for {run['collection_date']} was {run['input_day']}; today is {today}. Prepare a run for a later collection date")
cols = [c for c in repo.dd.collections(run_id) if c["status"] == "planned"]
missing = [c["mandate_ref"] for c in cols if c["notice_sent_on"] is None]
if missing or not cols:
raise RuleBroken("Advance notice has not gone for " + (", ".join(missing) if missing else "any collection") + ". Send the notices first")
bank, house = _house(repo, run["bank_account_id"])
pend = [m for m in repo.dd.mandates(sun=house["sun"] if bank["currency"] != "EUR" else house["creditor_id"], lifecycle="pending")
if m["bank_account_id"] == run["bank_account_id"]]
gone = [m for m in repo.dd.mandates(sun=house["sun"] if bank["currency"] != "EUR" else house["creditor_id"], lifecycle="cancelled")
if m["bank_account_id"] == run["bank_account_id"] and m["cancel_lodged_on"] is None]
# one claim per mandate, shared with cancel/reject/collect: a cancel racing this submit is refused or sees the result, never both
_run_claim(repo, run_id, who, "submit")
held = {}
for m in pend:
held[(m["sun"], m["mandate_ref"])] = (m, "lodge")
for m in gone:
held[(m["sun"], m["mandate_ref"])] = (m, "send-0C")
for c in cols:
held.setdefault((c["sun"], c["mandate_ref"]), ({"sun": c["sun"], "mandate_ref": c["mandate_ref"]}, "submit"))
for m, what in held.values():
_mclaim(repo, m, who, what)
# re-read each collection's mandate under the claim: one cancelled, dormant or disputed since the approval is dropped, not collected
dropped, keep = [], []
for c in cols:
m = repo.dd.mandate(c["sun"], c["mandate_ref"])
if m is None or m["lifecycle"] not in ("active", "pending"):
repo.dd.save_collection({**c, "status": "skipped"})
dropped.append({"mandate_ref": c["mandate_ref"], "amount": _s(c["amount"]), "reason": f"the mandate is {(m or {}).get('lifecycle', 'gone')}"})
else:
keep.append(c)
if not keep:
raise RuleBroken("Every collection's mandate was cancelled or is no longer usable since the approval (" +
", ".join(d["mandate_ref"] for d in dropped) + "), so there is nothing to submit. Prepare a new run")
cols = keep
body, ext, ctype = _file(repo, run, cols, bank, house, pend, gone)
key = f"dd/{run_id}.{ext}"
repo.objects.put(key, body, ctype, hashlib.sha256(body.encode()).hexdigest(), now)
repo.dd.save_run({**run, "lifecycle": "submitted", "file_key": key, "submission_ref": submission_ref,
"total": sum((c["amount"] for c in cols), ZERO), "count": len(cols)})
for m in pend:
repo.dd.save_mandate({**m, "lifecycle": "active", "lodged_on": today})
for m in gone:
repo.dd.save_mandate({**m, "cancel_lodged_on": today})
return {"run_id": run_id, "lifecycle": "submitted", "file_key": key, "file_format": "Bacs Standard 18" if ext == "bacs" else "SEPA pain.008",
"bytes": len(body), "lodged_0N": [m["mandate_ref"] for m in pend], "cancelled_0C": [m["mandate_ref"] for m in gone],
"collections": len(cols), "dropped": dropped, "total": _s(sum((c["amount"] for c in cols), ZERO)), "submission_ref": submission_ref}
@PAY.use_case(kind="query", route="GET /dd-runs/{run_id}", permission=uc.READ)
def get_dd_run(caller: dict, run_id: Ref) -> dict:
"""A run, its collections and the object that holds its file."""
with uc.repository() as repo:
return _run_view(repo, _run(repo, run_id))
@PAY.use_case(kind="query", route="GET /dd-runs/{run_id}/file", permission=DD_WRITE)
def get_dd_run_file(caller: dict, run_id: Ref) -> dict:
"""The submitted file's content (the Bacs or pain.008 file), the only place the payer's full bank details appear."""
with uc.repository() as repo:
run = _run(repo, run_id)
body = repo.objects.get(run["file_key"]) if run["file_key"] else None
if body is None:
raise Refused(404, f"run {run_id} has no file yet: it is {run['lifecycle']}")
return {"run_id": run_id, "file_key": run["file_key"], "content": body}
@PAY.use_case(kind="command", route="POST /dd-runs/{run_id}/collect", permission=uc.DD_COLLECT, idempotent=True, tool=False)
def collect_dd_run(caller: dict, run_id: Ref) -> dict:
"""On the entry day (the collection date) each collection becomes a receipt keyed dd:<run>:<mandate>, dated the collection date and
allocated to its invoices with decided_how direct_debit. A mandate cancelled, dormant or disputed since is not collected. Collecting
twice adds nothing. Not before the collection date."""
with uc.repository() as repo, uc._decide():
run = _run(repo, run_id)
if run["lifecycle"] == "collected":
return {"run_id": run_id, "lifecycle": "collected", "collected": 0, "unchanged": True}
if run["lifecycle"] != "submitted":
raise RuleBroken(f"Run {run_id} is {run['lifecycle']}; only a submitted run is collected")
today, now, settings = uc._today(repo), uc._now(), repo.settings.current()
if today < run["collection_date"]:
raise RuleBroken(f"The collection date is {run['collection_date']}; today is {today}. A run is collected on its entry day")
_run_claim(repo, run_id, "system", "collect")
out, skipped = [], []
for c in [c for c in repo.dd.collections(run_id) if c["status"] == "planned"]:
m = repo.dd.mandate(c["sun"], c["mandate_ref"])
if m is None or m["lifecycle"] != "active":
repo.dd.save_collection({**c, "status": "skipped"})
skipped.append({"mandate_ref": c["mandate_ref"], "reason": f"the mandate is {(m or {}).get('lifecycle', 'gone')}"})
continue
_mclaim(repo, m, "system", "collect")
pid = f"dd:{run_id}:{c['mandate_ref']}"
r = Receipt.record(payment_id=pid, bp_id=c["bp_id"], currency=run["currency"], amount=c["amount"], received_on=run["collection_date"],
today=today, tz=settings.business_timezone, by="system:dd", at=now, method="direct_debit",
bank_account_id=run["bank_account_id"], bank_reference=pid, remittance_text=f"Direct debit {c['mandate_ref']}",
origin=f"dd_run:{run_id}")
repo.receipts.add(r)
repo.receipts.claim_ref(run["bank_account_id"], pid, pid, now)
ids = json.loads(c["invoice_ids"] or "[]")
docs = {d.invoice_id: d for d in seams.invoices(ids=ids, caller=caller)}
by = uc._group(repo.allocations.for_invoices(ids))
left, lines = c["amount"], []
for iid in ids:
d = docs.get(iid)
if d is None:
continue
take = min(stl.outstanding(d.total, by.get(iid, [])), left)
if take > ZERO:
lines.append(LineIn(invoice_id=iid, amount=take))
left -= take
allocated = []
if lines:
done, _ = uc._allocate_core(repo, {"email": "system:dd"}, source_kind="receipt", source_id=pid, source_currency=run["currency"],
source_bp=c["bp_id"], source_remainder=c["amount"], lines=lines, decided_how="direct_debit",
by="system:dd")
allocated = done["allocations"]
uc._post_receipt(repo, r, actor="system:dd")
repo.dd.save_collection({**c, "status": "collected", "payment_id": pid})
repo.dd.save_mandate({**m, "last_collected_on": run["collection_date"]})
out.append({"mandate_ref": c["mandate_ref"], "payment_id": pid, "amount": _s(c["amount"]), "allocations": len(allocated)})
repo.dd.save_run({**run, "lifecycle": "collected"})
return {"run_id": run_id, "lifecycle": "collected", "collected": len(out), "receipts": out, "skipped": skipped}
# ═══ PAY-20: the bank's reports ═════════════════════════════════════════════════════════════════════════
def _collection_of(repo, rep: dd.Report, m: dict) -> Optional[dict]:
cols = [c for c in repo.dd.collections_of_mandate(m["sun"], m["mandate_ref"]) if (not rep.run_ref or c["run_id"] == rep.run_ref)]
cols = sorted(cols, key=lambda c: c["run_id"])
live = [c for c in cols if c["status"] == "collected"]
return live[-1] if live else (cols[-1] if cols else None)
def _return(repo, caller: dict, c: dict, rep: dd.Report, ref: str, code: str) -> dict:
"""Reverse the collection's receipt by PAY-07 (dated by the report, the reason recorded): the invoices are owed again."""
r = uc._receipt(repo, c["payment_id"])
done = uc.reverse_core(repo, caller, r, ref, rep.on, reason_code=code)
repo.dd.save_collection({**c, "status": "returned", "return_reason_code": code, "returned_on": rep.on})
return done
@PAY.use_case(kind="command", route="POST /dd-reports/import", permission=DD_WRITE)
def import_dd_reports(caller: dict, kind: Literal["arudd", "addacs", "ddica"], report_ref: Ref, content: text(1_000_000, min_length=1)) -> dict:
"""Import the bank's reports on our direct debits (CSV as the portal exports them). ARUDD: each returned collection reverses its
receipt by PAY-07 (dated by the report, the reason recorded) so its invoices are owed again; reason 0 (refer to payer) may be
re-presented once (code 18) in a later run, reasons that mean the instruction is gone cancel the mandate. ADDACS: 'C' changes the
mandate's bank details from the advice's date (a new dated row), any other code cancels. DDICA: an indemnity claim reverses
the collection and marks the mandate disputed. Every report line is applied ONCE (keyed report_ref + line): the same report
again, or a repeated line, changes nothing; a line for a mandate already cancelled or a collection already returned is reported, not undone."""
with uc.repository() as repo, uc._decide():
lines = dd.parse_report(kind, content)
who, now, today = uc._who(caller), uc._now(), uc._today(repo)
results = []
for rep in lines:
key = f"{report_ref}:{rep.line_no}"
if repo.claims.count("dd_report", key):
results.append({"line": rep.line_no, "mandate_ref": rep.mandate_ref, "result": "already applied"})
continue
try:
m = _find_mandate(repo, rep.mandate_ref)
except (Refused, RuleBroken) as e:
results.append({"line": rep.line_no, "mandate_ref": rep.mandate_ref, "result": f"not applied: {e}"})
continue
repo.claims.take("dd_report", key, now, who, kind)
_mclaim(repo, m, who, kind)
out = _apply(repo, caller, kind, rep, m, report_ref, today)
results.append({"line": rep.line_no, "mandate_ref": rep.mandate_ref, **out})
return {"kind": kind, "report_ref": report_ref, "lines": len(lines), "results": results}
def _apply(repo, caller: dict, kind: str, rep: dd.Report, m: dict, report_ref: str, today: date) -> dict:
ref = f"{kind}:{report_ref}:{rep.line_no}"
final = m["lifecycle"] in ("cancelled", "dormant")
if kind == "addacs":
if rep.code in dd.ADDACS_CHANGE:
if final:
return {"result": f"not applied: the mandate is {m['lifecycle']}, which is final"}
sort, number = dd.valid_sort_code(rep.sort_code), dd.valid_account_number(rep.account_number)
repo.dd.add_details({"sun": m["sun"], "mandate_ref": m["mandate_ref"], "valid_from": rep.on, "sort_code": sort, "account_number": number,
"source": ref}, uc._now())
repo.dd.save_mandate({**m, "payer_sort_code": sort, "payer_account_masked": dd.mask(number)} if rep.on <= today else m)
return {"result": "bank details changed", "effective_from": rep.on.isoformat()}
if final:
return {"result": f"not applied: the mandate is {m['lifecycle']}, which is final"}
repo.dd.save_mandate({**m, "lifecycle": "cancelled", "cancelled_reason": f"ADDACS {rep.code}", "cancel_lodged_on": today})
return {"result": "mandate cancelled", "code": rep.code}
c = _collection_of(repo, rep, m)
if c is None or c["status"] not in ("collected", "returned"):
return {"result": "not applied: that collection has not been collected yet (nothing to return)"}
if c["status"] == "returned":
return {"result": "not applied: that collection was already returned"}
if kind == "arudd":
done = _return(repo, caller, c, rep, ref, rep.code)
words = dd.ARUDD_WORDS.get(rep.code, "returned")
if rep.code in dd.ARUDD_CANCEL and not final:
repo.dd.save_mandate({**m, "lifecycle": "cancelled", "cancelled_reason": f"ARUDD {rep.code}: {words}", "cancel_lodged_on": today})
return {"result": "returned", "payment_id": c["payment_id"], "reason": f"{rep.code} {words}", "reversed_allocations": len(done["allocation_reversals"]),
"represent": "may be re-presented once (code 18)" if rep.code in dd.ARUDD_REPRESENT else "",
"mandate": "cancelled" if rep.code in dd.ARUDD_CANCEL and not final else m["lifecycle"]}
done = _return(repo, caller, c, rep, ref, "DDICA") # ddica
repo.dd.save_mandate({**m, "lifecycle": "cancelled" if not final else m["lifecycle"], "disputed_on": rep.on,
"cancelled_reason": m["cancelled_reason"] or "DDICA indemnity claim", "cancel_lodged_on": m["cancel_lodged_on"] or today})
return {"result": "returned", "payment_id": c["payment_id"], "reason": "DDICA indemnity claim", "mandate": "disputed",
"reversed_allocations": len(done["allocation_reversals"])}