dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/domain/directdebit.py
"""Direct debit (Bacs / SEPA Core), pure: working days, the mandate and run rules, the Bacs Standard 18 and pain.008 files, and the
ARUDD / ADDACS / DDICA report lines. No storage, no clock - the use cases hand in the dates.
Working days are Monday to Friday except the bank holidays of England and Wales (below). A mandate is `pending` until the next
submission lodges it (AUDDIS 0N), `active` when no rejection came, `cancelled` (AUDDIS 0C when WE cancel; a rejection or an
ARUDD / ADDACS code when the bank says so) or `dormant` (no collection for 13 months). Those are final: a state never moves back.
"""
import csv
import io
from dataclasses import dataclass
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from xml.sax.saxutils import escape
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.readiness import Gate, met, unmet
# ── working days ────────────────────────────────────────────────────────────────────────────────────────
def _easter(year: int) -> date:
a, b, c = year % 19, year // 100, year % 100
d, e = b // 4, b % 4
g = (8 * b + 13) // 25
h = (19 * a + b - d - g + 15) % 30
i, k = c // 4, c % 4
el = (32 + 2 * e + 2 * i - h - k) % 7
m = (a + 11 * h + 19 * el) // 433
month, day = divmod(h + el - 7 * m + 114, 31)
return date(year, month, day + 1)
def _substitute(d: date, taken: set) -> date:
while d.weekday() >= 5 or d in taken:
d += timedelta(days=1)
return d
def bank_holidays(year: int) -> set:
"""England and Wales."""
easter = _easter(year)
out = {easter - timedelta(days=2), easter + timedelta(days=1)} # Good Friday, Easter Monday
out.add(_substitute(date(year, 1, 1), out))
may1 = date(year, 5, 1)
out.add(may1 + timedelta(days=(7 - may1.weekday()) % 7)) # early May: first Monday
last_may = date(year, 5, 31)
out.add(last_may - timedelta(days=last_may.weekday())) # spring: last Monday of May
last_aug = date(year, 8, 31)
out.add(last_aug - timedelta(days=last_aug.weekday())) # summer: last Monday of August
xmas = _substitute(date(year, 12, 25), out)
out.add(xmas)
out.add(_substitute(date(year, 12, 26), out | {xmas}))
return out
def is_working_day(d: date) -> bool:
return d.weekday() < 5 and d not in bank_holidays(d.year)
def add_working_days(d: date, n: int) -> date:
step = 1 if n >= 0 else -1
left = abs(n)
while left:
d += timedelta(days=step)
if is_working_day(d):
left -= 1
return d
def working_days_between(a: date, b: date) -> int:
"""Working days after `a` up to and including `b` (negative when b is before a)."""
if b < a:
return -working_days_between(b, a)
n, d = 0, a
while d < b:
d += timedelta(days=1)
n += is_working_day(d)
return n
# ── gates ────────────────────────────────────────────────────────────────────────────────────────────────
def collection_day_ok(collection_date: date, today: date) -> Gate:
if not is_working_day(collection_date):
return unmet("working_day", f"{collection_date} is not a working day", "Choose the next working day")
if collection_date <= today:
return unmet("working_day", f"{collection_date} is not after today ({today})", "Choose a collection date in the future")
return met("working_day")
def advance_notice(today: date, collection_date: date, notice_days: int) -> Gate:
"""The notice goes no later than `notice_days` working days before the collection date."""
last = add_working_days(collection_date, -notice_days)
if today > last:
return unmet("advance_notice", f"Notice sent {today}; {notice_days} working days before {collection_date} was {last}",
"The invoice waits for the next run")
return met("advance_notice")
def input_day_of(collection_date: date) -> date:
"""Bacs: the file is submitted by the input day, two working days before the entry (collection) day."""
return add_working_days(collection_date, -2)
def second_person(prepared_by: str, prepared_sub: str, approver_ids: tuple) -> Gate:
theirs = {x.strip().lower() for x in (prepared_by, prepared_sub) if x and x.strip()}
mine = {x.strip().lower() for x in approver_ids if x and x.strip()}
if theirs & mine:
return unmet("second_person", "You prepared this run", "Another approver approves it")
return met("second_person")
def valid_mandate_ref(ref: str) -> bool:
return 6 <= len(ref) <= 18 and ref.isascii() and all(c.isalnum() or c in "-/.&" for c in ref)
def valid_sort_code(v: str) -> str:
digits = "".join(ch for ch in v if ch.isdigit())
if len(digits) != 6:
raise RuleBroken("A sort code has six digits, like 20-00-00")
return digits
def valid_account_number(v: str) -> str:
digits = "".join(ch for ch in v if ch.isdigit())
if len(digits) != 8:
raise RuleBroken("A UK account number has eight digits")
return digits
def mask(v: str) -> str:
return "*" * max(0, len(v) - 4) + v[-4:]
# ── ARUDD / ADDACS / DDICA codes ──────────────────────────────────────────────────────────────────────────
ARUDD_REPRESENT = {"0"} # refer to payer: may be re-presented ONCE (code 18)
ARUDD_CANCEL = {"1", "2", "3", "5", "6", "A", "B"} # the instruction is gone: the mandate is cancelled
ARUDD_WORDS = {"0": "refer to payer", "1": "instruction cancelled", "2": "payer deceased", "3": "account transferred",
"5": "no account / wrong account type", "6": "no instruction", "7": "amount differs", "8": "amount not yet due",
"9": "presentation overdue", "A": "service user differs", "B": "account closed"}
ADDACS_CHANGE = {"C"} # account transferred to new details: details change from its date
def transaction_code(*, first: bool, represent: bool, final: bool) -> str:
"""01 first collection, 17 regular, 18 re-presentation, 19 final."""
return "18" if represent else "19" if final else "01" if first else "17"
@dataclass(frozen=True)
class Report:
kind: str # arudd | addacs | ddica
line_no: int
mandate_ref: str
code: str
on: date
run_ref: str = ""
sort_code: str = ""
account_number: str = ""
def parse_report(kind: str, content: str) -> list:
"""The reports as the bank portal exports them, a CSV with a header.
arudd: mandate_ref, reason_code, returned_on [, run_ref]; addacs: mandate_ref, reason_code, effective_on [, new_sort_code,
new_account_number]; ddica: mandate_ref, claimed_on [, run_ref]."""
out = []
reader = csv.DictReader(io.StringIO(content))
need = {"arudd": ("mandate_ref", "reason_code", "returned_on"), "addacs": ("mandate_ref", "reason_code", "effective_on"),
"ddica": ("mandate_ref", "claimed_on")}[kind]
names = [h.strip().lower() for h in (reader.fieldnames or [])]
for col in need:
if col not in names:
raise RuleBroken(f"The {kind.upper()} report needs a column {col}. Export it again with the header row")
for n, row in enumerate(reader, start=1):
row = {k.strip().lower(): (v or "").strip() for k, v in row.items() if k}
try:
when = date.fromisoformat((row.get("returned_on") or row.get("effective_on") or row.get("claimed_on"))[:10])
except (ValueError, TypeError):
raise RuleBroken(f"Line {n} of the {kind.upper()} report has a date that cannot be read (2026-12-01)") from None
if not row["mandate_ref"]:
raise RuleBroken(f"Line {n} of the {kind.upper()} report names no mandate")
out.append(Report(kind, n, row["mandate_ref"], row.get("reason_code", ""), when, row.get("run_ref", ""),
row.get("new_sort_code", ""), row.get("new_account_number", "")))
if not out:
raise RuleBroken(f"The {kind.upper()} report has no lines")
return out
# ── the files ────────────────────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class FileLine:
code: str # 01 | 17 | 18 | 19 | 0N | 0C
sort_code: str
account_number: str
account_name: str
amount: Decimal
reference: str
def _pad(v: str, n: int) -> str:
return (v or "")[:n].upper().ljust(n)
def bacs_standard18(*, sun: str, service_user_name: str, origin_sort: str, origin_account: str, processing_on: date,
lines: list) -> str:
"""Bacs Standard 18: a UHL1 header (SUN, processing day), one 100-character record per line (destination sort code and account,
'0', transaction code, originating sort code and account, 4 blank, amount in pence, service user name, reference, destination
name) and a UTL1 trailer with the count and the total. AUDDIS instructions (0N lodge, 0C cancel) travel as records with amount 0."""
if len(sun) != 6 or not sun.isdigit():
raise RuleBroken(f"The Service User Number {sun!r} is not six digits. Fix it in the house bank")
out = [f"UHL1 {processing_on:%y%j} {sun} {_pad(service_user_name, 18)}"]
total, count = 0, 0
for ln in lines:
pence = int((ln.amount * 100).to_integral_value())
out.append(f"{ln.sort_code:6}{ln.account_number:8}0{ln.code:2}{origin_sort:6}{origin_account:8}{'':4}{pence:011d}"
f"{_pad(service_user_name, 18)}{_pad(ln.reference, 18)}{_pad(ln.account_name, 18)}")
count += ln.code not in ("0N", "0C")
total += pence if ln.code not in ("0N", "0C") else 0
out.append(f"UTL1 {total:013d} {count:07d} {len(lines):07d}")
return "\n".join(out) + "\n"
def pain008(*, message_id: str, creditor_name: str, creditor_id: str, creditor_iban: str, bic: str, collection_date: date,
currency: str, lines: list) -> str:
"""SEPA Core direct debit initiation (pain.008.001.02): one PmtInf, one DrctDbtTxInf per collection."""
if not creditor_id:
raise RuleBroken("The house bank has no SEPA creditor identifier. Fix it in the house bank")
total = sum((ln.amount for ln in lines), Decimal("0.00"))
tx = "".join(
f"<DrctDbtTxInf><PmtId><EndToEndId>{escape(ln.reference)}</EndToEndId></PmtId>"
f"<InstdAmt Ccy=\"{currency}\">{ln.amount:.2f}</InstdAmt><DrctDbtTx><MndtRltdInf><MndtId>{escape(ln.reference)}</MndtId>"
f"</MndtRltdInf></DrctDbtTx><Dbtr><Nm>{escape(ln.account_name)}</Nm></Dbtr><DbtrAcct><Id><IBAN>{escape(ln.account_number)}"
f"</IBAN></Id></DbtrAcct></DrctDbtTxInf>" for ln in lines)
return ("<?xml version=\"1.0\" encoding=\"UTF-8\"?><Document xmlns=\"urn:iso:std:iso:20022:tech:xsd:pain.008.001.02\">"
f"<CstmrDrctDbtInitn><GrpHdr><MsgId>{escape(message_id)}</MsgId><NbOfTxs>{len(lines)}</NbOfTxs><CtrlSum>{total:.2f}</CtrlSum>"
f"<InitgPty><Nm>{escape(creditor_name)}</Nm></InitgPty></GrpHdr><PmtInf><PmtInfId>{escape(message_id)}</PmtInfId>"
f"<PmtMtd>DD</PmtMtd><PmtTpInf><SvcLvl><Cd>SEPA</Cd></SvcLvl><LclInstrm><Cd>CORE</Cd></LclInstrm></PmtTpInf>"
f"<ReqdColltnDt>{collection_date.isoformat()}</ReqdColltnDt><Cdtr><Nm>{escape(creditor_name)}</Nm></Cdtr>"
f"<CdtrAcct><Id><IBAN>{escape(creditor_iban)}</IBAN></Id></CdtrAcct><CdtrAgt><FinInstnId><BIC>{escape(bic)}</BIC></FinInstnId></CdtrAgt>"
f"<CdtrSchmeId><Id><PrvtId><Othr><Id>{escape(creditor_id)}</Id></Othr></PrvtId></Id></CdtrSchmeId>{tx}</PmtInf>"
"</CstmrDrctDbtInitn></Document>")
def due_for(collection_date: date, due_dates: Optional[date]) -> bool:
return due_dates is not None and due_dates <= collection_date