dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/payments.py
"""Payment runs and supplier payments (PUR-14 to PUR-16, PUR-19, PUR-29, PUR-59 to PUR-61, PUR-66).
DODIL never moves the money. A run proposes what falls due (one payment per supplier and currency), a SECOND person approves it (never the proposer,
nor the clerk who recorded any bill in it), the system writes a file for the bank (pain.001, BACS Standard 18 or CSV), a person uploads it, and a
payment is 'paid' only when the bank's reference and date are confirmed — by a person, or by payments' statement match. Allocations are plain INSERTs
claimed per bill, so two clerks allocating the same bill cannot over-pay it; a correction is a reversing row."""
import json
import uuid
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform import work
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from erp_platform.types import BigId, IsoDate, Limit, text
from brick_purchasing.application import base
from brick_purchasing.application.base import (
PAY_APPROVE,
PAY_CONFIRM,
PAY_EXPORT,
PAY_READ,
PAY_WRITE,
PURCHASING,
_decide,
_same_version,
_sub,
_today,
_who,
unit,
view,
)
from brick_purchasing.application.bills import account_for, outstanding
from brick_purchasing.application.inputs import AllocationIn, Code, Currency, Money, Offset, OptCode, Reason, Ref, Version
from brick_purchasing.application.master import bank_in_force, profile_on
from brick_purchasing.domain import files
from brick_purchasing.domain.approval import identities
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.payment import Allocation, PaymentRun, RunItem, SupplierPayment
from brick_purchasing.domain.readiness import no, ok, require, views
from brick_purchasing.domain.values import ZERO, money_round
from brick_purchasing.infrastructure import vault
from brick_purchasing.infrastructure.seams import Missing, TableObjects, ports
repository = lambda: base.repository() # noqa: E731
HOOK_DOWN_PAYMENTS = lambda repo, run, pay_date, today: ([], [], []) # noqa: E731 — (items, left_out, payments) the down-payment module adds
HOOK_BILL_HELD = lambda repo, bill: None # noqa: E731 — a reason a bill stays out of a run (an uncleared down payment)
def objects():
return ports.objects or _default_objects()
def _default_objects():
from brick_purchasing.infrastructure.seams import S3Objects
s3 = S3Objects()
return s3 if s3.configured else TableObjects(repository)
# ═══ views ═════════════════════════════════════════════════════════════════════════════════════════
def payment_view(repo, p: SupplierPayment, version: Optional[int] = None) -> dict:
out = view(p)
t = repo.conn
from sqlalchemy import select
from brick_purchasing.infrastructure.tables import pur_run_item
items = [dict(r._mapping) for r in t.execute(select(pur_run_item).where(pur_run_item.c.payment_id == p.payment_id))] if p.run_id else []
out["bills"] = [{"bill_id": i["bill_id"], "amount": i["amount"]} for i in items]
if version is not None:
out["version"] = version
return out
def run_view(repo, r: PaymentRun, version: int) -> dict:
out = view(r)
out["left_out"] = json.loads(r.left_out or "[]")
out["version"] = version
pays = repo.payments.of_run(r.run_id)
out["payments"] = [payment_view(repo, p) for p in sorted(pays, key=lambda x: x.payment_id)]
out["total"] = sum((p.amount for p in pays if p.currency == (r.currency or p.currency)), ZERO) if pays else ZERO
if r.status in ("draft", "approved"):
out["readiness"] = views(run_gates(repo, r, pays, _today(repo)))
return out
def _payee_bank(repo, p: SupplierPayment, r: PaymentRun, today: date) -> Optional[dict]:
"""The bank row a payment is paid to: the one in force when the run was proposed (a run proposed today uses today's details), stored on the
payment; else the one in force on the pay date."""
if p.payee_bank_valid_from:
return repo.banks.get(p.bp_id, p.payee_bank_valid_from)
return bank_in_force(repo, p.bp_id, r.pay_date or today)
def _payees(repo, r: PaymentRun, pays: list, today: date) -> list:
out = []
for p in sorted(pays, key=lambda x: x.payment_id):
bank = _payee_bank(repo, p, r, today)
acct = ""
if bank:
try:
acct = vault.unseal(bank["iban_or_account"])
except Exception: # noqa: BLE001 — no key: the file cannot be written
acct = ""
prof = profile_on(repo, p.bp_id, r.pay_date or today)
name = (bank or {}).get("account_name") or prof.get("legal_name") or f"supplier {p.bp_id}"
out.append(files.Payee(p.payment_id, name, acct, (bank or {}).get("bic_or_sort_code") or "", p.amount, p.currency, p.reference, (bank or {}).get("country") or ""))
return out
def run_gates(repo, r: PaymentRun, pays: list, today: date) -> list:
"""The readiness of approving a run: every payee's bank details approved and in force on the pay date; for a BACS file, GBP, a UK sort code and
account per payee, the service user number set and the pay date at least the BACS cycle away."""
gates = []
missing = [p.bp_id for p in pays if bank_in_force(repo, p.bp_id, r.pay_date or today) is None and (profile_on(repo, p.bp_id, today).get("payment_method") != "outside")]
gates.append(ok("bank_details") if not missing else
no("bank_details", f"supplier {', '.join(map(str, missing))} has no approved bank details in force on the pay date", "Suppliers: bank details (PUR-30), or leave the payee out"))
s = base.Settings(repo, today)
fmt = r.file_format or s["payment_file_format"]
if fmt == "bacs_std18":
payees = _payees(repo, r, pays, today)
gates += files.bacs_ready(payees, r.currency or "GBP", s["bacs_service_user_number"], r.pay_date or today, today, int(s["bacs_cycle_working_days"]))
return gates
# ═══ proposing ══════════════════════════════════════════════════════════════════════════════════════
def _bank_account(bank_account_id: str) -> dict:
if not ports.installed("payments"):
return {"bank_account_id": bank_account_id, "name": bank_account_id, "currency": "", "gl_account_id": ""}
try:
rows = ports.payments.bank_accounts().get("bank_accounts") or []
except Missing as m:
raise base._refuse(m, "payments") from None
found = next((a for a in rows if a["bank_account_id"] == bank_account_id), None)
if found is None:
raise Refused(422, f"there is no bank account {bank_account_id!r} in payments; fix: add it there (payments owns the bank accounts)")
return found
def _originator(r: PaymentRun) -> dict:
"""OUR paying account's details for one file, asked of payments (PAY-27, machine permission pay:bank_account:read_unmasked, every read audited there
with this purpose). Returned to the caller's stack frame only: never stored, logged or returned. A refusal is the gate `originator_account`."""
def gate(why: str, fix: str) -> Refused:
return Refused(409, f"originator_account: {why}; fix: {fix}")
if not ports.installed("payments"):
raise gate("payments is not part of this installation, so our own account number cannot be read", "install payments and record the account there")
try:
d = ports.payments.account_details(r.bank_account_id, f"payment_run:{r.run_id}")
except Missing as m:
if m.status == 403:
raise gate("payments refused purchasing the unmasked account", "grant purchasing's service client pay:bank_account:read_unmasked") from None
if m.status == 404:
raise gate(f"payments has no paying account {r.bank_account_id!r}", "record it in payments with purpose pay") from None
if m.status == 503:
raise gate("payments could not log the read of our account, so it refused it", "try again once payments can write its audit row") from None
raise gate(f"payments did not answer ({m.status or 'no reply'})", "try again") from None
sort, number, iban = (d.get("sort_code") or "").strip(), (d.get("account_number") or "").strip(), (d.get("iban") or "").replace(" ", "")
return {"bacs": f"{sort} {number}".strip() if sort and number else "", "iban": iban, "bic": d.get("bic") or ""}
def _cis(repo, bp_id: int, pay_date: date) -> Optional[dict]:
prof = profile_on(repo, bp_id, pay_date)
if not prof.get("cis_subcontractor"):
return None
if not ports.installed("tax"):
raise Refused(409, f"cis_verified: supplier {bp_id} is a CIS subcontractor and tax is not part of this installation; fix: verify them in Tax (CIS)")
try:
got = ports.tax.cis_subcontractor(bp_id, pay_date)
except Missing as m:
raise Refused(409, f"cis_verified: supplier {bp_id}: {m.reason}; fix: verify them in Tax (CIS), or pay at the unmatched rate tax answers") from None
bad = [g for g in got.get("readiness") or [] if not g.get("met") and g.get("gate") == "cis_verified"]
if bad and got.get("status") != "unmatched":
raise Refused(409, f"cis_verified: supplier {bp_id}: {bad[0].get('reason', 'no verification in force')}; fix: verify them in Tax (CIS), or pay at the unmatched rate tax answers")
return got
@PURCHASING.use_case(kind="command", route="POST /payment-runs", permission=PAY_WRITE)
def propose_run(caller: dict, bank_account_id: Code, pay_date: IsoDate, file_format: Optional[Code] = None, currency: Optional[Currency] = None) -> dict:
"""Propose a payment run of what falls due by the pay date from a bank account: approved, undisputed bills due by then and approved down
payments due by then, one payment per supplier and currency. A CIS subcontractor is paid net of the deduction on the labour share paid, at
tax's rate on the pay date. Disputed bills, bills without approved bank details and bills already in an open run are left out and named.
`bank_account_id` is OUR paying account in payments; the number itself is never an input and never stored: the export asks payments for it (PAY-27)."""
def run():
with repository() as repo, _decide():
today = _today(repo)
when = base._day(pay_date, "pay_date")
if when < today:
raise RuleBroken(f"not_past: a pay date cannot be before today ({today}); fix: pick today or later")
acct = _bank_account(bank_account_id)
s = base.Settings(repo, today)
fmt = file_format or s["payment_file_format"]
if fmt not in files.EXT:
raise Refused(422, f"file_format is one of {', '.join(files.EXT)}")
rid = f"run:{bank_account_id}:{when}:{work.now().strftime('%Y%m%dT%H%M%S%f')}-{uuid.uuid4().hex[:6]}"
r = PaymentRun(rid, bank_account_id, acct.get("gl_account_id") or "", currency or acct.get("currency") or "", when, fmt, "draft",
_who(caller), _sub(caller))
left, groups = [], {}
open_ = {b for b, _ in repo.runs.open_items()}
repo.banks.cache, repo.suppliers.cache = {}, {} # a supplier's bank and profile rows are read once, however many bills it has
heads = repo.bills.approved()
sums = repo.allocations.sums([b.bill_id for b in heads if b.kind != "credit"])
credit_used = repo.allocations.credit_used([b.bill_id for b in heads if b.kind == "credit"])
for b in heads:
if currency and b.currency != currency:
continue
if acct.get("currency") and b.currency != acct["currency"]:
continue
if b.kind == "credit":
if b.total - credit_used.get(b.bill_id, ZERO) > 0:
left.append({"bill_id": b.bill_id, "reason": "an unallocated credit: allocate it against a bill first"})
continue
if b.kind != "bill" or b.source_kind == "odoo" and False:
continue
o = b.total - sums.get(b.bill_id, ZERO)
if o <= 0 or (b.due_date and b.due_date > when):
continue
if b.status == "disputed":
left.append({"bill_id": b.bill_id, "reason": "disputed: " + b.dispute_reason})
continue
if b.bill_id in open_:
left.append({"bill_id": b.bill_id, "reason": "already in an open payment run"})
continue
held = HOOK_BILL_HELD(repo, b)
if held:
left.append({"bill_id": b.bill_id, "reason": held})
continue
if bank_in_force(repo, b.bp_id, when) is None and profile_on(repo, b.bp_id, when).get("payment_method") != "outside":
left.append({"bill_id": b.bill_id, "reason": "no approved bank details in force on the pay date"})
continue
groups.setdefault((b.bp_id, b.currency), []).append((b, o))
extra_items, extra_left, extra_pays = HOOK_DOWN_PAYMENTS(repo, r, when, today)
left += extra_left
pays = []
for n, ((bp, cur), items) in enumerate(sorted(groups.items()), 1):
pid = f"run:{rid}:{bp}:{cur}"
gross = sum((o for _, o in items), ZERO)
cis = _cis(repo, bp, when)
deduction, rate, labour_total = ZERO, ZERO, ZERO
p = SupplierPayment(pid, rid, bp, cur, gross, gross, ZERO, ZERO, "", "", "", bank_account_id, "", f"{rid}/{n}", None, "planned", "run",
(bank_in_force(repo, bp, today) or bank_in_force(repo, bp, when) or {}).get("valid_from"))
by_bill = {}
for ln in repo.bills.lines_of([b.bill_id for b, _ in items]): # approved() answers heads: the lines are read here
by_bill.setdefault(ln.bill_id, []).append(ln)
for b, o in items:
share = o / b.total if b.total else Decimal(1)
lab = money_round(sum((ln.net_amount for ln in by_bill.get(b.bill_id, []) if ln.cis_part == "labour"), ZERO) * share, cur)
mat = money_round(sum((ln.net_amount for ln in by_bill.get(b.bill_id, []) if ln.cis_part == "materials"), ZERO) * share, cur)
labour_total += lab
r.items.append(RunItem(rid, pid, b.bill_id, o, "bill", lab, mat, "|".join(sorted(identities(b.recorded_by_email, b.recorded_by_sub)))))
if cis:
rate = Decimal(str(cis.get("rate_pct") or 0))
deduction = money_round(labour_total * rate / 100, cur)
p.cis_deduction, p.cis_rate_pct, p.cis_status = deduction, rate, cis.get("status") or ""
p.amount = gross - deduction
pays.append(p)
r.items += extra_items
pays += extra_pays
if not pays and not extra_items:
raise RuleBroken("nothing falls due by " + str(when) + (": " + "; ".join(f"{x['bill_id']} ({x['reason']})" for x in left) if left else ""))
r.left_out = json.dumps(left)
r.currency = r.currency or (pays[0].currency if pays else "")
# "a bill already in an open run is left out" holds under concurrency: two runs proposed at once both read the bill as free,
# and both claim run:<bill>#<finished runs that held it + 1> - exactly one commits it, the other re-runs and finds it taken
held = repo.runs.closed_inclusions({it.bill_id for it in r.items if it.bill_id})
repo.claims.claim_many([(f"run:{b}", held.get(b, 0) + 1) for b in sorted({it.bill_id for it in r.items if it.bill_id})], rid)
v = repo.runs.save(r, 0)
for p in pays:
repo.payments.save(p, 0)
return run_view(repo, r, v)
return unit(None, run)
def _load_run(repo, run_id: str) -> tuple:
got = repo.runs.get(run_id)
if got is None:
raise Refused(404, f"no payment run {run_id}")
return got
@PURCHASING.use_case(kind="command", route="POST /payment-runs/{run_id}/approve", permission=PAY_APPROVE)
def approve_run(caller: dict, run_id: Ref, version: Optional[Version] = None) -> dict:
"""Release a run: the approver is not the clerk who proposed it, nor the clerk who recorded ANY bill in it (same email OR same sub — segregation
of duties), every payee's bank details are approved and in force on the pay date, and for a BACS file the run is GBP with UK details, a service
user number and a pay date at least the BACS cycle away. Records who, and today."""
def run():
with repository() as repo, _decide():
r, v = _load_run(repo, run_id)
if r.status == "approved":
return run_view(repo, r, v)
_same_version("the run", v, version)
today = _today(repo)
r.approve(caller.get("email"), caller.get("sub"), today, run_gates(repo, r, repo.payments.of_run(r.run_id), today))
return run_view(repo, r, repo.runs.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /payment-runs/{run_id}/cancel", permission=PAY_APPROVE, idempotent=True, destructive=True)
def cancel_run(caller: dict, run_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Cancel a run before its file is written, with a reason: its bills are free for another run."""
def run():
with repository() as repo, _decide():
r, v = _load_run(repo, run_id)
if r.status == "cancelled":
return run_view(repo, r, v)
_same_version("the run", v, version)
r.cancel(reason)
planned = repo.payments.get_many(p.payment_id for p in repo.payments.of_run(run_id))
for got in planned.values():
got[0].status, got[0].reject_reason = "rejected", "the run was cancelled"
repo.payments.save(got[0], got[1])
return run_view(repo, r, repo.runs.save(r, v))
return unit(version, run)
# ═══ the bank file ══════════════════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="command", route="POST /payment-runs/{run_id}/export", permission=PAY_EXPORT, idempotent=True)
def export_bank_file(caller: dict, run_id: Ref) -> dict:
"""Write the approved run as a file for the bank — pain.001.001.03 or .09, BACS Standard 18 (one 100-character detail record per payment, the
contra record, header and trailer labels with the totals in pence) or CSV — kept ONCE as the object raw/purchasing/runs/<run_id>.<ext>. A second
export returns the same file. DODIL does not send it: a person signs it in the bank's portal."""
def run():
with repository() as repo, _decide():
r, v = _load_run(repo, run_id)
today = _today(repo)
if r.file_key:
body = objects().get(r.file_key) or b""
return {"run_id": run_id, "file_key": r.file_key, "file_format": r.file_format, "sha256": r.file_sha256, "bytes": len(body), "replay": True,
"content": body.decode("utf-8", "replace")}
if r.status != "approved":
raise RuleBroken(f"the run is {r.status}: only an approved run is written as a file")
pays = [p for p in repo.payments.of_run(run_id) if p.status != "rejected"]
payees = _payees(repo, r, pays, today)
s = base.Settings(repo, today)
fmt = r.file_format or s["payment_file_format"]
if fmt == "bacs_std18":
mine = _originator(r) # in memory for this file only
require(files.bacs_ready(payees, r.currency or "GBP", s["bacs_service_user_number"], r.pay_date, today, int(s["bacs_cycle_working_days"]),
mine["bacs"]))
text_ = files.bacs_std18(payees, service_user=s["bacs_service_user_number"], user_name=profile_name(), originator=mine["bacs"],
pay_date=r.pay_date, today=today, run_ref=r.run_id.split(":")[-1][:18])
elif fmt.startswith("pain.001"):
mine = _originator(r)
text_ = files.pain001(payees, version=fmt, run_id=r.run_id, debtor_name=profile_name(), debtor_iban=mine["iban"],
debtor_bic=mine["bic"], pay_date=r.pay_date, created=work.now(), currency=r.currency or "GBP")
else:
text_ = files.csv_file(payees, pay_date=r.pay_date)
import hashlib
data = text_.encode("utf-8")
key = f"raw/purchasing/runs/{r.run_id}.{files.EXT[fmt]}"
objects().put(key, data, "text/plain" if fmt != "csv" else "text/csv")
r.mark_exported(key, hashlib.sha256(data).hexdigest(), today)
repo.runs.save(r, v)
return {"run_id": run_id, "file_key": key, "file_format": fmt, "sha256": r.file_sha256, "bytes": len(data), "replay": False, "content": text_}
return unit(None, run)
def profile_name() -> str:
import os
return os.environ.get("COMPANY_NAME", "DODIL CUSTOMER")
# ═══ allocations ═════════════════════════════════════════════════════════════════════════════════════
def allocate(repo, caller: dict, source_kind: str, source_id: str, bill_id: str, amount: Decimal, on: date) -> Allocation:
"""Allocate money (or a credit) to a bill: refused when it is more than the bill's outstanding. Takes the bill's claim sequence with a plain
INSERT, so two clerks allocating the same bill at once collide, and the loser re-reads the outstanding."""
got = repo.bills.get(bill_id)
if got is None:
raise Refused(404, f"no bill {bill_id}")
b = got[0]
if b.status != "approved":
raise RuleBroken(f"the bill is {b.status}: money is allocated to an approved bill")
if amount <= 0:
raise RuleBroken("an allocation is above zero")
out = outstanding(repo, b)
if amount > out:
raise RuleBroken(f"not_over_paid: {amount} is more than the outstanding {out} on bill {b.bill_no or bill_id}")
seq = repo.claims.take(f"alloc:{bill_id}", source_id)
a = Allocation(source_kind, source_id, bill_id, seq, amount, on, _who(caller))
repo.allocations.put(a)
return a
def post_payment_journal(caller: dict, payment_id: str, event: str = "pay") -> dict:
"""Post a confirmed payment (event pay) or its reversal (event reverse): payables debited for the gross, the bank's ledger account credited for
what was paid and the CIS liability for the deduction. A refusal keeps the payment and lists the posting for a retry."""
with repository() as repo:
got = repo.payments.get(payment_id)
if not got:
raise Refused(404, f"no payment {payment_id}")
p = got[0]
if p.source_kind == "odoo" or not ports.installed("gl"):
return {"payment_id": payment_id, "posted": False, "note": "not posted: the ledger is not installed" if not ports.installed("gl") else "migrated history is not re-posted"}
if event == "pay" and p.posted_journal_id:
return {"payment_id": payment_id, "posted": True, "journal_id": p.posted_journal_id}
on = p.paid_on or _today(repo)
run = repo.runs.get(p.run_id)[0] if p.run_id else None
bank_acct = (run.gl_account_id if run else "") or (_bank_account(p.bank_account_id).get("gl_account_id") if p.bank_account_id else "")
rev = event == "reverse"
d, c = ("credit", "debit") if rev else ("debit", "credit")
lines = [{d: str(p.gross), "account_id": account_for(repo, "supplier_down_payment" if p.kind == "down_payment" else "payables", "*", on), "memo": f"payment {p.reference or p.payment_id}", "bp_id": str(p.bp_id)},
{c: str(p.amount), "account_id": bank_acct or account_for(repo, "bank", "*", on), "memo": f"bank {p.bank_reference}"}]
if p.cis_deduction:
lines.append({c: str(p.cis_deduction), "purpose": "cis_liability", "memo": "CIS deduction", "bp_id": str(p.bp_id)})
body = {"journal_id": str(derived_id(payment_id, event)), "source_module": "purchasing", "source_document": payment_id, "event": event,
"journal_date": (_today(repo) if rev else on).isoformat(), "currency": p.currency, "memo": f"supplier payment {p.reference}", "lines": lines}
note, jid = "", ""
try:
jid = str(ports.gl.post_journal(body).get("journal_id") or body["journal_id"])
except Missing as m:
note = f"the ledger refused ({m.status}): " + ("; ".join(f"{g.get('gate')}: {g.get('reason')}" for g in m.gates) or m.reason)
with repository() as repo:
q, v = repo.payments.get(payment_id)
if event == "pay":
q.posted_journal_id = jid
q.post_note = note
repo.payments.save(q, v)
return {"payment_id": payment_id, "posted": bool(jid), "journal_id": jid, "note": note}
# ═══ confirming ═════════════════════════════════════════════════════════════════════════════════════
def confirm_core(caller: dict, payment_id: str, bank_reference: str, paid_on: date, by: str) -> dict:
def run():
with repository() as repo, _decide():
got = repo.payments.get(payment_id)
if got is None:
raise Refused(409, "payment_open: there is no such payment")
p, v = got
today = _today(repo)
if p.run_id:
run_, rv = _load_run(repo, p.run_id)
if run_.status not in ("exported", "done") and p.status == "planned":
raise RuleBroken(f"payment_open: the run is {run_.status}: a payment is confirmed once its file was written and uploaded")
if not p.confirm(bank_reference, paid_on, today, by):
return {"payment_id": payment_id, "status": "confirmed", "replay": True}, False
items = _items_of(repo, payment_id)
for i in items:
allocate_payment_item(repo, caller, p, i, paid_on)
repo.payments.save(p, v)
if p.run_id:
_finish(repo, p.run_id)
return {"payment_id": payment_id, "status": "confirmed", "replay": False}, True
out, fresh = unit(None, run)
if fresh:
post_payment_journal(caller, payment_id, "pay")
record_cis(caller, payment_id)
return out
def _items_of(repo, payment_id: str) -> list:
from sqlalchemy import select
from brick_purchasing.infrastructure.tables import _make, pur_run_item
return [_make(RunItem, dict(r._mapping)) for r in repo.conn.execute(select(pur_run_item).where(pur_run_item.c.payment_id == payment_id))]
def allocate_payment_item(repo, caller, p: SupplierPayment, i: RunItem, paid_on: date) -> None:
if i.kind == "down_payment":
from brick_purchasing.application import down_payments
down_payments.on_paid(repo, caller, i, p, paid_on)
return
seq = repo.claims.take(f"alloc:{i.bill_id}", p.payment_id)
repo.allocations.put(Allocation("payment", p.payment_id, i.bill_id, seq, i.amount, paid_on, _who(caller)))
def _finish(repo, run_id: str) -> None:
got = repo.runs.get(run_id)
if got and got[0].finish_if_done(repo.payments.of_run(run_id)):
repo.runs.save(got[0], got[1])
def record_cis(caller: dict, payment_id: str) -> None:
"""Tell tax about the CIS deduction of a paid payment (POST /cis/deductions, keyed pur:<payment_id>); keep its answer on the payment and a
deduction row per bill. A retry records nothing new. Purchasing never computes the return."""
with repository() as repo:
got = repo.payments.get(payment_id)
p = got[0]
if not p.cis_deduction or p.cis_ref or not ports.installed("tax"):
return
items = _items_of(repo, payment_id)
body = {"payment_ref": f"pur:{payment_id}", "bp_id": str(p.bp_id), "paid_on": p.paid_on.isoformat(), "bill_ids": [i.bill_id for i in items],
"gross_ex_vat": str(sum((i.labour_paid + i.materials_paid for i in items), ZERO)), "materials": str(sum((i.materials_paid for i in items), ZERO))}
try:
got_ = ports.tax.cis_deduct(body)
except Missing as m:
with repository() as repo:
q, v = repo.payments.get(payment_id)
q.post_note = f"tax refused the CIS deduction ({m.status}): {m.reason}"
repo.payments.save(q, v)
return
with repository() as repo:
q, v = repo.payments.get(payment_id)
q.cis_ref, q.cis_tax_month = f"pur:{payment_id}", str(got_.get("tax_month") or "")
repo.payments.save(q, v)
for i in _items_of(repo, payment_id):
repo.cis.put({"payment_id": payment_id, "bill_id": i.bill_id, "labour_paid": i.labour_paid, "materials_paid": i.materials_paid,
"rate_pct": Decimal(str(got_.get("rate_pct") or q.cis_rate_pct)), "deduction": money_round(i.labour_paid * q.cis_rate_pct / 100, q.currency),
"tax_month": q.cis_tax_month, "verification_number": "", "status": "recorded", "deduction_ref": f"pur:{payment_id}:{i.bill_id}"})
@PURCHASING.use_case(kind="command", route="POST /payments/{payment_id}/confirm", permission=PAY_WRITE, idempotent=True)
def confirm_paid(caller: dict, payment_id: Ref, bank_reference: text(100, min_length=1), paid_on: IsoDate) -> dict:
"""Confirm what the bank paid: the bank's reference and the bank's date (today or earlier, never later). Allocates the payment to its bills
(dated by the bank), posts payables / bank (and the CIS liability) to the ledger, records any CIS deduction with tax, and finishes the run when
every payment is confirmed or rejected. Confirming twice answers the same and changes nothing."""
return confirm_core(caller, payment_id, bank_reference, base._day(paid_on, "paid_on"), _who(caller))
@PURCHASING.use_case(kind="command", route="POST /supplier-payments/{payment_id}/confirm", permission=PAY_CONFIRM, idempotent=True, tool=False)
def confirm_paid_from_bank(caller: dict, payment_id: Ref, bank_ref: text(100, min_length=1), paid_on: IsoDate) -> dict:
"""Payments matched a debit on the bank statement to this supplier payment: confirm it with the bank's reference and date, exactly as a person
would (confirm_paid). A second confirm answers the same, unchanged; an unknown or cancelled payment is refused (409 {gate: payment_open})."""
got = confirm_core(caller, payment_id, bank_ref, base._day(paid_on, "paid_on"), "payments")
return {"payment_id": payment_id, "status": "confirmed", "replay": bool(got.get("replay"))}
@PURCHASING.use_case(kind="command", route="POST /payments/{payment_id}/reject", permission=PAY_WRITE, idempotent=True)
def reject_payment(caller: dict, payment_id: Ref, reason: Reason) -> dict:
"""Record a payment the bank rejected, with a reason: the allocations are reversed with reversing rows and a reversing journal (dated today), so
the bills are open again; a CIS deduction is reversed with tax. The original rows are never edited."""
def run():
with repository() as repo, _decide():
got = repo.payments.get(payment_id)
if got is None:
raise Refused(404, f"no payment {payment_id}")
p, v = got
if p.status == "rejected":
return {"payment_id": payment_id, "status": "rejected", "replay": True}, False
was_paid = p.status == "paid"
today = _today(repo)
p.reject(reason)
if was_paid:
for a in repo.allocations.of_source("payment", payment_id):
if a.amount > 0 and not any(x.reverses_seq == a.seq and x.bill_id == a.bill_id for x in repo.allocations.of_source("payment", payment_id)):
seq = repo.claims.take(f"alloc:{a.bill_id}", payment_id)
repo.allocations.put(Allocation("payment", payment_id, a.bill_id, seq, -a.amount, today, _who(caller), a.seq))
repo.payments.save(p, v)
if p.run_id:
_finish(repo, p.run_id)
return {"payment_id": payment_id, "status": "rejected", "replay": False, "reversed": was_paid}, was_paid
out, reverse = unit(None, run)
if reverse:
post_payment_journal(caller, payment_id, "reverse")
with repository() as repo:
p = repo.payments.get(payment_id)[0]
if p.cis_ref and ports.installed("tax"):
try:
ports.tax.cis_reverse(p.cis_ref, {"bp_id": str(p.bp_id), "reason": reason, "reversed_on": _today_of().isoformat()})
with repository() as repo:
for r in repo.cis.all(payment_id=payment_id):
r["status"] = "reversed"
repo.cis.put(r)
except Missing:
pass
return out
def _today_of() -> date:
with repository() as repo:
return _today(repo)
@PURCHASING.use_case(kind="command", route="POST /payments", permission=PAY_WRITE, idempotent=True)
def record_payment(caller: dict, bp_id: BigId, bank_account_id: Code, bank_reference: text(100, min_length=1), paid_on: IsoDate, amount: Money, currency: Currency,
allocations: list[AllocationIn]) -> dict:
"""Record a payment made outside a run, or one transfer covering several bills, and allocate it: each bill's outstanding falls by its share.
Keyed pay:<bank account>:<bank reference>, so recording it again returns it. Allocating more than the payment, or more than a bill's outstanding,
is refused; two clerks allocating the same bill at once cannot over-pay it (the loser re-reads). Dated by the bank (today or earlier)."""
pid = f"pay:{bank_account_id}:{bank_reference.strip()}"
def run():
with repository() as repo, _decide():
have = repo.payments.get(pid)
if have:
return payment_view(repo, have[0], have[1]), False
today = _today(repo)
when = base._day(paid_on, "paid_on")
if when > today:
raise RuleBroken(f"not_future: a payment cannot be dated after today ({today}); fix: confirm it on the day the bank shows it")
cur = base._currency(currency)
if amount <= 0:
raise RuleBroken("a payment is above zero")
if sum((a.amount for a in allocations), ZERO) > amount:
raise RuleBroken("the allocations add up to more than the payment")
p = SupplierPayment(pid, "", bp_id, cur, amount, amount, ZERO, ZERO, "", "", "", bank_account_id, bank_reference.strip(), bank_reference.strip(), when,
"paid", "outside", None, _who(caller))
for a in allocations:
got = repo.bills.get(a.bill_id)
if got is None or got[0].bp_id != bp_id or got[0].currency != cur:
raise Refused(422, f"bill {a.bill_id} is not an approved bill of supplier {bp_id} in {cur}")
allocate(repo, caller, "payment", pid, a.bill_id, a.amount, when)
p.gross = p.amount = sum((a.amount for a in allocations), ZERO) or amount
v = repo.payments.save(p, 0)
return payment_view(repo, p, v), True
out, fresh = unit(None, run)
if fresh:
post_payment_journal(caller, pid, "pay")
return out
# ═══ reading ═══════════════════════════════════════════════════════════════════════════════════════
@PURCHASING.use_case(kind="query", route="GET /payment-runs/{run_id}", permission=PAY_READ, etag="version")
def get_payment_run(caller: dict, run_id: Ref) -> dict:
"""One run: its payments (each with the bills it pays and any CIS deduction), what was left out and why, the file, and the readiness gates."""
with repository() as repo:
r, v = _load_run(repo, run_id)
return run_view(repo, r, v)
@PURCHASING.use_case(kind="query", route="GET /payment-runs", permission=PAY_READ)
def list_payment_runs(caller: dict, limit: Limit = 50, offset: Offset = 0) -> dict:
"""The payment runs (heads), newest first."""
with repository() as repo:
rows, total = repo.runs.page(limit, offset)
return {"total": total, "runs": [{k: getattr(r, k) for k in ("run_id", "bank_account_id", "pay_date", "status", "file_format", "file_key")} for r in rows]}
STATUS = {"planned": "sent", "paid": "confirmed", "rejected": "rejected"}
@PURCHASING.use_case(kind="query", route="GET /supplier-payments", permission=PAY_READ, tool=False)
def list_supplier_payments(caller: dict, bank_account_id: Optional[OptCode] = None, status: Optional[OptCode] = None, from_: Optional[IsoDate] = None,
to: Optional[IsoDate] = None) -> dict:
"""Supplier payments for payments' statement matching (status=sent: the candidates a statement debit is matched to) and for tax's cash accounting
(from/to: confirmed payments dated by the bank). Each lists the bills it pays and their amounts."""
inv = {v: k for k, v in STATUS.items()}
with repository() as repo:
rows = repo.payments.find(bank_account_id or None, inv.get(status) if status else None, base._day(from_, "from"), base._day(to, "to"))
out = []
for p in rows:
if p.status == "planned" and p.run_id:
r = repo.runs.get(p.run_id)
if not r or r[0].status != "exported":
continue
if from_ or to:
if p.status != "paid":
continue
d = payment_view(repo, p)
out.append({"payment_id": p.payment_id, "run_id": p.run_id, "bp_id": str(p.bp_id), "bank_account_id": p.bank_account_id, "amount": p.amount,
"currency": p.currency, "reference": p.reference, "status": STATUS[p.status], "paid_on": p.paid_on, "bills": d["bills"]})
return {"payments": out}
@PURCHASING.use_case(kind="query", route="GET /payments/{payment_id}", permission=PAY_READ)
def get_supplier_payment(caller: dict, payment_id: Ref) -> dict:
"""One supplier payment with the bills it pays, its CIS deduction and its bank reference."""
with repository() as repo:
got = repo.payments.get(payment_id)
if got is None:
raise Refused(404, f"no payment {payment_id}")
return payment_view(repo, got[0], got[1])