dodil/erp-brick-pos
DODIL ERP brick: point of sale
src/brick_pos/domain/journal.py
"""The session's summary journal for gl (POS-20, POS-29): the takings of ONE session as one balanced journal, not a line per coffee.
Pure rules over figures the use case derived from the session's own rows (one statement each, however many receipts):
debit pos_cash_in_till the cash taken (and cash tips) credit revenue the net of sales, by cost centre
card_clearing card takings and their tips tax VAT, one line per tax code
voucher_liability vouchers redeemed tips_payable tips (a liability to staff)
cash_over_short a drawer short pos_cash_in_till paid-outs, safe drops, an over count
shop_supplies paid-outs voucher_liability vouchers sold
pos_cash_in_safe safe drops cash_over_short a drawer over
A refund (negative amounts) lands on the opposite side of the same lines. A journal that does not balance to the penny is never
built: it is a rule broken here, not a 422 from gl. journal_id is derived from (source_document, event), so a retry is gl's replay."""
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.seam import derived_id
from brick_pos.domain.errors import RuleBroken
ZERO = Decimal("0.00")
EVENT = "day_close"
def source_document(session_id: str) -> str:
return f"pos:{session_id}"
def journal_id(session_id: str) -> str:
return str(derived_id(source_document(session_id), EVENT))
class _Lines:
def __init__(self, cost_centre: str):
self.cost_centre, self.rows = cost_centre, []
def add(self, purpose: str, amount: Decimal, *, side: str, tax_code: Optional[str] = None, cost_centre: bool = False,
memo: str = "") -> None:
"""`amount` is on `side` (debit | credit) when positive and on the other side when negative; zero is no line."""
if amount == 0:
return
if amount < 0:
amount, side = -amount, ("credit" if side == "debit" else "debit")
row = {"purpose": purpose, side: format(amount, "f")}
if tax_code:
row["tax_code"] = tax_code
if cost_centre and self.cost_centre:
row["cost_centre"] = self.cost_centre
if memo:
row["memo"] = memo
self.rows.append(row)
def session_journal(*, session_id: str, business_date: date, currency: str, company_code: str, cost_centre: str, cash: Decimal,
cash_tips: Decimal, card: Decimal, card_tips: Decimal, voucher_redeemed: Decimal, sales_by_code: list,
voucher_sold: Decimal, tips_treatment: str, difference: Decimal, moves: dict) -> dict:
"""The GL-30 body for source_module pos. `sales_by_code`: [(tax_code, gross, vat)] of the session's product lines (a refund
negative). `moves`: kind -> total of the session's cash moves. `difference`: counted less expected (negative: short)."""
L = _Lines(cost_centre)
g = lambda kind: moves.get(kind, ZERO) # noqa: E731
tips = cash_tips + card_tips
L.add("pos_cash_in_till", cash + cash_tips, side="debit", memo="cash taken")
L.add("card_clearing", card + card_tips, side="debit", memo="card takings with tips")
L.add("voucher_liability", voucher_redeemed, side="debit", memo="vouchers redeemed")
L.add("cash_over_short", -difference, side="debit", memo="drawer difference")
L.add("shop_supplies", g("paid_out"), side="debit", memo="paid out of the drawer")
L.add("pos_cash_in_safe", g("safe_drop"), side="debit", memo="safe drops")
net = sum((gross - vat for _c, gross, vat in sales_by_code), ZERO)
L.add("revenue", net + (tips if tips_treatment == "revenue" else ZERO), side="credit", cost_centre=True, memo="sales net of VAT")
for code, _gross, vat in sorted(sales_by_code):
L.add("tax", vat, side="credit", tax_code=code, memo=f"VAT {code}")
L.add("voucher_liability", voucher_sold, side="credit", memo="vouchers sold")
L.add("tips_payable", tips if tips_treatment == "liability" else ZERO, side="credit", memo="tips passed to staff")
L.add("pos_cash_in_till", g("paid_out"), side="credit", memo="paid out")
L.add("pos_cash_in_till", g("safe_drop"), side="credit", memo="safe drops")
L.add("pos_cash_in_till", difference, side="debit", memo="drawer difference")
L.add("shop_supplies", g("paid_in"), side="credit", memo="paid in")
L.add("pos_cash_in_till", g("paid_in"), side="debit", memo="paid in")
L.add("pos_cash_in_safe", g("float_top_up"), side="credit", memo="float top-up from the safe")
L.add("pos_cash_in_till", g("float_top_up"), side="debit", memo="float top-up")
debit = sum((Decimal(r["debit"]) for r in L.rows if "debit" in r), ZERO)
credit = sum((Decimal(r["credit"]) for r in L.rows if "credit" in r), ZERO)
if debit != credit:
raise RuleBroken(f"the journal of {session_id} does not balance: debit {debit}, credit {credit}", gate="balanced",
fix="check the session's takings, moves and count")
if not L.rows:
raise RuleBroken(f"session {session_id} took nothing: there is no journal to post", gate="nothing_to_post",
fix="nothing to do")
return {"journal_id": journal_id(session_id), "source_module": "pos", "source_document": source_document(session_id),
"event": EVENT, "company_code": company_code, "journal_date": business_date.isoformat(), "currency": currency,
"memo": f"POS {session_id} {business_date.isoformat()}", "lines": L.rows}