dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/domain/remittance.py
"""Remittance advice (PAY-17), pure parsing: a customer's own list of the invoices a payment settles, in CSV, ISO 20022 camt.054
(structured remittance) or a BAI2 lockbox file. Nothing here touches storage or decides anything.
Cash = SUM(amount - deduction) over the lines. A deduction is what the customer held back from an invoice, with THEIR reason: it
becomes a dispute on that invoice, never a write-off."""
import csv
import io
import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Optional
from brick_payments.application.inputs import _money
from brick_payments.domain.bankfile import _find, _local, _text
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO
@dataclass(frozen=True)
class RLine:
invoice_ref: str
amount: Decimal
deduction: Decimal = ZERO
reason: str = ""
@property
def net(self) -> Decimal:
return self.amount - self.deduction
@dataclass
class RemitFile:
format: str
file_ref: str
payer: str = ""
currency: str = ""
paid_on: Optional[date] = None
lines: list = field(default_factory=list)
@property
def cash(self) -> Decimal:
return sum((ln.net for ln in self.lines), ZERO)
def _dec(v: str, what: str) -> Decimal:
try:
d = Decimal((v or "0").strip().replace(",", ""))
except InvalidOperation:
raise RuleBroken(f"{what} {v!r} is not an amount. Check the remittance file") from None
if not d.is_finite() or d.as_tuple().exponent < -2 or d < 0:
raise RuleBroken(f"{what} {v!r} is not an amount with at most two decimals. Check the remittance file")
try:
return _money(d) if d else ZERO
except ValueError:
raise RuleBroken(f"{what} {v!r} is not an amount below 1,000,000,000,000,000. Check the remittance file") from None
def parse_csv(content: str, file_ref: str) -> RemitFile:
"""Header row: invoice_ref, amount, and optionally deduction_amount, deduction_reason (and payer, currency, paid_on on any row)."""
reader = csv.DictReader(io.StringIO(content))
names = [h.strip().lower() for h in (reader.fieldnames or [])]
for need in ("invoice_ref", "amount"):
if need not in names:
raise RuleBroken(f"The remittance CSV needs a column {need} in its header row")
out = RemitFile("csv", file_ref)
for n, raw in enumerate(reader, start=2):
row = {k.strip().lower(): (v or "").strip() for k, v in raw.items() if k}
if not row.get("invoice_ref"):
raise RuleBroken(f"Row {n} of the remittance names no invoice")
ded = _dec(row.get("deduction_amount", ""), f"Row {n} deduction")
if ded and not row.get("deduction_reason"):
raise RuleBroken(f"Row {n} has a deduction with no reason. A deduction is the customer's own reason, not a guess")
out.lines.append(RLine(row["invoice_ref"], _dec(row["amount"], f"Row {n} amount"), ded, row.get("deduction_reason", "")))
out.payer = out.payer or row.get("payer", "")
out.currency = out.currency or row.get("currency", "").upper()
if row.get("paid_on") and out.paid_on is None:
try:
out.paid_on = date.fromisoformat(row["paid_on"][:10])
except ValueError:
raise RuleBroken(f"Row {n} has a paid_on that is not a date (2026-11-20)") from None
if not out.lines:
raise RuleBroken("The remittance file has no lines")
return out
def parse_camt054(content: str) -> list:
"""Each Ntry with structured remittance (Strd): RfrdDocInf/Nb is the invoice number, RfrdDocAmt/RmtdAmt what is paid against it,
RfrdDocAmt/AdjstmntAmtAndRsn the amount held back and the customer's reason. One RemitFile per entry, keyed on its reference."""
if "<!DOCTYPE" in content or "<!ENTITY" in content:
raise RuleBroken("That file declares a DOCTYPE, which a bank message never does. Export the camt.054 again")
try:
root = ET.fromstring(content)
except ET.ParseError as e:
raise RuleBroken(f"That is not a readable camt.054 file ({e}). Export it again from the bank") from None
out = []
for ntry in (e for e in root.iter() if _local(e.tag) == "Ntry"):
ref = _text(ntry, "NtryRef") or _text(ntry, "AcctSvcrRef")
if not ref:
raise RuleBroken("An entry of the camt.054 has no reference (NtryRef), so it could not be recognised again")
f = RemitFile("camt054", ref, next((_text(p, "Nm") for p in ntry.iter() if _local(p.tag) == "Dbtr"), ""),
(_find(ntry, "Amt").get("Ccy", "") if _find(ntry, "Amt") is not None else "").upper())
booked = _text(ntry, "BookgDt", "Dt")
f.paid_on = date.fromisoformat(booked[:10]) if booked else None
for strd in (e for e in ntry.iter() if _local(e.tag) == "Strd"):
number = _text(strd, "RfrdDocInf", "Nb")
paid = _text(strd, "RfrdDocAmt", "RmtdAmt")
adj = next((a for a in strd.iter() if _local(a.tag) == "AdjstmntAmtAndRsn"), None)
ded = _text(adj, "Amt") if adj is not None else ""
reason = _text(adj, "AddtlInf") or _text(adj, "Rsn") if adj is not None else ""
if not number or not paid:
raise RuleBroken(f"A structured remittance in entry {ref} lacks the invoice number or the amount")
f.lines.append(RLine(number, _dec(paid, "A remitted amount"), _dec(ded, "A deduction"), reason or ("deduction" if ded else "")))
if f.lines:
out.append(f)
if not out:
raise RuleBroken("The camt.054 has no entry with a structured remittance (Strd). Ask the customer for a CSV, or import it as a statement")
return out
@dataclass(frozen=True)
class LockboxItem:
item_ref: str
amount: Decimal
text: str
def parse_bai2(content: str) -> tuple:
"""A BAI2 lockbox file: 01 file header, 02 group header (as-of date), 03 account, 16 transaction detail (type, amount in cents,
funds type, bank reference, customer reference, text), 88 continuation. Returns (batch, as-of date or None, [LockboxItem]); the batch is the file
id with the group's as-of date, so the same file again names the same items. The items are the BANK's own record of each
cheque, so they create receipts themselves."""
rows = [ln.strip().rstrip("/") for ln in content.replace("\r", "").split("\n") if ln.strip()]
if not rows or not rows[0].startswith("01,"):
raise RuleBroken("That is not a BAI2 file (it should start with an 01 record). Ask the bank for the lockbox file again")
file_id = rows[0].split(",")[3] if len(rows[0].split(",")) > 3 else "FILE"
asof, items, n = "", [], 0
for rec in rows[1:]:
parts = rec.split(",")
if parts[0] == "02" and len(parts) > 4:
asof = parts[4]
elif parts[0] == "16":
n += 1
try:
amount = _money(Decimal(parts[2]) / 100)
except (InvalidOperation, ValueError, IndexError):
raise RuleBroken(f"Lockbox record {n} has an amount that cannot be read (BAI2 amounts are in cents)") from None
ref = (parts[5].strip() if len(parts) > 5 else "") or (parts[4].strip() if len(parts) > 4 else "") or str(n)
items.append(LockboxItem(ref, amount, ",".join(parts[6:]) if len(parts) > 6 else ""))
elif parts[0] == "88" and items:
last = items[-1]
items[-1] = LockboxItem(last.item_ref, last.amount, (last.text + " " + ",".join(parts[1:])).strip())
if not items:
raise RuleBroken("The BAI2 file has no lockbox items (16 records)")
try:
on = date(2000 + int(asof[:2]), int(asof[2:4]), int(asof[4:6])) if len(asof) >= 6 and asof[:6].isdigit() else None
except ValueError:
on = None
return f"{file_id}-{re.sub(r'[^0-9A-Za-z]', '', asof) or '0'}", on, items