dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/gates.py
"""The readiness of closing a period: every gate as {gate, met, reason, fix}, computed at read time (shared.yaml
conventions.readiness). A gate that asks a neighbour that is not installed is simply not asked. The close refuses with the
unmet gates' own words; the Periods screen and the closing cockpit read the same list, so they cannot disagree."""
from datetime import date
from decimal import Decimal
from typing import Optional
from sqlalchemy import func, select, type_coerce
from brick_gl.application import carry
from brick_gl.domain import dates, schedule
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import Gate, met, unmet
from brick_gl.infrastructure import seams
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.tables import MONEY
def _ledgers(repo, company: dict, period: dict) -> list:
return repo.company_ledgers.assigned(company["company_code"], dates.fy_number(period["fiscal_year"]))
def prior_closed(repo, company, period, settings) -> Gate:
prev = repo.periods.get(company["company_code"], dates.prev_period(period["period_id"]))
if prev is not None and prev["status"] != "closed":
return unmet("prior_closed", f"the previous period {prev['period_id']} is still open", "close it first")
return met("prior_closed")
def no_proposed(repo, company, period, settings) -> Gate:
waiting = repo.journals.proposed_in(company["company_code"], period["period_id"])
if waiting:
return unmet("no_proposed", f"{len(waiting)} manual journal(s) dated in {period['period_id']} are waiting for approval "
f"({', '.join(str(j['journal_id']) for j in waiting[:5])})", "approve or reject them (GL-11)")
return met("no_proposed")
def pending_reversals_posted(repo, company, period, settings) -> Gate:
start = period["starts_on"]
late = [j for j in repo.journals.pending_reversals(company["company_code"])
if dates.first_of_next_month(j["journal_date"]) <= start]
if late:
return unmet("pending_reversals_posted", f"reversals due on {start} are not posted: {', '.join(str(j['journal_id']) for j in late[:5])}",
"a platform fault if the period is open: report it")
return met("pending_reversals_posted")
def carry_forward_current(repo, company, period, settings) -> Gate:
start, _ = carry.year_bounds(company, dates.fy_number(period["fiscal_year"]))
if period["starts_on"] != start:
return met("carry_forward_current")
ok, why = carry.is_current(repo, company, dates.fy_number(period["fiscal_year"]))
return met("carry_forward_current") if ok else unmet("carry_forward_current", why, "the system re-derives them (GL-23)")
def close_signed_off(repo, company, period, settings) -> Gate:
if not settings.get("close_cockpit"):
return met("close_signed_off")
from brick_gl.application import ledgerops
last = repo.periods.of_company(company["company_code"], period["fiscal_year"])[-1]["period_id"] == period["period_id"]
run = repo.conn.execute(select(t.gl_close_run).where(
t.gl_close_run.c.company_code == company["company_code"], t.gl_close_run.c.period_id == period["period_id"],
t.gl_close_run.c.kind == ("year" if last else "month"))).first()
row = dict(run._mapping) if run is not None else None
if row is None or row["status"] != "signed":
return unmet("close_signed_off", "the close run is not signed off by a second person",
"Close cockpit: finish the tasks and ask another accountant to sign (close_cockpit on)")
if row["tb_digest"] != ledgerops.tb_digest(repo, company, period):
return unmet("close_signed_off", "a posting landed in the period after the sign-off: the sign-off is stale",
"Close cockpit: ask another accountant to sign off again")
return met("close_signed_off")
def trial_balance_zero(repo, company, period, settings) -> Gate:
bad = []
for ledger_id in _ledgers(repo, company, period):
rows = repo.ledger.sums(company["company_code"], ledger_id, upto=period["ends_on"], group_by=("segment",))
total = sum((r["f_debit"] - r["f_credit"] for r in rows), ZERO)
if total != 0:
bad.append(f"{ledger_id} nets to {total}")
if settings.get("document_splitting") == "on":
for r in rows:
net = r["f_debit"] - r["f_credit"]
if net != 0 and (r["segment"] or ""):
bad.append(f"{ledger_id} segment {r['segment']} nets to {net}")
if bad:
return unmet("trial_balance_zero", "the trial balance does not net to zero: " + "; ".join(bad), "a platform fault: stop and report it")
return met("trial_balance_zero")
def recurring_posted(repo, company, period, settings) -> Gate:
pid = period["period_id"]
missing = []
for rj in repo.conn.execute(select(t.gl_recurring_journal).where(
t.gl_recurring_journal.c.company_code == company["company_code"], t.gl_recurring_journal.c.status == "approved")):
rj = dict(rj._mapping)
if not schedule.recurring_due(rj["frequency"], rj["first_period"], rj["last_period"], pid):
continue
done = repo.conn.execute(select(t.gl_recurring_posting.c.recurring_id).where(
t.gl_recurring_posting.c.recurring_id == rj["recurring_id"], t.gl_recurring_posting.c.version == rj["version"],
t.gl_recurring_posting.c.period_id == pid)).first()
if done is None:
missing.append(rj["recurring_id"])
if missing:
return unmet("recurring_posted", f"approved recurring entries due in {pid} are not posted: {', '.join(missing)}",
"Recurring: run them (GL-45)")
return met("recurring_posted")
def accruals_posted(repo, company, period, settings) -> Gate:
pid = period["period_id"]
missing = []
for a in repo.conn.execute(select(t.gl_accrual).where(t.gl_accrual.c.company_code == company["company_code"],
t.gl_accrual.c.status == "active")):
a = dict(a._mapping)
if not (a["start_period"] <= pid <= a["end_period"]):
continue
done = repo.conn.execute(select(t.gl_accrual_release.c.accrual_code).where(
t.gl_accrual_release.c.company_code == a["company_code"], t.gl_accrual_release.c.accrual_code == a["accrual_code"],
t.gl_accrual_release.c.period_id == pid)).first()
if done is None:
missing.append(a["accrual_code"])
if missing:
return unmet("accruals_posted", f"active accruals or deferrals have no release in {pid}: {', '.join(missing)}",
"Accruals: run them (GL-49)")
return met("accruals_posted")
def allocations_posted(repo, company, period, settings) -> Gate:
pid = period["period_id"]
missing = []
for c in repo.conn.execute(select(t.gl_allocation_cycle).where(
t.gl_allocation_cycle.c.company_code == company["company_code"], t.gl_allocation_cycle.c.status == "approved")):
c = dict(c._mapping)
if not (c["valid_from_period"] <= pid and (not c["valid_to_period"] or pid <= c["valid_to_period"])):
continue
done = repo.conn.execute(select(t.gl_allocation_run.c.run_id).where(
t.gl_allocation_run.c.cycle_id == c["cycle_id"], t.gl_allocation_run.c.cycle_version == c["version"],
t.gl_allocation_run.c.period_id == pid, t.gl_allocation_run.c.status == "posted")).first()
if done is None:
missing.append(c["cycle_id"])
if missing:
return unmet("allocations_posted", f"approved allocation cycles have no posted run for {pid}: {', '.join(missing)}",
"Allocations: post them (GL-46), or retire the cycle")
return met("allocations_posted")
def depreciation_posted(repo, company, period, settings) -> Gate:
if not seams.ASSETS.installed:
return met("depreciation_posted")
from brick_gl.infrastructure import seams as s
bad = []
for ledger_id in _ledgers(repo, company, period):
a = s.neighbours.depreciation_status(company["company_code"], period["period_id"], ledger_id)
if not a.ok:
bad.append(f"{ledger_id}: fixed-assets answered {a.status or 'nothing'}")
continue
for g in a.data.get("readiness", []):
if not g.get("met", True):
bad.append(f"{ledger_id}: {g.get('reason')}")
if any(not ar.get("posted") for ar in a.data.get("areas", []) if ar.get("ledger_id") == ledger_id):
bad.append(f"{ledger_id}: depreciation not posted")
if bad:
return unmet("depreciation_posted", "fixed-assets has not posted the period's depreciation: " + "; ".join(bad),
"run depreciation in fixed-assets (only when it is installed)")
return met("depreciation_posted")
def _recon(repo, company_code: str, kind: str, as_at: date) -> Optional[dict]:
r = repo.conn.execute(select(t.gl_reconciliation).where(
t.gl_reconciliation.c.company_code == company_code, t.gl_reconciliation.c.kind == kind,
t.gl_reconciliation.c.as_at == as_at))
rows = [dict(x._mapping) for x in r]
return rows[0] if rows else None
def _reconciled(kind: str, seam, gate: str, what: str):
def check(repo, company, period, settings) -> Gate:
if not seam.installed:
return met(gate)
rows = [dict(x._mapping) for x in repo.conn.execute(select(t.gl_reconciliation).where(
t.gl_reconciliation.c.company_code == company["company_code"], t.gl_reconciliation.c.kind == kind,
t.gl_reconciliation.c.as_at == period["ends_on"]))]
if not rows:
return unmet(gate, f"the {what} control account is not reconciled as at {period['ends_on']}",
"Reconcile: run the reconciliation, then work the breaks or accept the difference with a reason")
bad = [r for r in rows if (r["variance"] or ZERO) != 0 and not r["accepted_by"]]
if bad:
return unmet(gate, f"the {what} control account has breaks as at {period['ends_on']} (variance "
f"{bad[0]['variance']})", "Reconcile: work the breaks or accept the difference with a reason")
return met(gate)
return check
receivable_reconciled = _reconciled("receivable", seams.PAYMENTS, "receivable_reconciled", "receivable")
payable_reconciled = _reconciled("payable", seams.PURCHASING, "payable_reconciled", "payables")
def ic_matched(repo, company, period, settings) -> Gate:
mode = settings.get("ic_match_required")
if mode == "off":
return met("ic_matched")
code, end = company["company_code"], period["ends_on"]
rows = repo.ledger.sums(code, _ledgers(repo, company, period)[0], upto=end, group_by=("partner_company",),
extra=[t.gl_split_line.c.partner_company.isnot(None), t.gl_split_line.c.partner_company != ""])
problems = []
tol = Decimal(str(settings.get("ic_tolerance")))
for r in rows:
partner = r["partner_company"]
m = repo.conn.execute(select(t.gl_ic_match).where(t.gl_ic_match.c.match_id == f"{code}:{partner}:{end}")).first()
m = dict(m._mapping) if m is not None else None
if m is None:
problems.append(f"{partner}: not matched as at {end}")
continue
import json
breaks = json.loads(m["breaks_json"] or "[]")
open_ = [b for b in breaks if abs(Decimal(str(b["difference"]))) > tol and not b.get("accepted")]
if open_ and not m["accepted_by"]:
problems.append(f"{partner}: {len(open_)} break(s) above tolerance")
if problems:
text = "intercompany breaks above tolerance and not accepted: " + "; ".join(problems)
if mode == "warn":
return Gate("ic_matched", True, text, "Intercompany: fix or accept them")
return unmet("ic_matched", text, "Intercompany: fix or accept them (ic_match_required refuse; warn lists them)")
return met("ic_matched")
def revalued(repo, company, period, settings) -> Gate:
if not settings.get("revaluation"):
return met("revalued")
code, end = company["company_code"], period["ends_on"]
missing = []
for ledger_id in _ledgers(repo, company, period):
s = t.gl_split_line.c
foreign = repo.conn.execute(select(func.count()).select_from(t.gl_split_line).join(
t.gl_account, t.gl_account.c.account_id == s.account_id).where(
s.company_code == code, s.currency != company["functional_currency"], s.journal_date <= end,
t.gl_account.c.monetary.is_(True))).scalar()
if foreign:
run = repo.conn.execute(select(t.gl_fx_reval_run.c.run_id).where(
t.gl_fx_reval_run.c.run_id == f"{code}:{ledger_id}:{end}:balances")).first()
if run is None:
missing.append(f"{ledger_id} balances")
if missing:
return unmet("revalued", "foreign balances or open items are not revalued at the period end: " + ", ".join(missing),
"Currencies: revalue (or revaluation is off)")
return met("revalued")
def resplit_done(repo, company, period, settings) -> Gate:
if settings.get("document_splitting") != "on":
return met("resplit_done")
code, end = company["company_code"], period["ends_on"]
from erp_platform.seam import derived_id
needs = repo.conn.execute(select(func.count()).select_from(t.gl_journal_line).join(
t.gl_journal, t.gl_journal.c.journal_id == t.gl_journal_line.c.journal_id).join(
t.gl_account, t.gl_account.c.account_id == t.gl_journal_line.c.account_id).where(
t.gl_journal.c.company_code == code, t.gl_journal.c.period_id == period["period_id"],
t.gl_account.c.subtype.in_(("receivable", "payable")), t.gl_journal_line.c.against_document.is_(None),
t.gl_journal.c.source_module.in_(("payments", "pay")))).scalar()
if not needs:
return met("resplit_done")
missing = [ledger_id for ledger_id in _ledgers(repo, company, period)
if repo.journals.get(derived_id(code, ledger_id, str(end), "resplit")) is None]
if missing:
return unmet("resplit_done", f"open items are not re-split by profit centre as at {end} ({', '.join(missing)})",
"the system runs it as a close task (splitting on only)")
return met("resplit_done")
GATE_NAMES = ("prior_closed", "no_proposed", "receivable_reconciled", "payable_reconciled", "revalued", "recurring_posted", "accruals_posted",
"allocations_posted", "depreciation_posted", "ic_matched", "resplit_done", "pending_reversals_posted", "carry_forward_current",
"close_signed_off", "trial_balance_zero")
GATES = (prior_closed, no_proposed, receivable_reconciled, payable_reconciled, revalued, recurring_posted, accruals_posted,
allocations_posted, depreciation_posted, ic_matched, resplit_done, pending_reversals_posted, carry_forward_current,
close_signed_off, trial_balance_zero)
def named(repo, company: dict, period: dict, settings, name: str) -> Gate:
"""One gate by its name (a system task of the closing cockpit is met when its gate is)."""
return dict(zip(GATE_NAMES, GATES))[name](repo, company, period, settings)
def period_gates(repo, company: dict, period: dict, settings) -> list:
"""Every gate of close_period for a period, in the spec's order."""
return [g(repo, company, period, settings) for g in GATES]
__all__ = ["GATES", "period_gates", "type_coerce", "MONEY"]