dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/domain/journal.py
"""A journal: one balanced entry of one company, built from what a caller sent (GL-02, GL-30, GL-22).
Everything a posting decides is decided HERE, with no database: the lookups it needs (an account, the account map, a cost
centre's profit centre, ...) come through a `Lookups` object the application provides. The rules that make a ledger:
* every journal balances to the penny in its currency AND in the functional currency (a conversion residual is one
line to the map's `rounding` purpose, never silently dropped);
* a line names an account OR a purpose, never both, and a purpose is resolved by the map row in force on the
journal's own date;
* debit and credit are bounded NUMERIC(18,2), never negative, never both on one line;
* the dimensions of a line (profit centre, segment, partner company) are DERIVED where nobody typed them, in a fixed
order: explicit, the cost centre's assignment, the account's default, the company's default.
"""
import hashlib
import json
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO, convert, q2
from brick_gl.domain.readiness import refusal
from brick_gl.domain.values import SOURCE_ONLY_SUBTYPES
MAP_FIX = "Account map: map it (GL-12)"
@dataclass(frozen=True)
class CompanyInfo:
company_code: str
functional_currency: str
chart_id: str
default_profit_centre: str = ""
bp_id: Optional[int] = None
@dataclass(frozen=True)
class AccountInfo:
account_id: int
chart_id: str
code: str
name: str
type: str
subtype: str
group_chart: bool = False
source_only: bool = False
monetary: bool = False
currency: Optional[str] = None
default_profit_centre: str = ""
fs_cost_centre: str = "optional"
fs_profit_centre: str = "optional"
fs_bp_id: str = "optional"
fs_tax_code: str = "optional"
active_until: Optional[date] = None
@dataclass
class LineIn:
"""One side of an entry as it was sent. A purpose may be qualified (tax_code, bank_account_id, a partner company, an
asset class); `qualifier` names it directly."""
account_id: Optional[int] = None
purpose: Optional[str] = None
qualifier: Optional[str] = None
tax_code: Optional[str] = None
bank_account_id: Optional[str] = None
asset_class: Optional[str] = None
debit: Decimal = ZERO
credit: Decimal = ZERO
memo: str = ""
bp_id: Optional[int] = None
cost_centre: Optional[str] = None
profit_centre: Optional[str] = None
partner_company: Optional[str] = None
movement_type: Optional[str] = None
against_document: Optional[str] = None
ic_reference: Optional[str] = None
asset_id: Optional[str] = None
# a migrated line keeps its own currency and the Odoo line (migration only)
currency: Optional[str] = None
amount_currency: Optional[Decimal] = None
odoo_line_id: Optional[str] = None
company_code: Optional[str] = None # a cross-company journal's line names its company
def purpose_qualifier(self) -> str:
if self.qualifier:
return self.qualifier
if self.purpose == "tax":
return self.tax_code or ""
if self.purpose == "bank":
return self.bank_account_id or ""
if self.purpose in ("ic_receivable", "ic_payable"):
return self.partner_company or ""
return self.asset_class or self.tax_code or ""
@dataclass
class Line:
"""A resolved line: an account, both amounts, the derived dimensions."""
line_no: int
account_id: int
purpose: str
debit: Decimal
credit: Decimal
f_debit: Decimal
f_credit: Decimal
memo: str = ""
bp_id: Optional[int] = None
tax_code: str = ""
cost_centre: str = ""
profit_centre: str = ""
segment: str = ""
partner_company: str = ""
movement_type: str = ""
against_document: str = ""
ic_reference: str = ""
asset_id: str = ""
currency: str = ""
amount_currency: Optional[Decimal] = None
odoo_line_id: str = ""
qualifier: str = ""
rounding: bool = False
@property
def f_net(self) -> Decimal:
return self.f_debit - self.f_credit
@dataclass
class Draft:
company_code: str
journal_date: date
currency: str
lines: list
ledger_id: Optional[str] = None
fx_rate: Optional[Decimal] = None
memo: str = ""
source_module: str = ""
source_document: str = ""
event: str = ""
reverse_next_period: bool = False
@dataclass
class Built:
lines: list
fx_rate: Decimal
total: Decimal # one side, in the journal's currency
f_total: Decimal # one side, in the functional currency
class Lookups:
"""What a posting needs to know. The application implements it over the repository; tests may fake it."""
company: CompanyInfo
def account(self, account_id: int) -> Optional[AccountInfo]: ...
def mapped(self, purpose: str, qualifier: str, on: date) -> Optional[int]: ...
def cost_centre(self, code: str, on: date) -> Optional[dict]: ... # {'status', 'company_code', 'profit_centre'}
def profit_centre_valid(self, code: str, on: date) -> bool: ...
def segment_of(self, profit_centre: str, on: date) -> str: ...
def company_of_partner(self, bp_id: int) -> Optional[str]: ...
def _dim(value) -> str:
return "" if value is None else str(value)
def build(d: Draft, ctx: Lookups, *, manual: bool = False, fx: Optional[Decimal] = None,
allow_zero_lines: bool = False) -> Built:
"""Resolve and check a draft: raises RuleBroken (with its gate) or returns the resolved lines."""
company = ctx.company
foreign = d.currency != company.functional_currency
if foreign:
if fx is None:
raise refusal("fx_rate", f"{d.currency} is not {company.functional_currency}, the functional currency of "
f"{company.company_code}, and no rate was given or is in force on {d.journal_date}",
"Currencies: enter the rate for the date", status=422)
rate = fx
else:
rate = Decimal(1)
if not d.lines:
raise refusal("balanced", "the journal has no lines", "the calling module's journal builder", status=422)
out: list = []
for i, li in enumerate(d.lines, 1):
debit, credit = li.debit or ZERO, li.credit or ZERO
if debit and credit:
raise refusal("bounds", f"line {i} has both a debit and a credit: one side only", "split it into two lines", status=422)
if not debit and not credit and not allow_zero_lines:
raise refusal("bounds", f"line {i} has neither a debit nor a credit", "leave a zero line out", status=422)
if (li.account_id is None) == (li.purpose is None):
raise refusal("bounds", f"line {i} names an account or a purpose, exactly one", "send account_id or purpose",
status=422)
purpose = li.purpose or ""
account_id = li.account_id
if account_id is None:
q = li.purpose_qualifier()
account_id = ctx.mapped(purpose, q, d.journal_date)
if account_id is None and q:
account_id = ctx.mapped(purpose, "", d.journal_date)
if account_id is None:
what = f"{purpose} ({q})" if q else purpose
raise refusal("purposes_mapped", f"the purpose {what} has no account in force on {d.journal_date} in the "
f"chart {company.chart_id}", MAP_FIX, status=422)
acc = ctx.account(account_id)
if acc is None:
raise refusal("account_known", f"line {i}: there is no account {account_id}", "Accounts: add it, or send another",
status=422)
if acc.chart_id != company.chart_id and not (acc.group_chart):
raise refusal("account_known", f"line {i}: account {acc.code} is in chart {acc.chart_id}, not in "
f"{company.chart_id}, the chart of {company.company_code}",
"use an account of the company's chart", status=422)
if acc.group_chart:
raise refusal("group_chart_not_postable", f"line {i}: {acc.code} is a group-chart account and cannot carry a line",
"use the operational account mapped to it", status=422)
if acc.active_until is not None and acc.active_until <= d.journal_date:
raise refusal("account_active", f"line {i}: account {acc.code} was deactivated on {acc.active_until}",
"Accounts: use an active account", status=422)
if manual and (acc.source_only or acc.subtype in SOURCE_ONLY_SUBTYPES):
raise refusal("source_only_account", f"line {i}: {acc.code} {acc.name} is kept by its module "
f"({acc.subtype}) and takes no manual line",
"post it in that module, or use a non-control account", status=422)
_field_status(i, acc, li)
cc = _dim(li.cost_centre)
cc_row = None
if cc:
cc_row = ctx.cost_centre(cc, d.journal_date)
if cc_row is None:
raise refusal("cost_centre_known", f"line {i}: there is no cost centre {cc}", "Dimensions: add it", status=422)
if cc_row["status"] == "closed":
raise refusal("cost_centre_open", f"line {i}: cost centre {cc} is closed", "use an open cost centre", status=422)
pc = _dim(li.profit_centre) or (cc_row or {}).get("profit_centre") or acc.default_profit_centre \
or company.default_profit_centre or ""
if pc and not ctx.profit_centre_valid(pc, d.journal_date):
raise refusal("profit_centre_valid", f"line {i}: profit centre {pc} is not valid for {company.company_code} on "
f"{d.journal_date}", "Dimensions: extend it, or send another", status=422)
partner = _dim(li.partner_company)
if not partner and li.bp_id is not None:
other = ctx.company_of_partner(li.bp_id)
if other and other != company.company_code:
partner = other
f_debit = convert(debit, rate) if foreign else debit
f_credit = convert(credit, rate) if foreign else credit
out.append(Line(
line_no=i, account_id=account_id, purpose=purpose, debit=debit, credit=credit, f_debit=f_debit, f_credit=f_credit,
memo=li.memo or "", bp_id=li.bp_id, tax_code=_dim(li.tax_code), cost_centre=cc, profit_centre=pc,
segment=ctx.segment_of(pc, d.journal_date) if pc else "", partner_company=partner,
movement_type=_dim(li.movement_type), against_document=_dim(li.against_document),
ic_reference=_dim(li.ic_reference), asset_id=_dim(li.asset_id), currency=_dim(li.currency),
amount_currency=li.amount_currency, odoo_line_id=_dim(li.odoo_line_id),
qualifier=li.purpose_qualifier() if li.purpose else ""))
total_d = sum((ln.debit for ln in out), ZERO)
total_c = sum((ln.credit for ln in out), ZERO)
if total_d != total_c:
raise refusal("balanced", f"debits and credits differ by {abs(total_d - total_c)} {d.currency} "
f"(debit {total_d}, credit {total_c})", "the calling module's journal builder", status=422)
f_d = sum((ln.f_debit for ln in out), ZERO)
f_c = sum((ln.f_credit for ln in out), ZERO)
if f_d != f_c:
residual = f_d - f_c
if abs(residual) > Decimal("0.01") * len(out):
raise refusal("balanced", f"debits and credits differ by {abs(residual)} {company.functional_currency} "
"in the functional currency", "the calling module's journal builder", status=422)
rid = ctx.mapped("rounding", "", d.journal_date)
if rid is None:
raise refusal("purposes_mapped", "the functional conversion leaves a rounding difference of "
f"{abs(residual)} and the purpose rounding has no account in force on "
f"{d.journal_date}", MAP_FIX, status=422)
pc = company.default_profit_centre
out.append(Line(
line_no=len(out) + 1, account_id=rid, purpose="rounding", debit=ZERO, credit=ZERO,
f_debit=abs(residual) if residual < 0 else ZERO, f_credit=residual if residual > 0 else ZERO,
memo="rounding", profit_centre=pc, segment=ctx.segment_of(pc, d.journal_date) if pc else "", rounding=True))
f_d = sum((ln.f_debit for ln in out), ZERO)
return Built(lines=out, fx_rate=rate, total=total_d, f_total=f_d)
def _field_status(i: int, acc: AccountInfo, li: LineIn) -> None:
for name, status, given in (("cost centre", acc.fs_cost_centre, li.cost_centre), ("profit centre", acc.fs_profit_centre, li.profit_centre),
("partner", acc.fs_bp_id, li.bp_id), ("tax code", acc.fs_tax_code, li.tax_code)):
present = given not in (None, "")
if status == "required" and not present and name != "profit centre":
raise refusal("field_status", f"line {i}: account {acc.code} requires a {name} on every line",
"the calling module's builder, or the account's field status", status=422)
if status == "forbidden" and present:
raise refusal("field_status", f"line {i}: account {acc.code} takes no {name}",
"the calling module's builder, or the account's field status", status=422)
def mirror(lines: list) -> list:
"""The reversal of resolved lines: debit and credit swapped, every dimension kept (neither original is deleted)."""
out = []
for ln in lines:
m = Line(**{**ln.__dict__})
m.debit, m.credit, m.f_debit, m.f_credit = ln.credit, ln.debit, ln.f_credit, ln.f_debit
if ln.amount_currency is not None:
m.amount_currency = -ln.amount_currency
out.append(m)
return out
# ── the digest of a request: a replay with the same body answers the journal, a different body is a conflict ──────────────
def _norm(v):
if isinstance(v, Decimal):
return format(v.normalize(), "f") if v != 0 else "0"
if isinstance(v, date):
return v.isoformat()
return v
def body_digest(d: Draft) -> str:
lines = []
for li in d.lines:
lines.append({k: _norm(v) for k, v in sorted(li.__dict__.items()) if v not in (None, "", ZERO) or k in ("debit", "credit")})
body = {"company": d.company_code, "ledger": d.ledger_id or "", "date": d.journal_date.isoformat(), "currency": d.currency,
"fx_rate": _norm(d.fx_rate) if d.fx_rate is not None else "", "memo": d.memo, "source": d.source_module,
"document": d.source_document, "event": d.event, "lines": lines}
if d.reverse_next_period: # counted only when true: the digest of a journal without it is what it always was
body["reverse_next_period"] = True
return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
@dataclass
class Totals:
debit: Decimal = ZERO
credit: Decimal = ZERO
f_debit: Decimal = ZERO
f_credit: Decimal = ZERO
count: int = 0
rows: list = field(default_factory=list)
def book_for(source_module: str, manual: bool = False, adjustment: bool = False) -> str:
"""The journal book a posting belongs to: numbers run per company, book and year."""
if adjustment:
return "adjustment"
if manual:
return "misc"
return {"ar": "sales", "purchasing": "purchases", "payments": "cash", "pay": "cash", "gl": "misc", "odoo": "migration"
}.get(source_module, "misc")
__all__ = ["AccountInfo", "Built", "CompanyInfo", "Draft", "Line", "LineIn", "Lookups", "RuleBroken", "build", "q2"]