dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/application/statements.py
"""The bank statement, line by line (PAY-02 v3, PAY-21, PAY-22, PAY-23).
EVERY line of a statement - credit or debit - is kept as a statement line keyed bank:<account>:<entry ref>, and ends matched or
posted, never deleted. A credit that is a customer's money also becomes a receipt (the same key); a credit from one of our own
accounts, or one a reconciliation rule names, does not.
ONE guard key per line: every operation that touches a line (import, match, post, unmatch, accept or reject a suggestion, a rule
posting) writes the line's next claim (pay_claim 'line') in its own transaction with a plain INSERT, so two different operations
on one line collide (409) instead of the second quietly winning. A counterpart (a receipt, a refund, a tax liability) is held by
one line at a time: its own claim key, checked and taken in the same transaction.
"""
import dataclasses
import hashlib
from datetime import date, timedelta
from decimal import Decimal
from typing import Literal, Optional
from erp_platform import auth
from erp_platform.bricks import Refused
from erp_platform.types import Limit, Name, text
from brick_payments.application import use_cases as uc
from brick_payments.application.inputs import Code, Day, Money, Ref
from brick_payments.domain import bankfile, reconcile
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.money import ZERO
from brick_payments.domain.readiness import fmt
from brick_payments.domain.receipt import Receipt
from brick_payments.infrastructure import seams
PAY = uc.PAY
RECONCILE, RULE_WRITE, SUGGEST = "pay:statement_line:reconcile", "pay:reconciliation_rule:write", "pay:statement_line:suggest"
NO_VAT = ("E", "Z", "O", "N", "OS", "NONE", "") # tax codes that carry no VAT to compute
NOT_YET = ("manual",)
KINDS = ("receipt", "refund", "tax_liability", "dd_run", "deposit_batch", "cash_deposit", "supplier_payment")
def _s(v) -> Optional[str]:
return None if v is None else f"{Decimal(v):.2f}"
def line_view(row: dict, repo=None) -> dict:
out = {k: (v.isoformat() if isinstance(v, date) else _s(v) if isinstance(v, Decimal) else v) for k, v in row.items()
if k not in ("created_at", "updated_at")}
out["direction"] = "credit" if Decimal(str(row["amount"])) > 0 else "debit"
if repo is not None:
out["suggestions"] = [s for s in repo.bank.suggestions_of(row["line_id"]) if s["status"] == "open"]
return out
def _line(repo, line_id: str) -> dict:
row = repo.bank.line(line_id)
if row is None:
raise Refused(404, f"no statement line {line_id}")
return row
def _guard(repo, line_id: str, who: str, what: str) -> int:
"""The line's claim: the next slot, in THIS transaction. A concurrent operation read the same state and claims the same slot."""
return repo.claims.take("line", line_id, uc._now(), who, what)
def _base_line(line_id, statement_id, account, ln, **more) -> dict:
signed = ln.amount if ln.credit else -ln.amount
return {"line_id": line_id, "statement_id": statement_id, "bank_account_id": account, "booked_on": ln.booked_on,
"value_on": ln.value_on or ln.booked_on, "amount": signed, "currency": ln.currency,
"counterparty_name": ln.payer, "counterparty_account_masked": _mask(ln.counterparty_account),
"text": ln.remittance, "bank_tx_code": ln.bank_tx_code, "lifecycle": "open", "matched_kind": "", "matched_ref": "",
"account_id": "", "tax_code": "", "cost_centre": "", "journal_id": "", "decided_how": "", "decided_by": "",
"decided_on": None, "receipt_image": "", "vat_amount": None, "vat_reclaimed": "", "reversal_reason": "", **more}
def _mask(v: str) -> str:
v = bankfile.normalise_account(v)
return ("*" * max(0, len(v) - 4) + v[-4:]) if v else ""
# ═══ the import (PAY-02) ═════════════════════════════════════════════════════════════════════════════════
def _batch_for(repo, bank_account_id: str, ln) -> Optional[dict]:
"""The ONE deposited, unmatched batch of this account whose total is this credit; two of the same total are ambiguous: none."""
if not ln.credit:
return None
found = [b for b in repo.deposits.search("deposited", bank_account_id) if Decimal(str(b["total"])) == ln.amount and b["currency"] == (ln.currency or b["currency"])]
return found[0] if len(found) == 1 else None
def _bag_named(repo, bank_account_id: str, ln) -> Optional[dict]:
"""v6: a credit whose text names an expected cash bag's number (dep:<module>:<shop>:<bag_no>) whose total it does not equal."""
if not ln.credit:
return None
text = f"{ln.payer} {ln.remittance}".upper()
for b in repo.deposits.search("deposited", bank_account_id):
bag = (b.get("batch_ref") or "").rsplit(":", 1)[-1].upper()
if b.get("kind") == "cash" and len(bag) >= 3 and bag in text and Decimal(str(b["total"])) != ln.amount:
return b
return None
def import_core(repo, caller: dict, format: str, content: str, bank_account_id: str, currency: Optional[str]) -> dict:
bank = repo.config.bank(bank_account_id)
if bank is None:
raise Refused(422, f"there is no bank account {bank_account_id} - set it up in the settings first")
return import_parsed(repo, caller, format, bankfile.parse_statement(format, content, currency or bank.get("currency") or ""),
content, bank_account_id)
def import_parsed(repo, caller: dict, format: str, st, content: str, bank_account_id: str) -> dict:
"""The import of a parsed statement (a file, or an open-banking feed's delivery). A statement whose opening + lines is not its closing
is refused naming the difference (PAY-34); one that does not start where the previous ended is imported and the gap named."""
today, now, settings = uc._today(repo), uc._now(), repo.settings.current()
bank = repo.config.bank(bank_account_id)
if bank is None:
raise Refused(422, f"there is no bank account {bank_account_id} - set it up in the settings first")
if bank.get("closed_from") and bank["closed_from"] <= today:
raise RuleBroken(f"Bank account {bank_account_id} was closed from {bank['closed_from']}. A statement for it cannot be imported")
named = st # the statement's id comes from what the file itself named
if st.lines and (st.from_date is None or st.to_date is None): # a file that names no period covers the days its lines are booked on
st = dataclasses.replace(st, from_date=st.from_date or min(ln.booked_on for ln in st.lines), to_date=st.to_date or max(ln.booked_on for ln in st.lines))
total = reconcile.signed_total(st.lines)
gap = reconcile.balance_gap(st.opening_balance, st.closing_balance, total)
if gap:
raise RuleBroken(f"statement_balances: opening {fmt(st.opening_balance)} + lines {fmt(total)} = {fmt(st.opening_balance + total)}, but the "
f"statement's closing balance is {fmt(st.closing_balance)} - {fmt(abs(gap))} {'missing from' if gap > 0 else 'too much in'} the lines. "
"The file is incomplete or damaged: get the bank's full statement")
cutover = settings.import_from()
sid = f"{bank_account_id}:{named.statement_ref or ((named.from_date.isoformat() + '..' + named.to_date.isoformat()) if named.from_date else hashlib.sha256(content.encode()).hexdigest()[:12])}"
who = uc._who(caller)
counts = {"lines": len(st.lines), "created": 0, "skipped_duplicate": 0, "skipped_before_cutover": 0, "debits_kept": 0,
"skipped_future": 0, "unassigned": 0, "own_transfers": 0, "named_by_a_rule": 0, "lines_created": 0}
ids = {ln.entry_ref: f"bank:{bank_account_id}:{ln.entry_ref}" for ln in st.lines}
have_lines, have_receipts = repo.bank.existing_lines(ids.values()), repo.receipts.existing(ids.values())
ours = [b for b in repo.config.banks() if (b.get("account_ref") or b.get("iban")) and b["bank_account_id"] != bank_account_id]
rules = repo.bank.rules_on(today)
fresh, made, new_lines, open_ids = set(), [], [], []
for ln in st.lines:
pid = ids[ln.entry_ref]
if cutover and ln.booked_on < cutover:
counts["skipped_before_cutover"] += 1
continue
if pid in have_lines or pid in fresh:
counts["skipped_duplicate"] += 1
continue
if ln.booked_on > today:
counts["skipped_future"] += 1
continue
fresh.add(pid)
row = _base_line(pid, sid, bank_account_id, ln, currency=ln.currency or bank.get("currency") or "GBP")
holder = None
if ln.credit:
holder = pid if pid in have_receipts else (repo.receipts.ref_holder(bank_account_id, ln.entry_ref)
or repo.receipts.ref_holder("", ln.entry_ref))
if holder: # a receipt already stands for this credit (v2 import, or typed in earlier)
counts["skipped_duplicate"] += 1
row.update(lifecycle="matched", matched_kind="receipt", matched_ref=holder, decided_how="import", decided_by=who, decided_on=today)
elif not ln.credit:
counts["debits_kept"] += 1
open_ids.append(pid)
elif any(bankfile.same_account(ln.counterparty_account, b["account_ref"] or b["iban"]) or (b.get("iban") and bankfile.same_account(ln.counterparty_account, b["iban"]))
for b in ours):
counts["own_transfers"] += 1 # our own money moving between our accounts (PAY-25): not a customer's
open_ids.append(pid)
elif reconcile.matching_rules(rules, {**row, "amount": ln.amount}):
counts["named_by_a_rule"] += 1 # interest, a supplier's refund: a rule names it, so it is not a receipt
open_ids.append(pid)
elif ln.credit and ln.payer and ln.payer in [p.strip() for p in settings.card_providers.split(",") if p.strip()]:
counts["card_provider_credits"] = counts.get("card_provider_credits", 0) + 1 # v6 (PAY-50): a provider's payout is matched to its expected
open_ids.append(pid) # settlements (match_card_settlement), never a customer's receipt
elif _bag_named(repo, bank_account_id, ln):
counts["cash_bag_difference"] = counts.get("cash_bag_difference", 0) + 1 # v6 (PAY-49): names a bag the count of which differs: an approver
open_ids.append(pid) # accepts the difference (accept_deposit_difference), it is no receipt
elif _batch_for(repo, bank_account_id, ln):
batch = _batch_for(repo, bank_account_id, ln) # the paying-in slip's credit: the batch, never a receipt again (PAY-28)
journal = ""
if batch.get("kind") == "cash": # v6 (PAY-49): a bag posts bank / cash_in_transit when its credit lands
from brick_payments.application import expected
try:
journal = expected.post_deposit_match(repo, row, batch, who, 1)
except RuleBroken:
counts["cash_bag_not_posted"] = counts.get("cash_bag_not_posted", 0) + 1 # the ledger is not there: left open, matched by hand later
open_ids.append(pid)
new_lines.append(row)
continue
repo.claims.take("counterpart", f"deposit_batch:{batch['batch_id']}", now, who, pid)
repo.deposits.save({**batch, "lifecycle": "matched", "statement_line_id": pid})
counts["matched_to_deposit_batch"] = counts.get("matched_to_deposit_batch", 0) + 1
row.update(lifecycle="matched", matched_kind="deposit_batch", matched_ref=batch["batch_id"], journal_id=journal, decided_how="import", decided_by=who, decided_on=today)
else:
bp = ln.bp_id or (settings.suspense_bp_id or None)
r = Receipt.record(payment_id=pid, bp_id=bp, currency=ln.currency or bank.get("currency") or "GBP", amount=ln.amount,
received_on=ln.booked_on, today=today, tz=settings.business_timezone, by=who, at=now,
method="bank_statement", bank_account_id=bank_account_id, bank_reference=ln.entry_ref,
remittance_text=" ".join(x for x in (ln.payer, ln.remittance) if x), origin=f"statement:{format}",
statement_line_id=pid)
repo.receipts.add(r)
repo.receipts.claim_ref(bank_account_id, ln.entry_ref, pid, now)
made.append(r)
counts["created"] += 1
counts["unassigned"] += bp is None
row.update(lifecycle="matched", matched_kind="receipt", matched_ref=pid, decided_how="import", decided_by=who, decided_on=today)
new_lines.append(row)
have = repo.bank.statement(sid)
if have is not None and new_lines:
if have["lifecycle"] == "reconciled":
raise RuleBroken(f"Statement {sid} is reconciled. New lines cannot be added to it")
repo.claims.take("statement", sid, now, who, "import") # the statement's one guard key (a reconcile / an unmatch take it too)
if have is None:
cont = continuity(repo, bank_account_id, sid, st)
lifecycle = "balanced" if (gap is not None and cont["met"]) else "imported"
repo.bank.add_statement({"statement_id": sid, "bank_account_id": bank_account_id, "from_date": st.from_date, "to_date": st.to_date,
"opening_balance": st.opening_balance, "closing_balance": st.closing_balance, "format": format,
"file_key": "", "lifecycle": lifecycle, "reconciled_by": "", "imported_by": who,
"imported_on": today}, now)
repo.claims.take("statement", sid, now, who, "import")
counts["statement"] = sid
counts["statement_lifecycle"] = lifecycle
if not cont["met"]:
counts["continuity"] = cont
for row in new_lines:
repo.bank.add_line(row, now)
repo.claims.take("line", row["line_id"], now, who, "import")
counts["lines_created"] += 1
suggestions = 0
for r in made: # after the receipts are written; one suggestion each
if r.bp_id is not None and r.bp_id != settings.suspense_bp_id:
got = uc._suggest(repo, r, caller, settings.ai_suggest_match)
got.pop("_rows", None)
suggestions += bool(got.get("suggested"))
posted = [{"payment_id": r.payment_id, **uc._post_receipt(repo, r, actor=who)} for r in made]
return {**counts, "suggestions": suggestions, "receipt_ids": [r.payment_id for r in made], "open_line_ids": open_ids,
"ledger": {"posted": sum(1 for p in posted if p.get("posted")), "not_posted": sum(1 for p in posted if not p.get("posted")),
"reason": next((p["reason"] for p in posted if not p.get("posted")), "")}}
def continuity(repo, bank_account_id: str, sid: str, st) -> dict:
"""gate statement_continuous: this statement's opening is the previous statement's closing, or the gap is named."""
prev = repo.bank.previous_statement(bank_account_id, st.from_date, sid) if st.from_date else None
gap = reconcile.continuity_gap(Decimal(str(prev["closing_balance"])) if prev and prev["closing_balance"] is not None else None, st.opening_balance)
if not gap:
return {"gate": "statement_continuous", "met": True}
return {"gate": "statement_continuous", "met": False, "previous_statement": prev["statement_id"], "gap": _s(gap),
"reason": f"opening {fmt(st.opening_balance)} is not the closing {fmt(Decimal(str(prev['closing_balance'])))} of {prev['statement_id']} - "
f"{fmt(abs(gap))} {'more' if gap > 0 else 'less'}",
"fix": "import the statement that is missing between them"}
def import_and_reconcile(caller: dict, format: str, content: str, bank_account_id: str, currency: Optional[str]) -> dict:
"""Import in ONE transaction; then, once it has COMMITTED (a transaction does not see its own writes), let the rules and the
suggestion step look at the lines that are still open (the PAY-22 job: after a statement import commits)."""
with uc.repository() as repo, uc._decide():
out = import_core(repo, caller, format, content, bank_account_id, currency)
return after_import(out, caller)
def after_import(out: dict, caller: dict) -> dict:
"""Once the import has COMMITTED: the rules and the suggestion step on the open lines, then our own accounts' two sides of a transfer."""
open_ids = out.pop("open_line_ids")
out["reconcile"] = process_lines(open_ids, caller) if open_ids else {"suggested": 0, "rule_proposals": 0, "auto_posted": 0}
from brick_payments.application import transfers
out["transfers"] = transfers.reconcile_core(caller) # PAY-25: our own accounts' two sides
return out
def process_lines(line_ids: list, caller: dict) -> dict:
"""For each still-open line: the rules first (exactly one rule: propose it, or post it when its mode is auto), else the AI's
suggestion from the payee and past postings. Each line in its own transaction, so one line that cannot be handled is reported
and never loses the others."""
tally = {"suggested": 0, "rule_proposals": 0, "auto_posted": 0, "failed": []}
for lid in line_ids:
try:
with uc.repository() as repo, uc._decide():
got = _process_one(repo, caller, lid)
except Refused as e:
tally["failed"].append({"line_id": lid, "reason": str(e)})
continue
for k in ("suggested", "rule_proposals", "auto_posted"):
tally[k] += got.get(k, 0)
return tally
def _next_n(repo, line_id: str) -> int:
return len(repo.bank.suggestions_of(line_id)) + 1
def _process_one(repo, caller: dict, line_id: str) -> dict:
line = repo.bank.line(line_id)
if line is None or line["lifecycle"] != "open":
return {}
today, now, settings = uc._today(repo), uc._now(), repo.settings.current()
rules = [r for r in repo.bank.rules_on(today) if reconcile.rule_matches(r, line)]
fresh = [s for s in repo.bank.suggestions_of(line_id) if s["status"] == "open"]
if rules:
auto = rules[0] if len(rules) == 1 and rules[0]["mode"] == "auto" else None
if auto is not None:
_post_core(repo, caller, line, account_id=auto["account_id"], tax_code=auto["tax_code"], cost_centre=auto["cost_centre"] or "",
receipt_image="", how="auto", by=f"rule:{auto['rule_id']}", supersede=fresh)
return {"auto_posted": 1}
n = _next_n(repo, line_id)
for s in fresh:
repo.bank.move_suggestion(s["suggestion_id"], status="superseded")
for i, r in enumerate(rules): # two rules on one line: both are shown, neither is applied
repo.bank.add_suggestion({"suggestion_id": f"{line_id}#{n + i}", "line_id": line_id, "account_id": r["account_id"],
"tax_code": r["tax_code"], "cost_centre": r["cost_centre"] or "", "receipt_needed": "",
"evidence": {"rule_id": r["rule_id"], "valid_from": r["valid_from"].isoformat(), "mode": r["mode"],
"also_matched": len(rules) - 1}, "confidence": "1.00", "model": f"rule:{r['rule_id']}",
"status": "open", "decided_by": "", "decided_on": None}, now)
return {"rule_proposals": len(rules)}
if settings.ai_suggest_statement_line == "off" or Decimal(str(line["amount"])) > 0:
return {}
found = reconcile.suggest(line, repo.bank.posted_history())
if found is None:
return {}
for s in fresh:
repo.bank.move_suggestion(s["suggestion_id"], status="superseded") # re-running supersedes an open one, never a decided one
repo.bank.add_suggestion({"suggestion_id": f"{line_id}#{_next_n(repo, line_id)}", "line_id": line_id, "account_id": found.account_id,
"tax_code": found.tax_code, "cost_centre": found.cost_centre, "receipt_needed": "Y" if found.receipt_needed else "",
"evidence": found.evidence, "confidence": str(found.confidence), "model": found.model, "status": "open",
"decided_by": "", "decided_on": None}, now)
return {"suggested": 1}
# ═══ posting a line (PAY-21 post_statement_line, PAY-22, PAY-23) ═════════════════════════════════════════
def _posting_key(line: dict, n: int) -> str:
"""The journal's source document: the line alone for its first posting; once the line was undone (a reversal exists for the old
posting), the posting after it is a NEW journal keyed <line>@<claim n>, or the ledger would replay the reversed one."""
return f"{line['line_id']}@{n}" if line.get("reversal_reason") else line["line_id"]
def _post_core(repo, caller: dict, line: dict, *, account_id: str, tax_code: str, cost_centre: str, receipt_image: str, how: str,
by: str, supersede: list, suggestion_id: str = "", suggestion_status: str = "") -> dict:
"""Post an OPEN line to the ledger: an account and a tax code; the VAT of a payment OUT is computed by tax (direction purchase)
and is reclaimed only with the receipt image kept. The journal is derived from the line (event bank_line), so posting twice
changes nothing. The line's claim is taken FIRST."""
if line["lifecycle"] != "open":
raise RuleBroken(f"Line {line['line_id']} is already {line['lifecycle']}. Undo that first, with a reason")
if not account_id.strip():
raise RuleBroken("Posting a line needs the ledger account it belongs to. Name one")
if not tax_code.strip():
raise RuleBroken("Posting a line needs a tax code (E for no VAT). Name one")
bank = repo.config.bank(line["bank_account_id"]) or {}
if not bank.get("gl_account_id"):
raise RuleBroken(f"Bank account {line['bank_account_id']} has no ledger account, so the line cannot be posted. Set it in the settings")
today = uc._today(repo)
amount = Decimal(str(line["amount"]))
gross, out_ = abs(amount), amount < 0
n = _guard(repo, line["line_id"], by, "post")
vat, reclaimed = ZERO, False
if out_ and tax_code not in NO_VAT:
try:
got = seams.tax_compute(f"pay_line:{line['line_id']}", line["booked_on"], line["currency"], tax_code, gross,
line["text"] or line["counterparty_name"] or line["line_id"], by)
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
vat, reclaimed = got["vat"], bool(receipt_image.strip()) and got["vat_reclaimable"] > ZERO
expense = {"account_id": account_id, **({"tax_code": tax_code} if reclaimed else {}), **({"cost_centre": cost_centre} if cost_centre else {})}
side = seams.account_line(bank["gl_account_id"])
g = f"{gross:.2f}"
lines = [{**expense, "debit": g}, {**side, "credit": g}] if out_ else [{**side, "debit": g}, {**expense, "credit": g}]
posted = seams.post_journal(_posting_key(line, n), "bank_line", line["booked_on"], line["currency"],
f"{line['counterparty_name'] or 'Bank line'} {line['text']}"[:200].strip(), lines, by, bank.get("company_code") or "")
if not posted.get("posted"):
raise RuleBroken(f"The ledger did not take the posting ({posted.get('reason', 'no reason given')}). Nothing was posted; try again")
repo.bank.save_line({**line, "lifecycle": "posted", "account_id": account_id, "tax_code": tax_code, "cost_centre": cost_centre,
"journal_id": posted["journal_id"], "decided_how": how, "decided_by": by, "decided_on": today,
"receipt_image": receipt_image, "vat_amount": vat if out_ else None, "vat_reclaimed": "Y" if reclaimed else "",
"matched_kind": "rule" if how in ("rule", "auto") else "manual", "matched_ref": account_id})
for s in supersede:
if s["suggestion_id"] == suggestion_id:
repo.bank.move_suggestion(s["suggestion_id"], status=suggestion_status, decided_by=by, decided_on=today)
elif s["status"] == "open":
repo.bank.move_suggestion(s["suggestion_id"], status="superseded")
return {"journal_id": posted["journal_id"], "vat": _s(vat), "vat_reclaimed": reclaimed}
@PAY.use_case(kind="command", route="POST /statement-lines/{line_id}/post", permission=RECONCILE)
def post_statement_line(caller: dict, line_id: Ref, account_id: Code, tax_code: Code, cost_centre: Optional[Code] = None,
receipt_image: Optional[text(300)] = None) -> dict:
"""Post an open statement line yourself: the ledger account and tax code you confirm (E when there is no VAT). The VAT on a
payment out is computed by tax (direction purchase) and is reclaimed only when a VAT receipt image is attached. The journal is
dated by the bank and keyed on the line, so posting twice changes nothing; a line already matched or posted is refused."""
with uc.repository() as repo, uc._decide():
line = _line(repo, line_id)
opened = repo.bank.open_suggestions_of(line_id)
done = _post_core(repo, caller, line, account_id=account_id, tax_code=tax_code, cost_centre=cost_centre or "",
receipt_image=receipt_image or "", how="manual", by=uc._who(caller), supersede=opened)
return {**done, "line_id": line_id, "lifecycle": "posted", "decided_how": "manual"}
@PAY.use_case(kind="command", route="POST /statement-lines/{line_id}/decide-suggestion", permission=RECONCILE)
def decide_line_suggestion(caller: dict, line_id: Ref, decision: Literal["accept", "change", "reject"], suggestion_id: Optional[Ref] = None,
account_id: Optional[Code] = None, tax_code: Optional[Code] = None, cost_centre: Optional[Code] = None,
receipt_image: Optional[text(300)] = None) -> dict:
"""Accept, change or reject the suggestion on an open line (the AI's, or a rule in suggest mode). Accept posts what it proposed;
change posts the account (and tax code) you give; reject leaves the line open. The AI never posts: the line carries
decided_how accepted | changed (a rule's proposal: rule) and you as the decider. A suggestion already decided or superseded is
final - a late decision on it is refused."""
with uc.repository() as repo, uc._decide():
line = _line(repo, line_id)
sugs = repo.bank.suggestions_of(line_id)
sug = next((s for s in sugs if s["suggestion_id"] == suggestion_id), None) if suggestion_id else next((s for s in sugs if s["status"] == "open"), None)
if sug is None or sug["line_id"] != line_id:
raise Refused(404, f"no suggestion for line {line_id}" + (f" named {suggestion_id}" if suggestion_id else ""))
if sug["status"] != "open":
raise RuleBroken(f"Suggestion {sug['suggestion_id']} is already {sug['status']}. It cannot be decided again")
who, today = uc._who(caller), uc._today(repo)
if decision == "reject":
_guard(repo, line_id, who, "reject") # the line's own claim: a reject and an accept cannot both win
if line["lifecycle"] != "open":
raise RuleBroken(f"Line {line_id} is already {line['lifecycle']}")
repo.bank.move_suggestion(sug["suggestion_id"], status="rejected", decided_by=who, decided_on=today)
return {"decision": "rejected", "suggestion_id": sug["suggestion_id"], "line_id": line_id, "lifecycle": "open"}
if decision == "change" and not (account_id or "").strip():
raise RuleBroken("Changing a suggestion needs the account you choose instead")
account = (account_id or sug["account_id"]) if decision == "change" else sug["account_id"]
tax = (tax_code or sug["tax_code"]) if decision == "change" else sug["tax_code"]
cc = (cost_centre if cost_centre is not None else sug["cost_centre"]) if decision == "change" else sug["cost_centre"]
how = "changed" if decision == "change" else ("rule" if sug["model"].startswith("rule:") else "accepted")
done = _post_core(repo, caller, line, account_id=account, tax_code=tax, cost_centre=cc or "", receipt_image=receipt_image or "",
how=how, by=who, supersede=sugs, suggestion_id=sug["suggestion_id"],
suggestion_status="changed" if decision == "change" else "accepted")
return {"decision": "changed" if decision == "change" else "accepted", "suggestion_id": sug["suggestion_id"], "line_id": line_id,
"lifecycle": "posted", "decided_how": how, **done}
# ═══ matching a line to what it is (PAY-21) ══════════════════════════════════════════════════════════════
def _counterpart(repo, line: dict, kind: str, ref: str) -> dict:
"""The counterpart's own figures, or the refusal. Returns {amount, currency, note}."""
amount = Decimal(str(line["amount"]))
if kind == "receipt":
r = repo.receipts.get(ref)
if r is None:
raise Refused(404, f"no receipt {ref}")
if not r.live:
raise RuleBroken(f"Receipt {ref} was returned, so no bank line can be matched to it")
if amount <= 0:
raise RuleBroken("A receipt is matched to a credit; this line is money out")
return {"amount": r.amount, "currency": r.currency}
if kind == "refund":
rf = repo.refunds.get(ref)
if rf is None:
raise Refused(404, f"no refund {ref}")
if rf.status != "paid":
raise RuleBroken(f"Refund {ref} is {rf.status}, not paid, so no bank debit can be matched to it")
if amount >= 0:
raise RuleBroken("A refund is matched to a debit; this line is money in")
return {"amount": rf.amount, "currency": rf.currency}
if kind == "dd_run":
run = repo.dd.run(ref)
if run is None:
raise Refused(404, f"no direct-debit run {ref}")
if run["lifecycle"] != "collected":
raise RuleBroken(f"Run {ref} is {run['lifecycle']}, not collected, so no bank credit can be matched to it")
if amount <= 0:
raise RuleBroken("A collected run is matched to a credit; this line is money out")
return {"amount": run["total"], "currency": run["currency"]}
if kind == "tax_liability":
if amount >= 0:
raise RuleBroken("An HMRC payment is a debit; this line is money in")
try:
data = seams.vat_liabilities()
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
liab = next((x for x in data.get("liabilities", []) if f"{x['period_key']}:{x['charge_type']}" == ref), None)
if liab is None:
raise Refused(404, f"tax has no open liability {ref} (period_key:charge_type)")
return {"amount": Decimal(str(liab["outstanding_amount"])), "currency": line["currency"]}
if kind == "deposit_batch":
b = repo.deposits.get(ref)
if b is None:
raise Refused(404, f"no deposit batch {ref}")
if b["lifecycle"] != "deposited":
raise RuleBroken(f"Deposit batch {ref} is {b['lifecycle']}, not deposited, so no bank credit can be matched to it")
if amount <= 0:
raise RuleBroken("A deposit batch is matched to a credit; this line is money out")
if b["bank_account_id"] != line["bank_account_id"]:
raise RuleBroken(f"Batch {ref} was paid in to {b['bank_account_id']}, not {line['bank_account_id']}")
return {"amount": Decimal(str(b["total"])), "currency": b["currency"]}
if kind == "cash_deposit":
try:
jid, day_s, seq_s = ref.rsplit(":", 2)
e = repo.cash.entry(jid, date.fromisoformat(day_s), int(seq_s))
except ValueError:
raise Refused(422, "a cash deposit is named <cash_journal_id>:<day>:<seq>") from None
if e is None or e["kind"] != "deposit":
raise Refused(404, f"no cash deposit {ref}")
if amount <= 0:
raise RuleBroken("A cash deposit is matched to a credit; this line is money out")
return {"amount": Decimal(str(e["amount"])), "currency": e["currency"]}
if kind == "supplier_payment":
return _supplier_payment(line, ref, amount)
if kind in NOT_YET:
raise Refused(422, f"matching a line to a {kind.replace('_', ' ')} comes with its own story and is not in this release")
raise Refused(422, f"a line is matched to one of: {', '.join(KINDS)}")
RETURNED = ":returned"
def _find_supplier_payment(line: dict, payment_id: str, who: str) -> Optional[dict]:
"""purchasing's payment of this id on THIS bank account, whatever its status (the candidates are listed per status)."""
for status in ("sent", "confirmed", "rejected"):
for p in seams.supplier_payments(line["bank_account_id"], status, who):
if str(p.get("payment_id")) == payment_id:
return {**p, "status": p.get("status") or status}
return None
def _supplier_payment(line: dict, ref: str, amount: Decimal) -> dict:
"""PAY-48. A DEBIT is matched to purchasing's payment (sent: payments then confirms it; confirmed by its clerk already: matched
without a second confirmation). A CREDIT is matched to a REJECTED payment's return, named <payment_id>:returned."""
returned = ref.endswith(RETURNED)
pid = ref[:-len(RETURNED)] if returned else ref
try:
p = _find_supplier_payment(line, pid, "")
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
if p is None:
raise Refused(404, f"purchasing has no payment {pid} on bank account {line['bank_account_id']}")
if returned:
if p["status"] != "rejected":
raise RuleBroken(f"Payment {pid} is {p['status']}, not rejected, so no bank credit can be matched to its return")
if amount <= 0:
raise RuleBroken("A returned payment is matched to a credit; this line is money out")
else:
if p["status"] == "rejected":
raise RuleBroken(f"Payment {pid} was rejected. Match the credit that returned it, with ref {pid}{RETURNED}")
if amount >= 0:
raise RuleBroken("A supplier payment is matched to a debit; this line is money in")
return {"amount": Decimal(str(p["amount"])), "currency": str(p["currency"]).upper(), "payment": p, "payment_id": pid, "returned": returned}
@PAY.use_case(kind="query", route="GET /statement-lines/{line_id}/supplier-payment-candidates", permission=uc.READ)
def supplier_payment_candidates(caller: dict, line_id: Ref) -> dict:
"""The payments purchasing sent from this line's bank account for exactly this amount: what a debit can be matched to (PAY-48)."""
with uc.repository() as repo, uc._decide():
line = _line(repo, line_id)
amount = abs(Decimal(str(line["amount"])))
found = [p for p in seams.supplier_payments(line["bank_account_id"], "sent", uc._who(caller))
if Decimal(str(p["amount"])) == amount and str(p["currency"]).upper() == line["currency"]]
return {"line_id": line_id, "candidates": [{"payment_id": p["payment_id"], "bp_id": p.get("bp_id"), "amount": _s(Decimal(str(p["amount"]))),
"currency": p["currency"], "reference": p.get("reference"), "run_id": p.get("run_id")} for p in found]}
@PAY.use_case(kind="command", route="POST /statement-lines/{line_id}/match", permission=RECONCILE)
def match_statement_line(caller: dict, line_id: Ref, kind: Literal["receipt", "refund", "tax_liability", "dd_run", "deposit_batch", "cash_deposit",
"supplier_payment"], ref: Ref) -> dict:
"""Match an open line to what it is: a receipt (customer credit), a paid refund, a collected direct-debit run, or the HMRC
liability it pays (ref period_key:charge_type, posted bank against the VAT control account). The amount and currency must be
the same, and the counterpart cannot be matched to another line: it is held by ONE line at a time, claimed in the same
transaction as the match, so two lines racing for it collide (409). Also a deposit batch (PAY-28, the line equals the batch total and
is NOT made a receipt again) and a cash deposit (ref cash_journal_id:day:seq, posted bank / transfer clearing on the bank's date). A
debit is matched to purchasing's supplier payment (PAY-48, ref its payment_id): payments confirms it with the bank's reference and the
booking date unless its clerk already did; the credit that returned a rejected payment is matched with ref <payment_id>:returned.
Purchasing posts and allocates its own payment; payments posts nothing for it."""
with uc.repository() as repo, uc._decide():
return match_core(repo, caller, line_id, kind, ref)
def match_core(repo, caller: dict, line_id: str, kind: str, ref: str) -> dict:
line = _line(repo, line_id)
if line["lifecycle"] != "open":
raise RuleBroken(f"Line {line_id} is already {line['lifecycle']}. Undo that first, with a reason")
who, today = uc._who(caller), uc._today(repo)
c = _counterpart(repo, line, kind, ref)
if kind == "deposit_batch" and c["currency"] == line["currency"] and abs(Decimal(str(line["amount"]))) != c["amount"]:
raise Refused(409, f"The batch is {fmt(c['amount'])}; the line {fmt(abs(Decimal(str(line['amount']))))}", gate="batch_total_equals_line",
fix="Find the cheque missing from the batch; for a cash bag, an approver accepts the bank's count (PAY-49)")
if abs(Decimal(str(line["amount"]))) != c["amount"] or c["currency"] != line["currency"]:
raise RuleBroken(f"The line is {fmt(abs(Decimal(str(line['amount']))))} {line['currency']}; the {kind.replace('_', ' ')} is "
f"{fmt(c['amount'])} {c['currency']}. A match needs the same amount and currency")
holder = repo.bank.holder_of(kind, ref)
if holder and holder != line_id:
raise RuleBroken(f"The {kind.replace('_', ' ')} {ref} is already matched to line {holder}. A counterpart is matched to one line")
n = _guard(repo, line_id, who, "match")
repo.claims.take("counterpart", f"{kind}:{ref}", uc._now(), who, line_id)
journal = ""
if kind == "tax_liability":
journal = _post_liability(repo, line, ref, who, n)
elif kind == "cash_deposit":
from brick_payments.application import transfers
journal = transfers.post_side(repo, line, who, n)
jid, day_s, seq_s = ref.rsplit(":", 2)
repo.cash.save_entry({**repo.cash.entry(jid, date.fromisoformat(day_s), int(seq_s)), "statement_line_id": line_id})
elif kind == "deposit_batch":
batch = repo.deposits.get(ref)
if batch.get("kind") == "cash": # v6 (PAY-49): a cheque batch posts nothing (its receipts did); a bag posts bank / cash_in_transit
from brick_payments.application import expected
journal = expected.post_deposit_match(repo, line, batch, who, n)
repo.deposits.save({**batch, "lifecycle": "matched", "statement_line_id": line_id})
elif kind == "supplier_payment" and not c["returned"] and c["payment"]["status"] == "sent":
told = seams.confirm_supplier_payment(c["payment_id"], line_id.split(":", 2)[-1], line["booked_on"], who) # PAY-48
if not told["ok"]:
raise RuleBroken(f"Purchasing did not confirm payment {c['payment_id']} ({told.get('gate') or 'refused'}: {told.get('reason') or 'no reason given'}). "
f"{told.get('fix') or 'Nothing was matched'}")
for s in repo.bank.open_suggestions_of(line_id):
repo.bank.move_suggestion(s["suggestion_id"], status="superseded")
repo.bank.save_line({**line, "lifecycle": "matched", "matched_kind": kind, "matched_ref": ref, "journal_id": journal,
"decided_how": "manual", "decided_by": who, "decided_on": today})
return {"line_id": line_id, "lifecycle": "matched", "matched_kind": kind, "matched_ref": ref, "journal_id": journal}
def _post_liability(repo, line: dict, ref: str, who: str, n: int) -> str:
settings, bank = repo.settings.current(), repo.config.bank(line["bank_account_id"]) or {}
if not settings.vat_control_account_id:
raise RuleBroken("The VAT control account is not set. Set vat_control_account_id in the controls, then match the HMRC payment")
if not bank.get("gl_account_id"):
raise RuleBroken(f"Bank account {line['bank_account_id']} has no ledger account, so the payment cannot be posted")
g = f"{abs(Decimal(str(line['amount']))):.2f}"
posted = seams.post_journal(_posting_key(line, n), "bank_line", line["booked_on"], line["currency"], f"HMRC {ref}",
[{"account_id": settings.vat_control_account_id, "debit": g}, {**seams.account_line(bank["gl_account_id"]), "credit": g}],
who, bank.get("company_code") or "")
if not posted.get("posted"):
raise RuleBroken(f"The ledger did not take the posting ({posted.get('reason', 'no reason given')}). Nothing was matched; try again")
return posted["journal_id"]
@PAY.use_case(kind="command", route="POST /statement-lines/{line_id}/unmatch", permission=RECONCILE)
def unmatch_statement_line(caller: dict, line_id: Ref, reason: text(500, min_length=1)) -> dict:
"""Undo a match or a posting, with a reason: a posting gets a REVERSING journal (event bank_line_reversal), the counterpart is
released, the line is open again - nothing is deleted. Refused once its statement is reconciled, and for a customer credit whose
receipt was made from this very line (reverse the receipt instead)."""
with uc.repository() as repo, uc._decide():
line = _line(repo, line_id)
if line["lifecycle"] == "open":
raise RuleBroken(f"Line {line_id} is open: there is nothing to undo")
st = repo.bank.statement(line["statement_id"])
if st and st["lifecycle"] == "reconciled":
raise RuleBroken(f"Statement {line['statement_id']} is reconciled. Undoing a line of a reconciled statement is not allowed")
if line["matched_kind"] == "receipt":
r = repo.receipts.get(line["matched_ref"])
if r is not None and r.statement_line_id == line_id:
raise RuleBroken("This credit made its receipt. To take it back, reverse the receipt (the bank's return), not the line")
if line["matched_kind"] in ("card_settlement", "payout") or (line["matched_kind"] == "deposit_batch" and (repo.deposits.get(line["matched_ref"]) or {}).get("kind") == "cash"):
raise RuleBroken(f"A {line['matched_kind'].replace('_', ' ')} was posted to the ledger as its own journal: undo it through the module that handed it over "
"(a returned payout is recorded with record_payout_return), not by unmatching the line")
who = uc._who(caller)
repo.claims.take("statement", line["statement_id"], uc._now(), who, "unmatch") # a reconcile of the statement collides with this
claim_n = _guard(repo, line_id, who, "unmatch")
if line["matched_kind"] and line["matched_ref"]:
repo.claims.take("counterpart", f"{line['matched_kind']}:{line['matched_ref']}", uc._now(), who, line_id) # the counterpart's one guard key
if line["matched_kind"] == "transfer" and line["amount"] < 0 and line["matched_ref"]:
raise RuleBroken(f"Line {line_id} is paired with {line['matched_ref']}. Unmatch the arriving credit first; this debit is then in transit again")
if line["matched_kind"] == "transfer" and line["amount"] > 0 and line["matched_ref"]:
debit = repo.bank.line(line["matched_ref"])
if debit:
_guard(repo, debit["line_id"], who, "transfer_release")
repo.bank.save_line({**debit, "matched_ref": ""}) # the debit is in transit again
if line["matched_kind"] == "deposit_batch":
repo.deposits.save({**repo.deposits.get(line["matched_ref"]), "lifecycle": "deposited", "statement_line_id": ""})
if line["matched_kind"] == "cash_deposit":
jid, day_s, seq_s = line["matched_ref"].rsplit(":", 2)
repo.cash.save_entry({**repo.cash.entry(jid, date.fromisoformat(day_s), int(seq_s)), "statement_line_id": ""})
if line["lifecycle"] == "posted" or line["journal_id"]:
bank = repo.config.bank(line["bank_account_id"]) or {}
amount = Decimal(str(line["amount"]))
g = f"{abs(amount):.2f}"
kinds_liab = line["matched_kind"] == "tax_liability"
clearing = line["matched_kind"] in ("transfer", "cash_deposit")
other = ({"account_id": repo.settings.current().vat_control_account_id} if kinds_liab else
{"purpose": "transfer_clearing"} if clearing else {"account_id": line["account_id"]})
side = seams.account_line(bank.get("gl_account_id") or "")
if amount < 0: # the original: Dr other / Cr bank, so reverse it
lines = [{**side, "debit": g}, {**other, "credit": g}]
else:
lines = [{**other, "debit": g}, {**side, "credit": g}]
posted = seams.post_journal(f"{line_id}@{claim_n}", "bank_line_reversal", uc._today(repo), line["currency"], f"Undo: {reason}"[:200], lines, who,
bank.get("company_code") or "")
if not posted.get("posted"):
raise RuleBroken(f"The ledger did not take the reversal ({posted.get('reason', 'no reason given')}). Nothing was undone; try again")
repo.bank.save_line({**line, "lifecycle": "open", "matched_kind": "", "matched_ref": "", "account_id": "", "tax_code": "",
"cost_centre": "", "journal_id": "", "decided_how": "", "decided_by": "", "decided_on": None, "receipt_image": "",
"vat_amount": None, "vat_reclaimed": "", "reversal_reason": reason})
return {"line_id": line_id, "lifecycle": "open", "released": {"kind": line["matched_kind"], "ref": line["matched_ref"]}, "reason": reason}
# ═══ reading ═════════════════════════════════════════════════════════════════════════════════════════════
@PAY.use_case(kind="query", route="GET /statement-lines", permission=uc.READ)
def list_statement_lines(caller: dict, statement_id: Optional[Ref] = None, bank_account_id: Optional[Code] = None,
lifecycle: Optional[Literal["open", "matched", "posted"]] = None, limit: Limit = 200) -> dict:
"""Statement lines, oldest first: every line of an imported statement, whatever became of it."""
with uc.repository() as repo:
rows = repo.bank.lines(statement_id=statement_id, bank_account_id=bank_account_id, lifecycle=lifecycle, limit=limit)
return {"lines": [line_view(r, repo) for r in rows]}
@PAY.use_case(kind="query", route="GET /statement-lines/auto-posted", permission=uc.READ)
def auto_posted_lines(caller: dict, on: Optional[Day] = None) -> dict:
"""The lines an auto rule posted on a day (today by default), for the cashier to confirm or undo the same day."""
with uc.repository() as repo:
day = uc._date(on, "on") or uc._today(repo)
return {"on": day.isoformat(), "lines": [line_view(r) for r in repo.bank.posted_by_rule_on(day)]}
@PAY.use_case(kind="query", route="GET /statement-lines/{line_id}", permission=uc.READ)
def get_statement_line(caller: dict, line_id: Ref) -> dict:
"""One statement line with the suggestions still open on it."""
with uc.repository() as repo:
return line_view(_line(repo, line_id), repo)
@PAY.use_case(kind="command", route="POST /statement-lines/{line_id}/suggest", permission=SUGGEST, tool=False)
def suggest_statement_match(caller: dict, line_id: Ref) -> dict:
"""The system looks at one open line: the rules first (exactly one rule: its proposal, or - in auto mode - the posting), else the
AI's suggestion from the payee and past postings. A suggestion is a row keyed <line_id>#<n>; running it again supersedes an open
one and never a decided one; an auto rule posts a line once (the journal is keyed on the line)."""
with uc.repository() as repo, uc._decide():
_line(repo, line_id)
return _process_one(repo, caller, line_id) or {"suggested": 0}
# ═══ rules (PAY-23) ══════════════════════════════════════════════════════════════════════════════════════
@PAY.use_case(kind="command", route="POST /reconciliation-rules/{rule_id}", permission=RULE_WRITE)
def put_reconciliation_rule(caller: dict, rule_id: Code, direction: Literal["debit", "credit"], account_id: Code, tax_code: Code,
mode: Literal["suggest", "auto"] = "suggest", bank_account_id: Optional[Code] = None,
text_contains: Optional[text(200)] = None, counterparty_contains: Optional[text(200)] = None,
amount_min: Optional[Money] = None, amount_max: Optional[Money] = None,
cost_centre: Optional[Code] = None) -> dict:
"""Save a rule for recurring lines ("debit, text contains SERVICE CHARGE, account BARC-01 -> account 7900, tax code E"). A rule is a
DATED row that starts TOMORROW (business_timezone) - changing it adds a new row, history stays. In suggest mode a matching line
gets the rule's proposal for the cashier to accept; in auto mode (which needs pay:settings:controls as well) the line is posted
with decided_how auto and listed for the cashier to confirm or undo the same day. A rule never matches a line to a customer's invoice."""
if mode == "auto" and not auth.has_permission(caller, uc.CONTROLS):
raise Refused(403, f"missing permission: {uc.CONTROLS} (an auto rule posts without a person: only the owner's control turns it on)")
with uc.repository() as repo, uc._decide():
if not (text_contains or counterparty_contains or amount_min is not None or amount_max is not None):
raise RuleBroken("A rule needs something to recognise a line by: text, counterparty, or an amount band")
if amount_min is not None and amount_max is not None and amount_min > amount_max:
raise Refused(422, "amount_min is more than amount_max")
if bank_account_id and repo.config.bank(bank_account_id) is None:
raise Refused(422, f"there is no bank account {bank_account_id}")
start = uc._today(repo) + timedelta(days=1)
rows = [r for r in repo.bank.rule_rows() if r["rule_id"] == rule_id]
if any(r["valid_from"] == start for r in rows):
raise RuleBroken(f"Rule {rule_id} already has a change that starts {start}. It can be changed again from the day after")
repo.bank.add_rule({"rule_id": rule_id, "valid_from": start, "bank_account_id": bank_account_id or "", "direction": direction,
"text_contains": text_contains or "", "counterparty_contains": counterparty_contains or "",
"amount_min": amount_min, "amount_max": amount_max, "account_id": account_id, "tax_code": tax_code,
"cost_centre": cost_centre or "", "mode": mode, "set_by": uc._who(caller)}, uc._now())
return {"rule_id": rule_id, "valid_from": start.isoformat(), "mode": mode, "direction": direction, "account_id": account_id,
"tax_code": tax_code, "bank_account_id": bank_account_id or "any"}
@PAY.use_case(kind="query", route="GET /reconciliation-rules", permission=uc.READ)
def list_reconciliation_rules(caller: dict) -> dict:
"""Every rule row, with the one in force today marked."""
with uc.repository() as repo:
today = uc._today(repo)
inforce = {r["rule_id"]: r["valid_from"] for r in repo.bank.rules_on(today)}
out = []
for r in repo.bank.rule_rows():
out.append({**{k: (v.isoformat() if isinstance(v, date) else _s(v) if isinstance(v, Decimal) else v) for k, v in r.items()
if k not in ("created_at", "updated_at")}, "in_force": inforce.get(r["rule_id"]) == r["valid_from"]})
return {"rules": out}
@PAY.use_case(kind="command", route="POST /house-banks/{house_bank_id}", permission="pay:bank_account:write", idempotent=True)
def put_house_bank(caller: dict, house_bank_id: Code, name: Name, sort_code: Optional[text(10)] = None, bic: Optional[text(11)] = None,
address: Optional[text(300)] = None, sun: Optional[text(6)] = None, creditor_id: Optional[text(35)] = None) -> dict:
"""The banks we hold accounts with: their Bacs Service User Number (SUN, six digits, for direct debits) and SEPA creditor id."""
if sun and not (len(sun) == 6 and sun.isdigit()):
raise Refused(422, "a Service User Number is six digits")
with uc.repository() as repo, uc._decide():
row = {"house_bank_id": house_bank_id, "name": name, "bic": bic or "", "sort_code": sort_code or "", "address": address or "",
"sun": sun or "", "creditor_id": creditor_id or ""}
repo.bank.put_house_bank(row)
return row
@PAY.use_case(kind="query", route="GET /house-banks", permission=uc.READ)
def list_house_banks(caller: dict) -> dict:
"""The banks we hold accounts with."""
with uc.repository() as repo:
return {"house_banks": [{k: v for k, v in h.items() if k not in ("created_at", "updated_at")} for h in repo.bank.house_banks()]}