dodil/erp-brick-expenses
DODIL ERP brick: expenses
src/brick_expenses/domain/cards.py
"""Company cards (EXP-06, EXP-14, EXP-15): a transaction from the provider's feed is matched to a claim line OR marked personal, and personal
spend is recovered once - netted from the next payout or repaid, never both.
Plain Python. The card_txn state machine (module.yaml workflows.card_txn) only moves forward:
imported -> matched (an employee attaches it to a line) imported -> personal (the employee says it was not business)
personal -> recovered (netted from a payout OR repaid - one claim row decides)
A matched transaction is final: the signed process has no way back, so a line matched to a card cannot be removed (the line IS the receipt).
"""
import csv
import io
from dataclasses import dataclass
from datetime import date
from decimal import Decimal, InvalidOperation
from difflib import SequenceMatcher
from typing import Optional
from brick_expenses.domain.errors import RuleBroken
from brick_expenses.domain.money import ZERO, money
STATES = ("imported", "matched", "personal", "recovered")
TERMINAL = ("recovered",)
MOVES = {
"match_card_txn": (("imported",), "matched"),
"mark_personal": (("imported",), "personal"),
"recover_personal": (("personal",), "recovered"),
}
FEED_MAX = 5000
AMOUNT_MAX = Decimal("1e13")
HEADER = ("card_id", "ref", "posted_on", "merchant", "amount", "currency") # + optional original_amount, original_currency
def move(txn: dict, key: str) -> str:
frm, to = MOVES[key]
if txn["status"] not in frm:
raise RuleBroken(f"Card transaction {txn['card_txn_id']} is {txn['status']}: {key.replace('_', ' ')} needs it to be {' or '.join(frm)}")
return to
def txn_id(card_id: str, ref: str) -> str:
return f"card:{card_id}:{ref}"
def card_of(card_txn_id: str) -> str:
"""The card a transaction is on: card:<card_id>:<ref> (a ref may itself contain a colon)."""
return card_txn_id.split(":", 2)[1]
class FeedError(Exception):
"""The file is not the documented shape (422 {gate: bounds}), or its total is not its lines' sum (422 {gate: feed_total})."""
def __init__(self, message: str, gate: str = "bounds"):
super().__init__(message)
self.gate = gate
@dataclass(frozen=True)
class FeedRow:
card_id: str
ref: str
posted_on: date
merchant: str
amount: Decimal
currency: str
original_amount: Optional[Decimal] = None
original_currency: str = ""
@property
def id(self) -> str:
return txn_id(self.card_id, self.ref)
def _amount(raw: str, where: str) -> Decimal:
try:
d = Decimal(str(raw).strip())
except InvalidOperation:
raise FeedError(f"{where}: {raw!r} is not an amount") from None
if not d.is_finite() or d.as_tuple().exponent < -2 or not (ZERO < d < AMOUNT_MAX):
raise FeedError(f"{where}: an amount is more than zero, below 10,000,000,000,000, with at most two decimal places (credits are not imported yet)")
return d
def parse_feed(content: str, total: Decimal) -> list:
"""The provider's file, in the one documented CSV shape (a header, then card_id,ref,posted_on,merchant,amount,currency[,original_amount,
original_currency]). Every row is checked and the file's total must equal the sum of its rows - or nothing is imported."""
reader = csv.reader(io.StringIO(content.strip("\n ")))
try:
head = [h.strip().lower() for h in next(reader)]
except StopIteration:
raise FeedError("the file is empty") from None
if tuple(head[:6]) != HEADER:
raise FeedError(f"the first line must be the header {','.join(HEADER)}[,original_amount,original_currency]")
rows, seen = [], set()
for n, cells in enumerate(reader, start=2):
if not any(c.strip() for c in cells):
continue
if len(cells) < 6:
raise FeedError(f"line {n}: {len(cells)} columns, at least {len(HEADER)} are needed")
card_id, ref, posted, merchant, amount, cur = (c.strip() for c in cells[:6])
if not card_id or not ref:
raise FeedError(f"line {n}: card_id and ref are required")
try:
when = date.fromisoformat(posted)
except ValueError:
raise FeedError(f"line {n}: posted_on {posted!r} is a date like 2026-11-05") from None
if not (date(2000, 1, 1) <= when <= date(2100, 1, 1)):
raise FeedError(f"line {n}: posted_on {when} is outside 2000-01-01 .. 2100-01-01")
if len(cur) != 3 or not cur.isalpha():
raise FeedError(f"line {n}: currency {cur!r} is three letters such as GBP")
orig = _amount(cells[6], f"line {n} original_amount") if len(cells) > 6 and cells[6].strip() else None
orig_cur = cells[7].strip().upper() if len(cells) > 7 and cells[7].strip() else ""
if (orig is None) != (not orig_cur):
raise FeedError(f"line {n}: original_amount and original_currency go together")
row = FeedRow(card_id, ref, when, merchant[:200], _amount(amount, f"line {n} amount"), cur.upper(), orig, orig_cur)
if row.id in seen:
raise FeedError(f"line {n}: {row.id} appears twice in the file")
seen.add(row.id)
rows.append(row)
if not rows:
raise FeedError("the file has no transactions")
if len(rows) > FEED_MAX:
raise FeedError(f"{len(rows)} transactions: one file holds at most {FEED_MAX}")
if len({r.currency for r in rows}) > 1:
raise FeedError("one file is in one billing currency")
got = money(sum((r.amount for r in rows), ZERO))
if got != money(total):
raise FeedError(f"the file says {money(total):.2f} but its {len(rows)} lines add up to {got:.2f}: nothing was imported", "feed_total")
return rows
# ── matching ──────────────────────────────────────────────────────────────────────────────────
def _words(s: str) -> set:
return {w for w in "".join(c.lower() if c.isalnum() else " " for c in s).split() if len(w) > 1}
def similar(a: str, b: str) -> float:
"""0..1: how alike two merchant names are (shared words, else the character-level ratio)."""
wa, wb = _words(a), _words(b)
shared = len(wa & wb) / max(min(len(wa), len(wb)), 1) if wa and wb else 0.0
return max(shared, SequenceMatcher(None, a.lower(), b.lower()).ratio())
def suggest(txns: list, lines: list, *, days: int = 3, threshold: float = 0.6) -> list:
"""[{card_txn_id, line_no, score}] best first: a line and an imported transaction with the same amount, a date within `days`, and a
merchant alike. Each transaction and each line is suggested once (the best pairing)."""
cands = []
for ln in lines:
if ln.card_txn_id:
continue
for t in txns:
if t["status"] != "imported":
continue
amounts = {money(t["billed_amount"])} | ({money(t["original_amount"])} if t["original_amount"] is not None else set())
if ln.amount not in amounts and (ln.receipt_amount is None or money(ln.receipt_amount) not in amounts):
continue
if abs((t["posted_on"] - ln.expense_date).days) > days:
continue
like = similar(t["merchant"] or "", ln.merchant)
if like >= threshold:
cands.append((like, -abs((t["posted_on"] - ln.expense_date).days), t["card_txn_id"], ln.line_no))
cands.sort(reverse=True)
out, used_t, used_l = [], set(), set()
for like, _, tid, no in cands:
if tid in used_t or no in used_l:
continue
used_t.add(tid)
used_l.add(no)
out.append({"card_txn_id": tid, "line_no": no, "score": round(like, 2)})
return out
def check_match(txn: dict, line, report_currency: str, report_company: str) -> None:
"""The gate of match_card_txn: the transaction imported, the line unmatched, the same currency billed, the same company."""
if line.card_txn_id:
raise RuleBroken(f"Line {line.line_no} is already matched to {line.card_txn_id}: a line takes one card transaction")
if txn["billed_currency"] != report_currency:
raise RuleBroken(f"The card billed {txn['billed_currency']} and this claim is in {report_currency}: the same currency billed is needed to match")
if txn["company_code"] != report_company:
raise RuleBroken(f"Card transaction {txn['card_txn_id']} belongs to {txn['company_code']}, this claim to {report_company}")
# ── recovery ───────────────────────────────────────────────────────────────────────────────────
def choose_netting(personal: list, reimbursable: Decimal) -> list:
"""The personal transactions (oldest first) a payout nets: as many as fit inside what the claim reimburses; one that would take the payout
below zero stays owed and is recovered by repayment or the next claim."""
out, left = [], money(reimbursable)
for t in sorted(personal, key=lambda t: (t["posted_on"], t["card_txn_id"])):
amt = money(t["billed_amount"])
if amt <= left:
out.append(t)
left -= amt
return out
def recovery_journal_lines(txn: dict, bp_id: Optional[int]) -> list:
"""Personal spend on the company card: the employee owes it back (employee_receivable) and the card liability stands (EXP-14)."""
amt = f"{money(txn['billed_amount']):.2f}"
return [{"purpose": "employee_receivable", **({"bp_id": str(bp_id)} if bp_id else {}), "debit": amt},
{"purpose": "card_liability", "memo": f"card {txn['card_id']}", "credit": amt}]