dodil/erp-brick-expenses
DODIL ERP brick: expenses
src/brick_expenses/domain/report.py
"""The expense report: one claim by one person, approved as a whole (module.yaml workflows.expense_report).
Plain Python - no SQLAlchemy, FastAPI, MCP or identity. The rules a finance manager checks first live here:
- the state machine only moves FORWARD (a returned claim continues as revision n+1; nothing is re-entered)
- nobody approves their own claim: matched on email OR sub (either one is the same person), at every level
- the chain is derived from hr on the submission date and FIXED there; approvals of one revision never count for the next
- money is exact; the journal of an approved revision balances to the penny
"""
import json
from dataclasses import dataclass, field, replace
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Callable, Optional
from brick_expenses.domain.errors import RuleBroken
from brick_expenses.domain.money import ZERO, money
STATES = ("draft", "submitted", "approved", "posted", "paid", "returned", "rejected", "cancelled", "post_failed")
TERMINAL = ("paid", "returned", "rejected", "cancelled")
LINES_MAX = 200
LINE_MAX = Decimal("1e13") # 200 lines of this still fit NUMERIC(18,2) with room to spare
# transition key -> (states it may start from, the state it ends in). Every arrow goes forward; tests try the reverse of each.
MOVES = {
"submit_report": (("draft",), "submitted"),
"complete_approval": (("submitted",), "approved"),
"return_report": (("submitted",), "returned"),
"reject_report": (("submitted",), "rejected"),
"cancel_report": (("draft",), "cancelled"),
"post_report": (("approved", "post_failed"), "posted"),
"fail_report_post": (("approved",), "post_failed"),
"confirm_paid": (("posted",), "paid"),
}
@dataclass(frozen=True)
class Identity:
"""Who a person is to this brick: both of their logins' names. The same person under either is the same person."""
email: str = ""
sub: str = ""
def same_as(self, other: "Identity") -> bool:
a, b = _norm(self.email), _norm(other.email)
s, t = (self.sub or "").strip(), (other.sub or "").strip()
return bool((a and a == b) or (s and s == t))
def _norm(v: Optional[str]) -> str:
return (v or "").strip().lower()
@dataclass(frozen=True)
class Level:
level: int
kind: str # manager | finance
person_id: Optional[int] = None
email: str = ""
sub: str = ""
rerouted_from: Optional[str] = None
@dataclass
class Line:
line_no: int
category_code: str
expense_date: date
merchant: str
amount: Decimal
vat_amount: Decimal = ZERO
paid_with: str = "own"
kind: str = "receipt"
cost_centre: str = ""
receipt_key: str = ""
description: str = ""
receipt_currency: str = ""
receipt_amount: Optional[Decimal] = None
city: str = ""
nights: Optional[int] = None
attendees: list = field(default_factory=list) # names of the people a meal was for
justification: str = "" # the employee's reason for a warned line; the approver sees it
policy_results: list = field(default_factory=list) # [{rule_code, valid_from, mode, message, ...}] as last evaluated
unconfirmed: list = field(default_factory=list) # fields a reader suggested that the employee has not confirmed yet
stale: bool = False
stale_reason: str = ""
recomputed: list = field(default_factory=list) # [{field, was, now, cause, on}]: what a recompute changed and why (EXP-20)
card_txn_id: str = "" # the card transaction a paid_with=card line is matched to (EXP-06)
fx_rate: Optional[Decimal] = None # gl's spot rate on the receipt date for a foreign line, frozen at submission (EXP-05)
treatment_code: str = "" # how tax treats the merchant (uk_business unless the receipt shows otherwise)
contents: dict = field(default_factory=dict) # the receipt's particulars for tax's content check (supplier, address, VAT number ...)
tax_code: str = "" # tax's answer, kept as our copy (EXP-18)
vat_reclaimable: Decimal = ZERO
vat_non_reclaimable: Decimal = ZERO
tax: dict = field(default_factory=dict) # {digest, gates: [unmet gates shown to the approver], checked}
rebill_bp_id: Optional[int] = None # the customer this cost is re-billed to (EXP-21, an extension through ar)
rebill_markup_pct: Decimal = ZERO
rebill_tax_category: str = ""
ai: dict = field(default_factory=dict) # the receipt reader's footprint on the line: {finding_no, model, filled, suggested, changed}
miles: Optional[Decimal] = None # mileage: miles or kilometres driven (NUMERIC 19,4)
trip_from: Optional[datetime] = None # per diem: the trip's start and end (the traveller's own clock)
trip_to: Optional[datetime] = None
meals: list = field(default_factory=list) # per diem: [{date, meal}] the employer provided
calc: dict = field(default_factory=dict) # how the amount was computed: bands / days, the rate row used, the cumulative before
rate_country: str = ""
rate_variant: str = ""
possible_duplicate_of: str = "" # another claim's line the same receipt image was already attached to (EXP-16)
receipt_vat: Optional[Decimal] = None # the VAT the employee read off the receipt, in the RECEIPT's currency (tax's answer replaces it)
@dataclass
class Step:
level: int
kind: str
decision: str # approved | returned | rejected
decided_by: str
decided_sub: str
decided_at: object
reason: str = ""
approver_person_id: Optional[int] = None
delegate_for: str = "" # the approver this person decided for, when they were a delegate (EXP-09)
@dataclass
class Report:
report_id: str
person_id: int
company_code: str
bp_id: Optional[int]
claimant: Identity
currency: str
title: str = ""
revision: int = 1
status: str = "draft"
submitted_on: Optional[date] = None
submitted_at: object = None
chain: list = field(default_factory=list) # [Level], fixed at submission
waiting_level: Optional[int] = None
waiting_since: Optional[date] = None # the business day the claim began waiting at this level (EXP-17's business days start here)
journal_id: Optional[int] = None
payout_ref: Optional[str] = None
approved_on: Optional[date] = None
paid_on: Optional[date] = None
payment_id: Optional[str] = None
fail_reason: Optional[str] = None
total_amount: Optional[Decimal] = None
reimbursable_amount: Optional[Decimal] = None
netted_amount: Decimal = ZERO # personal card spend netted off the payout (EXP-13)
lines: list = field(default_factory=list)
steps: list = field(default_factory=list) # the CURRENT revision's decisions
# ── totals: derived from the lines, never incremented ─────────────────────────────────────
def total(self) -> Decimal:
return money(sum((ln.amount for ln in self.lines), ZERO))
def reimbursable(self) -> Decimal:
return money(sum((ln.amount for ln in self.lines if ln.paid_with == "own"), ZERO))
def payout_amount(self) -> Decimal:
"""What payments is asked to pay: what the claim reimburses less the personal card spend netted from it."""
return money(self.reimbursable() - (self.netted_amount or ZERO))
def level_row(self, n: int) -> Optional[Level]:
return next((lv for lv in self.chain if lv.level == n), None)
def decided_levels(self) -> set:
return {s.level for s in self.steps if s.decision == "approved"}
def next_level(self) -> Optional[Level]:
done = self.decided_levels()
return next((lv for lv in self.chain if lv.level not in done), None)
# ── the state machine ──────────────────────────────────────────────────────────────────
def move(self, key: str) -> None:
frm, to = MOVES[key]
if self.status not in frm:
raise RuleBroken(f"Report {self.report_id} is {self.status}: {key.replace('_', ' ')} needs it to be {' or '.join(frm)}")
self.status = to
def ensure_draft(self) -> None:
if self.status != "draft":
raise RuleBroken(f"Report {self.report_id} is {self.status}: its lines are frozen once it is submitted "
"(a returned claim continues as a new revision)")
# ── lines ──────────────────────────────────────────────────────────────────────────────────────
def check_line(line: Line, *, today: date, currency: str, report_currency: str) -> None:
if line.expense_date > today:
raise RuleBroken(f"A receipt dated {line.expense_date} is in the future (today is {today})")
if line.vat_amount > line.amount:
raise RuleBroken("The VAT cannot be more than the receipt's total")
def receipts_present(report: Report, threshold: Decimal, required: Callable[[str], bool]) -> list:
"""The lines above `threshold` whose category needs a receipt and that have none: named (line numbers)."""
return [ln.line_no for ln in report.lines
if ln.kind == "receipt" and ln.amount > threshold and required(ln.category_code) and not ln.receipt_key]
def readiness(report: Report, threshold: Decimal, required: Callable[[str], bool], may_submit: tuple = (True, ""),
unresolved_cards: Optional[list] = None) -> list:
"""Each gate of submit_report as {gate, met, reason, fix}: what the screen shows and what a refusal names. The order is the spec's
(module.yaml workflows.expense_report.readiness)."""
from brick_expenses.domain import policy
missing = receipts_present(report, threshold, required)
unconfirmed = [ln.line_no for ln in report.lines if ln.unconfirmed]
broken = policy.blocks(report.lines)
cards = unresolved_cards or []
stale = [ln.line_no for ln in report.lines if ln.stale]
return [
{"gate": "has_lines", "met": bool(report.lines), "reason": "a claim with no lines", "fix": "add what you spent"},
{"gate": "lines_ready", "met": not unconfirmed,
"reason": f"lines with fields not confirmed: {', '.join(map(str, unconfirmed))}" if unconfirmed else "",
"fix": "confirm or correct each suggested field"},
{"gate": "receipts_present", "met": not missing,
"reason": f"lines above {threshold:.2f} with no receipt: {', '.join(map(str, missing))}" if missing else "",
"fix": "add the receipt image to each of them"},
{"gate": "policy_blocks", "met": not broken,
"reason": "; ".join(f"line {n}: {m}" for n, m in broken) if broken else "", "fix": "remove or change the line"},
{"gate": "card_resolved", "met": not cards,
"reason": f"card transactions not matched or marked personal: {', '.join(cards)}" if cards else "",
"fix": "My expenses: match them"},
{"gate": "not_stale", "met": not stale,
"reason": "lines computed from rules that changed: " + "; ".join(
f"{ln.line_no} ({ln.stale_reason or 'a rule changed'})" for ln in report.lines if ln.stale) if stale else "",
"fix": "recompute (one click)"},
{"gate": "claimant_may_claim", "met": may_submit[0], "reason": may_submit[1],
"fix": "only people employed on the date (or within the leaver window) claim"},
]
# ── the approval chain ───────────────────────────────────────────────────────────────────────
def derive_chain(claimant: Identity, total: Decimal, threshold: Decimal, manager_at: Callable[[int], Optional[dict]]) -> list:
"""Level 1 is the claimant's manager from hr; a level whose approver IS the claimant goes one level up (never to the claimant),
and with no manager at all the claim goes to finance. Finance looks too when the total is ABOVE the threshold. The chain is
fixed at submission. `manager_at(n)` is hr's derived manager n levels up: {person_id, email, pool_user_sub} or None."""
levels: list = []
skipped: Optional[str] = None
for n in (1, 2):
m = manager_at(n)
if not m:
continue
who = Identity(m.get("email") or "", m.get("pool_user_sub") or "")
if who.same_as(claimant):
skipped = who.email or who.sub
continue
levels.append(Level(1, "manager", m.get("person_id"), who.email, who.sub, rerouted_from=skipped))
break
if total > threshold or not levels:
levels.append(Level(len(levels) + 1, "finance"))
return levels
def check_decider(report: Report, who: Identity, has_permission: Callable[[str], bool], *, what: str = "approve") -> Level:
"""The level this person decides now, or the rule that stops them. Separation of duties: nobody decides their own claim, nor a
second level of a claim they already decided - under either login."""
if report.status != "submitted":
raise RuleBroken(f"Report {report.report_id} is {report.status}: only a submitted claim is {what}d")
lv = report.next_level()
if lv is None:
raise RuleBroken(f"Every level of report {report.report_id} revision {report.revision} is already decided")
if who.same_as(report.claimant):
raise RuleBroken("You are the claimant: the claim goes to the next approver up", )
if any(Identity(s.decided_by, s.decided_sub).same_as(who) for s in report.steps if s.decision == "approved"):
raise RuleBroken("You decided an earlier level of this claim: another person decides this one")
return lv
def journal_lines(report: Report, purposes: dict, centre: str) -> list:
"""The approved revision's journal: a debit per (purpose, cost centre) for the cost - the line's amount LESS the VAT tax says is reclaimable -
a debit of that input VAT to purpose `tax` under tax's code, and the credits for what was paid: ONE employee_payable for what the employee is
owed and a card_liability per company card for what the card paid. Both sides equal to the penny, or it is not posted."""
debits: dict = {}
vat: dict = {}
for ln in report.lines:
reclaim = min(ln.vat_reclaimable or ZERO, ln.amount)
key = (purposes[ln.category_code], ln.cost_centre or centre)
debits[key] = debits.get(key, ZERO) + ln.amount - reclaim
if reclaim > ZERO:
vkey = (ln.tax_code, ln.cost_centre or centre)
vat[vkey] = vat.get(vkey, ZERO) + reclaim
out = [{"purpose": p, **({"cost_centre": c} if c else {}), "debit": f"{money(v):.2f}"} for (p, c), v in sorted(debits.items())]
out += [{"purpose": "tax", **({"tax_code": t} if t else {}), **({"cost_centre": c} if c else {}), "debit": f"{money(v):.2f}"}
for (t, c), v in sorted(vat.items())]
owed = report.reimbursable()
if owed > ZERO or not any(ln.paid_with == "card" for ln in report.lines):
out.append({"purpose": "employee_payable", **({"bp_id": str(report.bp_id)} if report.bp_id else {}), "credit": f"{owed:.2f}"})
by_card: dict = {}
for ln in report.lines:
if ln.paid_with == "card":
card = ln.card_txn_id.split(":", 2)[1] if ln.card_txn_id else ""
by_card[card] = by_card.get(card, ZERO) + ln.amount
out += [{"purpose": "card_liability", **({"memo": f"card {c}"} if c else {}), "credit": f"{money(v):.2f}"} for c, v in sorted(by_card.items())]
dr = sum(Decimal(x["debit"]) for x in out if "debit" in x)
cr = sum(Decimal(x["credit"]) for x in out if "credit" in x)
if dr != cr:
raise RuleBroken(f"The journal does not balance ({dr:.2f} against {cr:.2f})")
return out
# ── the leaver window (EXP-19) ──────────────────────────────────────────────────────────────
def claims_until(last_day: date, days: int) -> date:
return last_day + timedelta(days=days)
def may_claim(employed: bool, leaver_until: Optional[date], today: date) -> tuple:
if employed:
return True, ""
if leaver_until is not None and today <= leaver_until:
return True, ""
if leaver_until is not None:
return False, f"you left and the window to claim closed on {leaver_until}"
return False, "you are not employed on this date"
def chain_json(chain: list) -> str:
return json.dumps([level.__dict__ for level in chain], sort_keys=True)
def chain_from_json(raw: Optional[str]) -> list:
return [Level(**d) for d in json.loads(raw)] if raw else []
def copy_lines(lines: list) -> list:
return [replace(ln) for ln in lines]