dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/domain/files.py
"""The bank files a run is written as: ISO 20022 pain.001 (03 and 09), BACS Standard 18 and CSV (PUR-29, PUR-59).
Pure functions of the payments handed in; the bytes are kept once as an object by the use case. Every BACS detail record is exactly 100
characters and the trailer totals equal the sum of the detail records in pence (module.yaml invariants)."""
import csv
import io
import re
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from xml.sax.saxutils import escape
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.readiness import no, ok
@dataclass
class Payee:
payment_id: str
name: str
account: str # IBAN, or '<sort code> <account>' for BACS
bic_or_sort: str
amount: Decimal
currency: str
reference: str
country: str = ""
def pence(amount: Decimal) -> int:
return int((amount * 100).to_integral_value())
# ── BACS Standard 18 ────────────────────────────────────────────────────────────────────────────────
def split_uk(account: str, sort: str = "") -> Optional[tuple]:
"""(sort code 6 digits, account 8 digits) from '12-34-56 12345678' / '123456 12345678' / (sort, account) — or None."""
raw = f"{sort} {account}"
digits = re.sub(r"[^0-9 ]", "", raw.replace("-", ""))
parts = digits.split()
if len(parts) >= 2 and len(parts[0]) == 6 and len(parts[1]) == 8:
return parts[0], parts[1]
one = re.sub(r"\D", "", raw)
if len(one) == 14:
return one[:6], one[6:]
return None
def bacs_earliest(today: date, cycle_working_days: int) -> date:
"""The earliest BACS pay date: `cycle` working days after today (input, processing, entry)."""
d, n = today, 0
while n < cycle_working_days:
d = date.fromordinal(d.toordinal() + 1)
if d.weekday() < 5:
n += 1
return d
def bacs_ready(payees: list, currency: str, service_user: str, pay_date: date, today: date, cycle: int, originator: Optional[str] = None) -> list:
gates = []
why = []
if currency != "GBP":
why.append(f"the run is in {currency}, not GBP")
if not re.fullmatch(r"\d{6}", service_user or ""):
why.append("our BACS service user number is not set")
if originator is not None and split_uk(originator) is None:
why.append("our own paying account in payments has no UK sort code and account number")
for p in payees:
if split_uk(p.account, p.bic_or_sort) is None:
why.append(f"{p.name} has no UK sort code and 8-digit account")
earliest = bacs_earliest(today, cycle)
if pay_date < earliest:
why.append(f"the pay date is inside the BACS cycle (earliest {earliest.isoformat()})")
gates.append(ok("bacs_ready") if not why else no("bacs_ready", "; ".join(why),
"fix the payee, pick pain.001, or move the pay date to the earliest BACS date shown"))
return gates
def _f(s: str, n: int) -> str:
s = re.sub(r"[^A-Za-z0-9 .,&/'\-]", " ", s or "").upper()
return s[:n].ljust(n)
def _julian(d: date) -> str:
return f" {d:%y}{d.timetuple().tm_yday:03d}"
def bacs_std18(payees: list, *, service_user: str, user_name: str, originator: str, pay_date: date, today: date, run_ref: str,
file_no: int = 1) -> str:
"""One Standard 18 file: VOL1, HDR1, HDR2, UHL1; a 100-character detail record per payment (payee sort code and account,
transaction code 99, our sort code and account, the amount in pence, our reference, the payee's name); the contra record; EOF1,
EOF2 and UTL1 (the totals and the count)."""
mine = split_uk(originator)
if mine is None:
raise RuleBroken("bacs_ready: our own sort code and account are not on the run")
lines = []
ser = f"{file_no:06d}"
lines.append(("VOL1" + ser + " " + " " * 31 + _f(user_name, 10) + " " * 28 + "1").ljust(80)[:80])
hdr1 = "HDR1A" + service_user + "S" + ser + "00010001" + " " + f"{_julian(today)} 99365 000000"
lines.append(hdr1.ljust(80)[:80])
lines.append(("HDR2F" + "02000" + "00100" + " " * 35 + "00").ljust(80)[:80])
lines.append(("UHL1 " + _julian(pay_date)[1:].rjust(5) + " 999999 000000 000001 DAILY 001").ljust(80)[:80])
total = 0
for p in payees:
sc, acc = split_uk(p.account, p.bic_or_sort)
amt = pence(p.amount)
total += amt
rec = sc + acc + "0" + "99" + mine[0] + mine[1] + " " + f"{amt:011d}" + _f(user_name, 18) + _f(p.reference, 18) + _f(p.name, 18)
lines.append(rec)
contra = mine[0] + mine[1] + "0" + "99" + mine[0] + mine[1] + " " + f"{total:011d}" + _f(user_name, 18) + _f(run_ref, 18) + _f("CONTRA", 18)
lines.append(contra)
lines.append(("EOF1" + hdr1[4:]).ljust(80)[:80])
lines.append(("EOF2" + "F" + "02000" + "00100" + " " * 35 + "00").ljust(80)[:80])
lines.append(f"UTL1{total:013d}{total:013d}{1:07d}{len(payees):07d}".ljust(80)[:80])
for rec in lines:
if rec[:4] not in ("VOL1", "HDR1", "HDR2", "UHL1", "EOF1", "EOF2", "UTL1"):
assert len(rec) == 100, len(rec)
return "\n".join(lines) + "\n"
def bacs_totals(text: str) -> tuple:
"""(sum of the detail records' pence, the trailer's credit total, the count) read back from a Standard 18 file."""
details = [ln for ln in text.splitlines() if ln[:4] not in ("VOL1", "HDR1", "HDR2", "UHL1", "EOF1", "EOF2", "UTL1")]
utl = next(ln for ln in text.splitlines() if ln.startswith("UTL1"))
return sum(int(d[35:46]) for d in details[:-1]), int(utl[17:30]), int(utl[37:44])
# ── ISO 20022 pain.001 ─────────────────────────────────────────────────────────────────────────────
def pain001(payees: list, *, version: str, run_id: str, debtor_name: str, debtor_iban: str, debtor_bic: str, pay_date: date,
created: datetime, currency: str) -> str:
"""pain.001.001.03 or .09 (Customer Credit Transfer Initiation): one PmtInf from our account, one CdtTrfTxInf per payment."""
ns = {"pain.001.001.03": "urn:iso:std:iso:20022:tech:xsd:pain.001.001.03", "pain.001.001.09": "urn:iso:std:iso:20022:tech:xsd:pain.001.001.09"}[version]
ctrl = sum((p.amount for p in payees), Decimal(0))
n = len(payees)
nine = version.endswith(".09")
out = ['<?xml version="1.0" encoding="UTF-8"?>', f'<Document xmlns="{ns}">', "<CstmrCdtTrfInitn>", "<GrpHdr>",
f"<MsgId>{escape(run_id)[:35]}</MsgId>", f"<CreDtTm>{created.strftime('%Y-%m-%dT%H:%M:%S')}</CreDtTm>", f"<NbOfTxs>{n}</NbOfTxs>",
f"<CtrlSum>{ctrl}</CtrlSum>", f"<InitgPty><Nm>{escape(debtor_name)}</Nm></InitgPty>", "</GrpHdr>", "<PmtInf>",
f"<PmtInfId>{escape(run_id)[:35]}</PmtInfId>", "<PmtMtd>TRF</PmtMtd>", f"<NbOfTxs>{n}</NbOfTxs>", f"<CtrlSum>{ctrl}</CtrlSum>",
(f"<ReqdExctnDt><Dt>{pay_date.isoformat()}</Dt></ReqdExctnDt>" if nine else f"<ReqdExctnDt>{pay_date.isoformat()}</ReqdExctnDt>"),
f"<Dbtr><Nm>{escape(debtor_name)}</Nm></Dbtr>", f"<DbtrAcct><Id><IBAN>{escape(debtor_iban)}</IBAN></Id></DbtrAcct>",
(f"<DbtrAgt><FinInstnId><BICFI>{escape(debtor_bic)}</BICFI></FinInstnId></DbtrAgt>" if nine else
f"<DbtrAgt><FinInstnId><BIC>{escape(debtor_bic)}</BIC></FinInstnId></DbtrAgt>")]
for p in payees:
bic = (f"<CdtrAgt><FinInstnId><BICFI>{escape(p.bic_or_sort)}</BICFI></FinInstnId></CdtrAgt>" if nine else
f"<CdtrAgt><FinInstnId><BIC>{escape(p.bic_or_sort)}</BIC></FinInstnId></CdtrAgt>") if p.bic_or_sort else ""
out += ["<CdtTrfTxInf>", f"<PmtId><EndToEndId>{escape(p.reference)[:35]}</EndToEndId></PmtId>",
f'<Amt><InstdAmt Ccy="{escape(p.currency)}">{p.amount}</InstdAmt></Amt>', bic, f"<Cdtr><Nm>{escape(p.name)}</Nm></Cdtr>",
f"<CdtrAcct><Id><IBAN>{escape(p.account.replace(' ', ''))}</IBAN></Id></CdtrAcct>",
f"<RmtInf><Ustrd>{escape(p.reference)}</Ustrd></RmtInf>", "</CdtTrfTxInf>"]
out += ["</PmtInf>", "</CstmrCdtTrfInitn>", "</Document>"]
return "\n".join(x for x in out if x) + "\n"
def csv_file(payees: list, *, pay_date: date) -> str:
buf = io.StringIO()
w = csv.writer(buf, lineterminator="\n")
w.writerow(["payment_id", "payee", "account", "bic_or_sort_code", "amount", "currency", "reference", "pay_date"])
for p in payees:
w.writerow([p.payment_id, p.name, p.account.replace(" ", ""), p.bic_or_sort, p.amount, p.currency, p.reference, pay_date.isoformat()])
return buf.getvalue()
EXT = {"pain.001.001.03": "xml", "pain.001.001.09": "xml", "bacs_std18": "std18", "csv": "csv"}