dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/domain/bankfile.py
"""Bank statement files: ISO 20022 camt.053 (BankToCustomerStatement), SWIFT MT940 and CSV. EVERY line is read, credit or
debit (v4: each is kept as a statement line and reconciled); only a credit that is not our own transfer and that no rule
names becomes a customer receipt. Pure parsing - no storage."""
import csv
import io
import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Optional
from erp_platform.types import BigId
from pydantic import TypeAdapter, ValidationError
from brick_payments.application.inputs import _money # one definition of a sane amount: finite, 2 places, below 1e15
from brick_payments.domain.errors import RuleBroken
_BIG_ID = TypeAdapter(BigId)
def _amount(value: Decimal) -> Decimal:
"""An amount the parser lets through: finite, at most two places, below 1e15 (zero is read, and is no credit)."""
return value if value == 0 else _money(abs(value))
@dataclass(frozen=True)
class Line:
entry_ref: str
booked_on: date
amount: Decimal
currency: str
credit: bool
remittance: str = ""
payer: str = ""
bp_id: Optional[int] = None
counterparty_account: str = ""
bank_tx_code: str = ""
value_on: Optional[date] = None
@dataclass(frozen=True)
class Statement:
"""A parsed file: its lines, and (when the format carries them) the statement id, period and booked balances."""
lines: list
statement_ref: str = ""
opening_balance: Optional[Decimal] = None
closing_balance: Optional[Decimal] = None
from_date: Optional[date] = None
to_date: Optional[date] = None
account: str = ""
def _local(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
def _find(el, *path):
cur = [el]
for name in path:
cur = [c for e in cur for c in e if _local(c.tag) == name]
if not cur:
return None
return cur[0]
def _text(el, *path) -> str:
found = _find(el, *path)
return (found.text or "").strip() if found is not None else ""
def _camt_root(content: str):
if "<!DOCTYPE" in content or "<!ENTITY" in content:
raise RuleBroken("That file declares a DOCTYPE, which a bank statement never does. Export the camt.053 again")
try:
root = ET.fromstring(content)
except ET.ParseError as e:
raise RuleBroken(f"That is not a readable camt.053 file ({e}). Export it again from the bank") from None
return root
def parse_camt053(content: str) -> list:
"""Every entry of every statement. The entry reference is NtryRef, else AcctSvcrRef (on the entry or its
transaction), which is what makes a re-import recognisable."""
root = _camt_root(content)
out = []
for ntry in (e for e in root.iter() if _local(e.tag) == "Ntry"):
ref = (_text(ntry, "NtryRef") or _text(ntry, "AcctSvcrRef")
or next((_text(t, "Refs", "AcctSvcrRef") for t in ntry.iter() if _local(t.tag) == "TxDtls"), ""))
amt = _find(ntry, "Amt")
booked = _text(ntry, "BookgDt", "Dt") or _text(ntry, "BookgDt", "DtTm")[:10]
if amt is None or not booked or not ref:
raise RuleBroken("An entry in the statement has no amount, booking date or entry reference. "
"Every line needs all three to be recognised again")
try:
value = Decimal((amt.text or "").strip())
_amount(value)
credit = _text(ntry, "CdtDbtInd") == "CRDT"
party, acct = ("Dbtr", "DbtrAcct") if credit else ("Cdtr", "CdtrAcct")
account = next((_text(a, "Id", "IBAN") or _text(a, "Id", "Othr", "Id") for a in ntry.iter() if _local(a.tag) == acct), "")
valued = _text(ntry, "ValDt", "Dt")
out.append(Line(ref, date.fromisoformat(booked[:10]), value, amt.get("Ccy", "").upper(), credit,
" ".join(t.text.strip() for t in ntry.iter() if _local(t.tag) == "Ustrd" and t.text),
next((_text(p, "Nm") for p in ntry.iter() if _local(p.tag) == party), ""),
None, account, _text(ntry, "BkTxCd", "Domn", "Cd"), date.fromisoformat(valued[:10]) if valued else None))
except (InvalidOperation, ValueError):
raise RuleBroken(f"Entry {ref} has an amount or date that cannot be read. Check the file") from None
return out
CSV_ALIASES = {"date": ("booking_date", "date", "booked_on"), "amount": ("amount",), "currency": ("currency", "ccy"),
"entry_ref": ("entry_ref", "reference", "id", "entry_reference"),
"remittance": ("remittance", "description", "payment_ref", "narrative"),
"payer": ("payer", "name", "counterparty"), "bp_id": ("bp_id", "customer"),
"account": ("counterparty_account", "iban", "account"), "value_date": ("value_date", "value_on")}
def parse_csv(content: str, default_currency: str = "") -> list:
"""CSV with a header: booking_date, amount, currency, entry_ref and optionally remittance, payer, bp_id.
A negative amount is a debit."""
reader = csv.DictReader(io.StringIO(content))
cols = {k: next((h for h in (reader.fieldnames or []) if h and h.strip().lower() in names), None)
for k, names in CSV_ALIASES.items()}
for need in ("date", "amount", "entry_ref"):
if cols[need] is None:
raise RuleBroken(f"The CSV needs a column for the {need.replace('_', ' ')} "
f"({' / '.join(CSV_ALIASES[need])}). Add it to the header row")
out = []
for n, row in enumerate(reader, start=2):
try:
value = Decimal(row[cols["amount"]].strip().replace(",", ""))
_amount(value)
cur = (row.get(cols["currency"] or "", "") or default_currency).strip().upper()
bp = (row.get(cols["bp_id"] or "", "") or "").strip()
out.append(Line(row[cols["entry_ref"]].strip(), date.fromisoformat(row[cols["date"]].strip()[:10]),
abs(value), cur, value > 0, (row.get(cols["remittance"] or "", "") or "").strip(),
(row.get(cols["payer"] or "", "") or "").strip(), _BIG_ID.validate_python(int(bp)) if bp.isdigit() else None,
(row.get(cols["account"] or "", "") or "").strip(), "",
date.fromisoformat(row[cols["value_date"]].strip()[:10]) if cols["value_date"] and (row.get(cols["value_date"]) or "").strip() else None))
except (InvalidOperation, ValueError, KeyError, AttributeError, ValidationError):
raise RuleBroken(f"Row {n} of the CSV cannot be read (date as 2026-11-20, amount as 123.45). Fix it and import again") from None
if not out[-1].entry_ref:
raise RuleBroken(f"Row {n} of the CSV has no entry reference, so a re-import could not recognise it. Add one")
return out
def _balance(root, code: str) -> Optional[Decimal]:
for bal in (e for e in root.iter() if _local(e.tag) == "Bal"):
if _text(bal, "Tp", "CdOrPrtry", "Cd") == code:
amt = _find(bal, "Amt")
value = _amount_text(amt.text if amt is not None else "")
return -value if _text(bal, "CdtDbtInd") == "DBIT" else value
return None
def _amount_text(t: Optional[str]) -> Decimal:
try:
return _money_signed(Decimal((t or "").strip().replace(",", ".")))
except (InvalidOperation, ValueError):
raise RuleBroken("A balance in the statement cannot be read. Export it again from the bank") from None
def _money_signed(v: Decimal) -> Decimal:
return v if v == 0 else (-_money(abs(v)) if v < 0 else _money(v))
def parse_camt053_statement(content: str) -> Statement:
root = _camt_root(content)
stmt = next((e for e in root.iter() if _local(e.tag) == "Stmt"), None)
frm = _text(stmt, "FrToDt", "FrDtTm")[:10] if stmt is not None else ""
to = _text(stmt, "FrToDt", "ToDtTm")[:10] if stmt is not None else ""
return Statement(parse_camt053(content), _text(stmt, "Id") if stmt is not None else "", _balance(root, "OPBD"), _balance(root, "CLBD"),
date.fromisoformat(frm) if frm else None, date.fromisoformat(to) if to else None,
(_text(stmt, "Acct", "Id", "IBAN") or _text(stmt, "Acct", "Id", "Othr", "Id")) if stmt is not None else "")
# ── SWIFT MT940 ──────────────────────────────────────────────────────────────────────────────────────────
_IBAN = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{10,30}\b")
def _yymmdd(v: str) -> date:
return date(2000 + int(v[:2]), int(v[2:4]), int(v[4:6]))
def _mt_amount(v: str) -> Decimal:
return _money_signed(Decimal(v.replace(",", ".")))
def parse_mt940(content: str, default_currency: str = "") -> Statement:
"""A SWIFT MT940 statement: :20: reference, :25: account, :28C: statement number, :60F: opening, :61:/:86: each entry and
its information, :62F: closing. The entry reference is the bank's (after //), else the customer's; a line with neither is
named from its date, amount and place so a re-import still recognises it."""
text = content.replace("\r\n", "\n").replace("\r", "\n")
tags = re.findall(r"^:(\d{2}[A-Z]?):(.*?)(?=^:\d{2}[A-Z]?:|^-\}|\Z)", text, flags=re.S | re.M)
if not tags:
raise RuleBroken("That is not a readable MT940 file (no :20: / :61: tags). Export it again from the bank")
out, opening, closing, ref, account, number, ccy = [], None, None, "", "", "", default_currency.upper()
cur = None
try:
for tag, body in tags:
body = body.strip()
if tag == "20":
ref = body.split("\n")[0].strip()
elif tag == "25":
account = body.split("\n")[0].strip()
elif tag == "28C":
number = body.split("\n")[0].strip().replace("/", "-")
elif tag in ("60F", "60M", "62F", "62M"):
m = re.match(r"([CD])(\d{6})([A-Z]{3})([\d,]+)", body)
if not m:
raise ValueError(tag)
value = _mt_amount(m.group(4)) * (-1 if m.group(1) == "D" else 1)
ccy = ccy or m.group(3)
if tag.startswith("60"):
opening = value if opening is None else opening
else:
closing = value
elif tag == "61":
m = re.match(r"(\d{6})(\d{4})?(R?[CD])([A-Z]?)([\d,]+)([A-Z])([A-Z0-9]{3})([^\n]*?)(?://([^\n]*))?(?:\n|$)", body)
if not m:
raise ValueError("61")
valued = _yymmdd(m.group(1))
booked = date(valued.year, int(m.group(2)[:2]), int(m.group(2)[2:])) if m.group(2) else valued
if m.group(2) and abs((booked - valued).days) > 200: # across a year end
booked = date(valued.year + (1 if booked < valued else -1), booked.month, booked.day)
credit = m.group(3).endswith("C")
customer_ref, bank_ref = (m.group(8) or "").strip(), (m.group(9) or "").strip()
entry = bank_ref or (customer_ref if customer_ref.upper() != "NONREF" else "")
cur = {"booked": booked, "valued": valued, "credit": credit, "amount": _mt_amount(m.group(5)),
"ref": entry, "code": m.group(7), "info": ""}
out.append(cur)
elif tag == "86" and cur is not None:
cur["info"] = " ".join(body.split())
except (ValueError, InvalidOperation):
raise RuleBroken("An entry in the MT940 file cannot be read (:60F:, :61: or :62F: malformed). Export it again from the bank") from None
lines, seen = [], {}
for e in out:
info = e["info"]
acct = (_IBAN.search(info) or [None])[0] if _IBAN.search(info) else ""
name = (re.search(r"\?3[23]([^?]*)", info) or [None, ""])[1].strip() if "?3" in info else ""
key = e["ref"] or f"{e['booked']:%Y%m%d}-{e['amount']}-{'C' if e['credit'] else 'D'}"
seen[key] = seen.get(key, 0) + 1
entry_ref = key if seen[key] == 1 else f"{key}#{seen[key]}"
lines.append(Line(entry_ref, e["booked"], e["amount"], ccy, e["credit"], info, name, None, acct, e["code"], e["valued"]))
dates = [ln.booked_on for ln in lines]
return Statement(lines, number or ref, opening, closing, min(dates) if dates else None, max(dates) if dates else None, account)
def parse_statement(fmt: str, content: str, default_currency: str = "") -> Statement:
if fmt == "camt053":
return parse_camt053_statement(content)
if fmt == "mt940":
return parse_mt940(content, default_currency)
lines = parse_csv(content, default_currency)
dates = [ln.booked_on for ln in lines]
return Statement(lines, "", None, None, min(dates) if dates else None, max(dates) if dates else None)
def normalise_account(v: str) -> str:
return re.sub(r"[^A-Za-z0-9]", "", v or "").upper()
def same_account(a: str, b: str) -> bool:
"""Two account identifiers name the same account when one ends with the other (an IBAN and its BBAN, a sort code with the
number); at least 6 characters, so a short fragment never matches by accident."""
x, y = normalise_account(a), normalise_account(b)
return len(x) >= 6 and len(y) >= 6 and (x.endswith(y) or y.endswith(x))