dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/groupdata.py
"""What consolidation reads from the books (GL-38), through exact routes, so group figures come from the ledger and never from a
re-keyed trial balance: GET /group-reporting/balances (derived on read from split lines and the carry-forward through the
group account map in force on the period's last day), GET /intercompany/reconciliation (intercompany.py) and the readiness gate
group_mapped. gl never writes cons_ tables. Held by consolidation's service client only (gl:group_data:extract)."""
import hashlib
import json
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_gl.application import carry
from brick_gl.application.brick import (
COMPANY_READ,
GL,
GROUP_EXTRACT,
company_row,
repository,
today,
)
from brick_gl.application.inputs import CompanyCode, IsoDate, LedgerCode, PeriodId
from brick_gl.domain import dates
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import Gate, met, unmet
from brick_gl.domain.values import BALANCE_SHEET
@GL.use_case(kind="query", route="GET /group-reporting/balances", permission=GROUP_EXTRACT)
def get_group_balances(caller: dict, company_code: CompanyCode, ledger_id: LedgerCode, period_id: PeriodId) -> dict:
"""The company's year-to-date and period amounts in its functional currency per GROUP account, partner company, movement type and
segment, for a ledger and period: derived on read from the split lines and the carry-forward (balance-sheet accounts' opening is
movement type 'opening') through the group account map in force on the period's last day. Answers the period's status (open or
closed, so consolidation can refuse or warn), the ledger, the functional currency, the accounts with postings and no group
account (unmapped) and a digest of the rows: the same call twice answers the same digest while nothing posts. 404 when the
company is not visible; 422 ledger_assigned when the ledger is not kept by the company that year."""
with repository() as repo:
company = company_row(repo, caller, company_code)
code = company["company_code"]
p = repo.periods.get(code, period_id)
if p is None:
raise Refused(404, f"no period {period_id} of {code}")
year = dates.fy_number(p["fiscal_year"])
if ledger_id not in repo.company_ledgers.assigned(code, year):
raise Refused(422, f"[ledger_assigned] the ledger {ledger_id} is not assigned to {code} for {p['fiscal_year']} "
"(fix: Companies: assign the ledger)")
fy_start, _ = carry.year_bounds(company, year)
end = p["ends_on"]
grp = ("account_id", "partner_company", "movement_type", "segment")
ytd = repo.ledger.sums(code, ledger_id, since=fy_start, upto=end, group_by=grp)
per = repo.ledger.sums(code, ledger_id, since=p["starts_on"], upto=end, group_by=grp)
cache: dict = {}
def acct(aid):
if aid not in cache:
cache[aid] = repo.accounts.get(aid)
return cache[aid]
ytd_by: dict = {}
per_by: dict = {}
for r in ytd:
k = (r["account_id"], r["partner_company"] or "", r["movement_type"] or "", r["segment"] or "")
ytd_by[k] = ytd_by.get(k, ZERO) + r["f_debit"] - r["f_credit"]
for r in per:
k = (r["account_id"], r["partner_company"] or "", r["movement_type"] or "", r["segment"] or "")
per_by[k] = per_by.get(k, ZERO) + r["f_debit"] - r["f_credit"]
for (aid, pc, seg, partner), (f, _cur) in carry.opening_for_report(repo, company, ledger_id, year).items():
if acct(aid) and acct(aid)["type"] in BALANCE_SHEET and f != 0:
k = (aid, partner, "opening", seg)
ytd_by[k] = ytd_by.get(k, ZERO) + f
rows: dict = {}
unmapped: dict = {}
for k in set(ytd_by) | set(per_by):
aid, partner, mt, seg = k
gid = repo.group_map.of(aid, end)
if gid is None:
u = unmapped.setdefault(aid, ZERO)
unmapped[aid] = u + ytd_by.get(k, ZERO)
continue
gk = (gid, partner, mt, seg)
cur = rows.setdefault(gk, [ZERO, ZERO])
cur[0] += ytd_by.get(k, ZERO)
cur[1] += per_by.get(k, ZERO)
types: dict = {}
def kind(gid):
if gid not in types:
ga = repo.accounts.get(gid)
types[gid] = ga["type"] if ga else ""
return types[gid]
out_rows = [{"group_account_id": g, "account_type": kind(g), "sign_class": "balance_sheet" if kind(g) in BALANCE_SHEET else "income_statement",
"partner_company": pc, "movement_type": mt, "segment": seg, "ytd_amount": v[0], "period_amount": v[1]}
for (g, pc, mt, seg), v in sorted(rows.items()) if v[0] != 0 or v[1] != 0]
group_chart = ""
if out_rows:
ga = repo.accounts.get(out_rows[0]["group_account_id"])
group_chart = ga["chart_id"] if ga else ""
digest = hashlib.sha256(json.dumps([[str(r["group_account_id"]), r["partner_company"], r["movement_type"], r["segment"], str(r["ytd_amount"]),
str(r["period_amount"])] for r in out_rows]).encode()).hexdigest()
return {"company_code": code, "ledger_id": ledger_id, "period_id": period_id, "fiscal_year": p["fiscal_year"], "period_status": p["status"],
"functional_currency": company["functional_currency"], "group_chart_id": group_chart, "digest": digest, "rows": out_rows,
"unmapped": [{"account_id": aid, "code": acct(aid)["code"], "ytd_amount": v} for aid, v in sorted(unmapped.items()) if v != 0 or True]}
@GL.use_case(kind="query", route="GET /readiness/group-mapped", permission=COMPANY_READ)
def get_group_mapping_readiness(caller: dict, company_code: Optional[CompanyCode] = None, as_at: Optional[IsoDate] = None) -> dict:
"""The gate group_mapped: every account with a posting must report under a group account on the date (consolidation cannot
collect until it is met); the accounts that do not are listed."""
with repository() as repo:
company = company_row(repo, caller, company_code)
on = dates.parse(as_at) if as_at else today(repo)
code = company["company_code"]
from sqlalchemy import select
from brick_gl.infrastructure import tables as t
ids = [r[0] for r in repo.conn.execute(select(t.gl_split_line.c.account_id).where(t.gl_split_line.c.company_code == code).distinct())]
missing = []
for aid in ids:
a = repo.accounts.get(aid)
if a and repo.group_map.of(aid, on) is None:
missing.append({"account_id": aid, "code": a["code"], "name": a["name"]})
g: Gate = unmet("group_mapped", f"{len(missing)} account(s) with a posting report under no group account: "
+ ", ".join(m["code"] for m in sorted(missing, key=lambda m: m["code"])[:10]),
"Accounts: map each to a group account (map_group_account)") if missing else met("group_mapped")
return {"company_code": code, "as_at": on, "gate": g.as_dict(), "unmapped": sorted(missing, key=lambda m: m["code"])}
_ = Decimal