dodil/erp-brick-pos
DODIL ERP brick: point of sale
src/brick_pos/domain/receipt.py
"""The Receipt aggregate: one sale as printed for the customer, the legal document. The rules, and nothing else.
open (lines rung, payments taken) -> finalised a finalised receipt is immutable and never reopens
A refund is a NEW receipt that names the original (POS-11, `refund_of` below and domain/refund.py); an OPEN receipt can be voided (a void
record, its number never reused); a correction is a later document, never an update. Tips are not part of the bill: they ride on the payment line that carried them and belong to staff (a liability,
never revenue, unless the customer's setting says otherwise)."""
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from brick_pos.domain.errors import RuleBroken
from brick_pos.domain.events import PaymentTaken, ReceiptFinalised, ReceiptRefunded, ReceiptRung, ReceiptVoided
from brick_pos.domain.values import PaymentMethod, ReceiptKind, ReceiptState, SessionState, TerminalStatus
MAX_LINES = 200
ZERO = Decimal("0.00")
@dataclass
class ReceiptLine:
line_no: int
product_id: str
quantity: Decimal
unit_code: str
unit_price: Decimal
line_gross: Decimal
tax_category: str
tax_code: str
rate_pct: Decimal
description: str = ""
vat: Decimal = ZERO
original_line_no: Optional[int] = None # a refund line names the line of the original it gives back
restock: Optional[bool] = None # a refund line: the item goes back on the shelf
kind: str = "product" # product | voucher_sale (a gift voucher sold: outside VAT, a liability, no stock)
voucher_code: str = "" # the voucher a voucher_sale line issues
recipe_source: str = "" # "" (the product itself is consumed) | pos | manufacturing: where the components came from
recipe_ref: str = "" # the recipe_id (product) or manufacturing's bom_id, frozen
recipe_version: Optional[int] = None # and its version, frozen: a later change never changes this sale
modifiers: str = "" # the modifiers chosen, comma-separated
@dataclass
class LineComponent:
"""What one receipt line consumed from stock, frozen when it was rung: component 1 is the product itself when it has no recipe,
else the recipe's (or manufacturing's) components. Inventory's move_ref is pos:<receipt_id>:<line_no>:<component_no>."""
line_no: int
component_no: int
product_id: str
quantity: Decimal
unit_code: str
@dataclass
class PaymentLine:
pay_no: int
method: str
amount: Decimal # applied to the bill; what the customer handed over is amount + change_amount
tip_amount: Decimal = ZERO
change_amount: Decimal = ZERO
terminal_txn_ref: str = ""
terminal_status: str = ""
idempotency_key: str = ""
voucher_code: str = ""
@property
def counts(self) -> bool:
"""cash and a voucher (whose balance was claimed) always; a card only once the terminal authorised it (a declined or still
pending line is kept, and counts for nothing)."""
return self.method in (PaymentMethod.CASH.value, PaymentMethod.VOUCHER.value) or self.terminal_status == TerminalStatus.AUTHORISED.value
def receipt_id_of(session_id: str, receipt_no: int) -> str:
return f"{session_id}-{receipt_no:04d}"
@dataclass
class Receipt:
receipt_id: str
session_id: str = ""
shop_id: str = ""
register_id: str = ""
receipt_no: int = 0
business_date: Optional[date] = None
currency: str = ""
kind: str = ReceiptKind.SALE.value
eat_in: Optional[bool] = None
gross: Decimal = ZERO
net: Decimal = ZERO
vat: Decimal = ZERO
status: str = ReceiptState.OPEN.value
created_by: str = ""
created_sub: str = ""
finalised_at: Optional[datetime] = None
original_receipt_id: str = ""
void_reason: str = ""
voided_by: str = ""
voided_at: Optional[datetime] = None
lines: list = field(default_factory=list)
payments: list = field(default_factory=list)
components: list = field(default_factory=list)
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
# ── ring_up: none -> open ───────────────────────────────────────────────────────────────────────────────────
@classmethod
def ring_up(cls, *, session, receipt_no: int, lines: list, eat_in: Optional[bool], by: str, sub: str,
components: Optional[list] = None) -> "Receipt":
"""Lines rung in an open session, priced and taxed already (domain/pricing.py). The total is theirs."""
if session.status != SessionState.OPEN.value:
raise RuleBroken(f"session {session.session_id} is {session.status}: sales go in an open session",
gate="session_open", fix="open a session on the register")
if not lines:
raise RuleBroken("a receipt has at least one line", gate="bounds", fix="ring up a product")
if len(lines) > MAX_LINES:
raise RuleBroken(f"a receipt has at most {MAX_LINES} lines", gate="bounds", fix="split the sale in two receipts")
it = cls(receipt_id=receipt_id_of(session.session_id, receipt_no), session_id=session.session_id,
shop_id=session.shop_id, register_id=session.register_id, receipt_no=receipt_no,
business_date=session.business_date, currency=session.currency, eat_in=eat_in, created_by=by,
created_sub=sub, lines=list(lines), components=list(components or []))
it.gross = sum((ln.line_gross for ln in lines), ZERO)
it.vat = sum((ln.vat for ln in lines), ZERO)
it.net = it.gross - it.vat
if it.gross <= 0:
raise RuleBroken("a receipt's total is more than zero", gate="bounds", fix="ring up a priced product")
it._record(ReceiptRung(it.receipt_id, it.gross, by))
return it
# ── derived ─────────────────────────────────────────────────────────────────────────────────────────────────
def components_of(self, line: "ReceiptLine") -> list:
"""What the line takes from stock: its frozen components, or (a receipt rung before recipes) the product itself."""
mine = sorted((c for c in self.components if c.line_no == line.line_no), key=lambda c: c.component_no)
return mine or [LineComponent(line.line_no, 1, line.product_id, line.quantity, line.unit_code)]
def paid(self) -> Decimal:
return sum((p.amount for p in self.payments if p.counts), ZERO)
def remaining(self) -> Decimal:
return self.gross - self.paid()
def tips(self) -> Decimal:
return sum((p.tip_amount for p in self.payments if p.counts), ZERO)
# ── take_payment: open -> open ──────────────────────────────────────────────────────────────────────────────
def take_payment(self, *, method: str, amount: Decimal, tip: Decimal, terminal_txn_ref: str, terminal_status: str,
session_status: str, by: str, voucher_code: str = "") -> PaymentLine:
"""A tender: cash (change given when more than the bill is handed over), a gift voucher (its balance claimed by the use
case; no change, no tip) or a card the terminal answered for - authorised, declined, or `pending` when the terminal timed out
(the register asks the terminal for the outcome by the line's key before it offers to try again). The terminal's idempotency key
is the payment line, `<receipt_id>:<pay_no>`, never the receipt."""
if self.status == ReceiptState.VOIDED.value:
raise RuleBroken(f"receipt {self.receipt_id} was voided: it takes no payment", gate="receipt_open",
fix="ring the sale up again on a new receipt")
if self.status != ReceiptState.OPEN.value:
raise RuleBroken(f"receipt {self.receipt_id} is {self.status}: a finalised receipt is never changed",
gate="receipt_open", fix="a refund is a new receipt that names this one")
if session_status != SessionState.OPEN.value:
raise RuleBroken(f"session {self.session_id} is {session_status}: payments are taken in an open session",
gate="session_open", fix="open a session on the register")
try:
m = PaymentMethod(method)
except ValueError:
raise RuleBroken(f"{method!r} is not a way to pay: use cash or card", gate="method", fix="send method cash or card") from None
waiting = [p.pay_no for p in self.payments if p.terminal_status == TerminalStatus.PENDING.value]
if waiting:
raise RuleBroken(f"the terminal has not answered for payment line {waiting[0]} ({self.receipt_id}:{waiting[0]}): ask it for the "
"outcome by that key before another payment, so a card is never charged twice", gate="terminal_pending",
fix="resolve the pending card line first")
remaining = self.remaining()
if remaining <= 0:
raise RuleBroken(f"receipt {self.receipt_id} is paid in full", gate="within_remainder", fix="finalise it")
pay_no = len(self.payments) + 1
change, applied, ref, status, code = ZERO, amount, "", "", ""
if m == PaymentMethod.CASH:
if amount > remaining:
applied, change = remaining, amount - remaining
elif m == PaymentMethod.VOUCHER:
if not voucher_code:
raise RuleBroken("a voucher line names the voucher", gate="voucher_code", fix="send voucher_code")
if amount > remaining:
raise RuleBroken(f"a voucher pays at most the bill: {remaining} is left, and a voucher gives no change", gate="within_remainder",
fix=f"send amount {remaining} at most")
if tip > 0:
raise RuleBroken("a tip is paid in cash or on a card, not with a voucher", gate="tip_method", fix="take the tip in cash or by card")
code = voucher_code
else:
if not terminal_txn_ref and terminal_status != TerminalStatus.PENDING.value:
raise RuleBroken("a card line names the terminal's transaction", gate="terminal_ref", fix="send terminal_txn_ref")
if terminal_status not in (TerminalStatus.AUTHORISED.value, TerminalStatus.DECLINED.value, TerminalStatus.PENDING.value):
raise RuleBroken("the terminal's answer is authorised, declined or pending (it timed out)", gate="terminal_status",
fix="ask the terminal for the outcome by its key before sending")
if amount > remaining:
raise RuleBroken(f"a card is charged for the bill, not more: {remaining} is left", gate="within_remainder",
fix=f"send amount {remaining} at most (the tip is separate)")
ref, status = terminal_txn_ref, terminal_status
if status == TerminalStatus.DECLINED.value:
tip = ZERO
line = PaymentLine(pay_no=pay_no, method=m.value, amount=applied, tip_amount=tip, change_amount=change,
terminal_txn_ref=ref, terminal_status=status, idempotency_key=f"{self.receipt_id}:{pay_no}", voucher_code=code)
self.payments.append(line)
self._record(PaymentTaken(self.receipt_id, pay_no, m.value, applied, tip, by))
return line
# ── resolve_card_payment: a pending card line -> authorised | declined ──────────────────────────────────────
def resolve_payment(self, *, pay_no: int, terminal_status: str, terminal_txn_ref: str, session_status: str, by: str) -> Optional[PaymentLine]:
"""The terminal's outcome for a line that timed out, asked for by the line's key `<receipt_id>:<pay_no>`. pending -> authorised
| declined, once: the same outcome again changes nothing (None), a different one after it is final is refused (a line never
goes back to pending, and an authorised one is not later declined). A declined line keeps no tip; a new line is the retry."""
if self.status != ReceiptState.OPEN.value:
raise RuleBroken(f"receipt {self.receipt_id} is {self.status}: only an open receipt waits for its terminal", gate="receipt_open",
fix="a refund is a new receipt that names this one")
if session_status != SessionState.OPEN.value:
raise RuleBroken(f"session {self.session_id} is {session_status}: payments are settled in an open session", gate="session_open",
fix="open a session on the register")
line = next((p for p in self.payments if p.pay_no == pay_no), None)
if line is None or line.method != PaymentMethod.CARD.value:
raise RuleBroken(f"{self.receipt_id} has no card payment line {pay_no}", gate="payment_line", fix="resolve a card line of this receipt")
if terminal_status not in (TerminalStatus.AUTHORISED.value, TerminalStatus.DECLINED.value):
raise RuleBroken("the terminal's outcome is authorised or declined", gate="terminal_status",
fix="ask the terminal for the outcome by the line's key, and send what it says")
if line.terminal_status != TerminalStatus.PENDING.value:
if line.terminal_status == terminal_status and (terminal_status == TerminalStatus.DECLINED.value
or line.terminal_txn_ref == terminal_txn_ref):
return None
raise RuleBroken(f"payment line {pay_no} is already {line.terminal_status}: the terminal's answer is final", gate="terminal_final",
fix="take a new payment line if the customer pays again")
if terminal_status == TerminalStatus.AUTHORISED.value and not terminal_txn_ref:
raise RuleBroken("an authorised card line names the terminal's transaction", gate="terminal_ref", fix="send terminal_txn_ref")
line.terminal_status = terminal_status
if terminal_status == TerminalStatus.AUTHORISED.value:
line.terminal_txn_ref = terminal_txn_ref
else:
line.tip_amount = ZERO
self._record(PaymentTaken(self.receipt_id, pay_no, line.method, line.amount if line.counts else ZERO, line.tip_amount, by))
return line
def voucher_lines(self) -> list:
"""The voucher tenders taken on this receipt (their balances were claimed when they were taken)."""
return [p for p in self.payments if p.method == PaymentMethod.VOUCHER.value]
# ── void_open_receipt: open -> voided ───────────────────────────────────────────────────────────────────────
def void(self, *, reason: str, terminal_reversals: set, session_status: str, at: datetime, by: str) -> Optional[ReceiptVoided]:
"""Cancel a receipt before it is final. It stays as a void record with its lines and its number (fiscal law counts voids and
a number is never reused). Every card line the terminal authorised must have been reversed there first: the register sends the
terminal's references, and those lines become `reversed`. Cash taken on an open receipt was never in the takings: it goes back
to the customer. None: already voided, changing nothing."""
if self.status == ReceiptState.VOIDED.value:
return None
if self.status != ReceiptState.OPEN.value:
raise RuleBroken(f"receipt {self.receipt_id} is {self.status}: a final receipt is never voided", gate="receipt_open",
fix="a refund is a new receipt that names this one")
if session_status != SessionState.OPEN.value:
raise RuleBroken(f"session {self.session_id} is {session_status}: receipts are voided in an open session",
gate="session_open", fix="open a session on the register")
reason = (reason or "").strip()
if not reason:
raise RuleBroken("a void names its reason", gate="reason_required", fix="send reason")
waiting = [p for p in self.payments if p.terminal_status == TerminalStatus.PENDING.value]
if waiting:
raise RuleBroken(f"the terminal has not answered for payment line {waiting[0].pay_no} ({waiting[0].idempotency_key}): ask it for the "
"outcome first, then reverse it there if it was charged", gate="terminal_pending",
fix="resolve the pending card line, then void")
live = [p for p in self.payments if p.method == PaymentMethod.CARD.value and p.terminal_status == TerminalStatus.AUTHORISED.value]
unreversed = [p.terminal_txn_ref for p in live if p.terminal_txn_ref not in terminal_reversals]
if unreversed:
raise RuleBroken(f"the card payment {', '.join(unreversed)} is still authorised on the terminal", gate="card_reversed",
fix="reverse it on the terminal, then send its reference in terminal_reversals")
for p in live:
p.terminal_status = TerminalStatus.REVERSED.value
self.status, self.void_reason, self.voided_by, self.voided_at = ReceiptState.VOIDED.value, reason, by, at
return self._record(ReceiptVoided(self.receipt_id, reason, by))
# ── finalise_receipt: open -> finalised ─────────────────────────────────────────────────────────────────────
def finalise(self, *, at: datetime, by: str) -> Optional[ReceiptFinalised]:
"""Final when the payments sum EXACTLY to the bill. None: already final, changing nothing."""
if self.status == ReceiptState.FINALISED.value:
return None
if self.status == ReceiptState.VOIDED.value:
raise RuleBroken(f"receipt {self.receipt_id} was voided: it is never finalised", gate="receipt_open",
fix="ring the sale up again on a new receipt")
if self.status != ReceiptState.OPEN.value:
raise RuleBroken(f"receipt {self.receipt_id} is {self.status}", gate="receipt_open", fix="the accountant decides")
if self.paid() != self.gross:
raise RuleBroken(f"the payments come to {self.paid()}, the bill is {self.gross}", gate="paid_in_full",
fix=f"take {self.remaining()} more" if self.remaining() > 0 else "the payments are over the bill")
self.status, self.finalised_at = ReceiptState.FINALISED.value, at
return self._record(ReceiptFinalised(self.receipt_id, self.gross, by))
def refund_of(*, original: "Receipt", session, receipt_no: int, lines: list, method: str, terminal_txn_ref: str, reason: str, at: datetime,
by: str, sub: str) -> Receipt:
"""A refund: a NEW, final receipt that names the original. Amounts are negative (the lines' gross and VAT, the one payment line);
a line's quantity is the quantity given back. The money goes back the way it came: cash, or the card through the terminal as a
referenced refund (its idempotency key is this payment line, `<refund_receipt_id>:1`, never the sale's)."""
it = Receipt(receipt_id=receipt_id_of(session.session_id, receipt_no), session_id=session.session_id, shop_id=session.shop_id,
register_id=session.register_id, receipt_no=receipt_no, business_date=session.business_date,
currency=session.currency, kind=ReceiptKind.REFUND.value, original_receipt_id=original.receipt_id,
created_by=by, created_sub=sub, lines=list(lines), status=ReceiptState.FINALISED.value, finalised_at=at)
it.gross = sum((ln.line_gross for ln in lines), ZERO)
it.vat = sum((ln.vat for ln in lines), ZERO)
it.net = it.gross - it.vat
ref = terminal_txn_ref if method == PaymentMethod.CARD.value else ""
it.payments.append(PaymentLine(pay_no=1, method=method, amount=it.gross, terminal_txn_ref=ref,
terminal_status=TerminalStatus.AUTHORISED.value if ref else "",
idempotency_key=f"{it.receipt_id}:1"))
it._record(ReceiptRefunded(it.receipt_id, original.receipt_id, it.gross, reason, by))
return it