dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/close.py
"""The group close after the eliminations — the consolidated books, uploads for companies outside gl (CON-12), readiness (CON-13), sign-off
and reopen (CON-14).
Every gate that refuses answers `Refused(409, reason, gate=, fix=)`. Local rules are checked, and the group's version claimed (the use
case's save of the Group aggregate: the ONE key a run, a manual entry, a sign-off and a reopen of the same group collide on) before any
write; gl is only ever read.
"""
import csv
import hashlib
import io
import json
from decimal import Decimal
from erp_platform import work
from erp_platform.after import after_response
from erp_platform.bricks import Refused
from brick_consolidation.application import ownership, reports
from brick_consolidation.application.runs import _not_signed, gl_call, role_account, settings, stale_later_months, the_period
from brick_consolidation.application.structure import next_period
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import Invalid, RuleBroken
from brick_consolidation.domain.events import GroupPeriodCollectedFromSigned, GroupPeriodSigned
from brick_consolidation.infrastructure.seams import Missing, ports
def _who(email: str, sub: str) -> str:
return email or sub or ""
# ═══ CON-12 uploads ═════════════════════════════════════════════════════════════════════════════════════════════════
UPLOAD_COLUMNS = ("group_account_id", "partner_company", "movement_type", "segment", "ytd_amount")
def upload(repo, group, period_id: str, *, company: str, currency: str, content: str, by: str, sub: str) -> dict:
gp = the_period(group, period_id)
_not_signed(gp)
members = C.members_at(group, period_id)
if company not in members:
raise RuleBroken(f"{company} is not in the group's scope in {period_id}")
if not (len(currency or "") == 3 and currency.isalpha() and currency.isupper()):
raise Invalid("currency must be an ISO 4217 code (USD)")
kept = {c["company_code"] for c in gl_call(ports.gl.companies)}
if company in kept:
raise Refused(409, f"{company} is kept in gl: its figures come from gl, not from an upload", gate="not_kept_in_gl",
fix="collect it from gl (collect-group-data)")
chart = {str(a["account_id"]) for a in gl_call(ports.gl.chart_accounts, group.group_chart_id)} | \
{str(a["code"]) for a in gl_call(ports.gl.chart_accounts, group.group_chart_id) if a.get("code")}
reader = csv.DictReader(io.StringIO(content or ""))
if list(reader.fieldnames or []) != list(UPLOAD_COLUMNS):
raise Invalid("the file's first line must be: " + ",".join(UPLOAD_COLUMNS))
rows, seen, problems = [], set(), []
for n, raw in enumerate(reader, 2):
key = (raw["group_account_id"].strip(), raw["partner_company"].strip(), raw["movement_type"].strip(), raw["segment"].strip())
try:
value = C.amount(raw["ytd_amount"].strip(), f"line {n} amount")
except Invalid as bad:
problems.append(str(bad))
continue
if key[0] not in chart:
problems.append(f"line {n}: {key[0]} is not an account of the group chart {group.group_chart_id}")
if key[1] and key[1] not in members:
problems.append(f"line {n}: the partner {key[1]} is not a company of the group")
if key[2] and key[2] not in C.MOVEMENTS:
problems.append(f"line {n}: movement type {key[2]} is not one of {', '.join(C.MOVEMENTS)}")
if key in seen:
problems.append(f"line {n}: {key} appears twice")
seen.add(key)
rows.append({"group_account_id": key[0], "partner_company": key[1], "movement_type": key[2], "segment": key[3], "ytd_local": value})
if not rows:
problems.append("the file has no rows")
total = sum((r["ytd_local"] for r in rows), Decimal(0))
if total:
problems.append(f"the figures do not net to zero (year to date {total})")
if problems:
raise Refused(409, "the upload does not validate: " + "; ".join(problems[:10]), gate="upload_validates", fix="correct the file and upload it again")
held = repo.data.uploads(group.group_id, period_id, company)
number = max((u["upload_no"] for u in held), default=0) + 1 # erp-lint: allow L6 claimed by a plain INSERT on the key (group, period, company, upload_no)
sha = hashlib.sha256(content.encode()).hexdigest()
key = f"raw/consolidation/{group.group_id}/{period_id}/{company}-{number}-{sha[:12]}.csv"
repo.data.set_upload_status(group.group_id, period_id, company, [u["upload_no"] for u in held if u["status"] == "submitted"], "replaced")
repo.data.add_upload({"group_id": group.group_id, "period_id": period_id, "company_code": company, "upload_no": number, "object_key": key,
"sha256": sha, "rows": json.dumps([{**r, "ytd_local": str(r["ytd_local"])} for r in rows], sort_keys=True),
"status": "submitted", "submitted_by": by, "submitted_sub": sub, "accepted_by": "", "accepted_sub": ""})
after_response(ports.files.put, key, content.encode(), sha) # the file is kept once the upload is committed; the rows are in the table
return {"period_id": period_id, "company_code": company, "upload_no": number, "object_key": key, "sha256": sha, "rows": len(rows),
"currency": currency}
def accept(repo, group, period_id: str, *, company: str, currency: str, by: str, sub: str):
gp = the_period(group, period_id)
_not_signed(gp)
held = repo.data.uploads(group.group_id, period_id, company)
waiting = [u for u in held if u["status"] == "submitted"]
if not waiting:
if any(u["status"] == "accepted" for u in held):
return None # a repeat changes nothing
raise RuleBroken(f"no upload of {company} for {period_id} is waiting for acceptance")
up = waiting[-1]
if settings(repo)["signoff_second_person"] and not C.second_person(by, sub, up["submitted_by"], up["submitted_sub"]):
raise Refused(409, "the uploader cannot accept their own upload (matched on email and sub)", gate="second_person",
fix="another group controller accepts it")
rows = [{**r, "ytd_local": Decimal(r["ytd_local"])} for r in json.loads(up["rows"])]
base = {"group_id": group.group_id, "period_id": period_id, "company_code": company, "local_currency": currency or "", "source": "upload"}
from brick_consolidation.infrastructure.data import REPORTED_KEY, zero_fill
months = C.fiscal_months(period_id, group.fiscal_year_end_month)
prior = months[-2] if len(months) > 1 else None # the year's first month: its movement is the year to date
before, previous = set(), {}
for r in repo.data.reported(group.group_id, [period_id] + ([prior] if prior else []), [company]):
k = tuple(r[k] for k in REPORTED_KEY[2:])
if r["period_id"] == period_id:
before.add(k)
else:
previous[k] = r["ytd_local"]
# an upload carries year to date only: the month's movement is it less the previous month's year to date (CON-04 translates month by month)
fresh = [{**base, **r} for r in rows]
fresh = [{**r, "period_local": r["ytd_local"] - previous.get(tuple(r[k] for k in REPORTED_KEY[2:]), Decimal(0))} for r in fresh]
gone = [{**z, "period_local": -previous.get(tuple(z[k] for k in REPORTED_KEY[2:]), Decimal(0))}
for z in zero_fill(before, {tuple(r[k] for k in REPORTED_KEY[2:]) for r in fresh}, base)]
repo.data.write_reported(fresh + gone)
repo.data.set_upload_status(group.group_id, period_id, company, [u["upload_no"] for u in held if u["status"] == "accepted"], "replaced")
repo.data.set_upload_status(group.group_id, period_id, company, [up["upload_no"]], "accepted", accepted_by=by, accepted_sub=sub)
if gp.status == "open":
run_no = max(gp.run_no, 1)
else: # a replaced upload: later steps are stale, the next runs take run_no + 1
run_no, gp.run_no, gp.status, gp.stale_from_step = gp.run_no + 1, gp.run_no + 1, "collected", "translate"
stale_later_months(group, period_id)
repo.data.write_collections([{"group_id": group.group_id, "period_id": period_id, "company_code": company, "run_no": run_no,
"gl_digest": up["sha256"], "period_status": "closed", "local_currency": currency or "", "provisional": "false",
"collected_at": work.now(), "source": "upload"}])
return {"period_id": period_id, "company_code": company, "upload_no": up["upload_no"], "run_no": run_no, "rows": len(rows)}
# ═══ CON-13 readiness ═══════════════════════════════════════════════════════════════════════════════════════════════
def readiness(repo, group, period_id: str) -> dict:
gp = the_period(group, period_id)
members = C.members_at(group, period_id)
companies = sorted(members)
cfg_rules = group.account_rules
gates = []
def gate(name: str, problems: list, fix: str):
gates.append({"gate": name, "met": not problems, "reason": "; ".join(problems) if problems else "ok", "fix": "" if not problems else fix})
held = {}
for r in repo.data.collections(group.group_id, period_id):
if r["company_code"] not in held or r["run_no"] > held[r["company_code"]]["run_no"]:
held[r["company_code"]] = r
problems = [f"{c} is not collected" for c in companies if c not in held]
for c in companies:
if c in held and held[c]["source"] != "upload":
try:
live = ports.gl.balances(c, group.ledger_id, period_id).get("digest", "")
except Missing as e: # gl being away is a gate that is not met, never a crash
problems.append(f"{c}: gl could not be asked ({str(e)[:120]})")
continue
if live != held[c]["gl_digest"]:
problems.append(f"{c}'s gl figures changed since they were collected (digest {held[c]['gl_digest'][:8]} -> {live[:8]})"
+ (" — the signed figures are unchanged" if gp.status == "signed" else ""))
gate("collected_current", problems, "collect the period again (recollect-group-data); a signed period is reopened first (reopen-group-period)")
from brick_consolidation.application import carry
gate("carry_forward_current", carry.current_problems(repo, group, gp), "open the period again to derive the carry-forward from the previous year's signed close")
gate("account_roles_complete", carry.roles_problems(group, period_id), "Settings: account role retained_earnings")
from brick_consolidation.application.runs import deferred_tax_rate
gate("tax_rate_known", [f"an intercompany inventory entry of {i['holder_company']} has no tax rate, and {why}"
for i in repo.data.inventory(group.group_id, period_id)
if i["none_stated"] != "true" and i["inventory_amount"]
and (rw := deferred_tax_rate(i["holder_company"], i["tax_rate_pct"], C.period_end(period_id)))[0] is None
for why in [rw[1]]],
"enter the rate on the entry, or set statutory_tax_rate in gl (GL settings)")
gate("no_step_stale", ([f"the period is {gp.status}: translation and eliminations must be run"] if gp.status not in ("eliminated", "signed") else [])
+ ([f"{gp.stale_from_step} is stale"] if gp.stale_from_step else []), "run translate-currencies and run-eliminations again")
tb, entries = reports.consolidated(repo, group, gp, members)
reported: dict = {}
for r in repo.data.reported(group.group_id, [period_id], companies):
reported[r["company_code"]] = reported.get(r["company_code"], Decimal(0)) + r["ytd_local"]
gate("reported_nets_zero", [f"{c} nets to {v}" for c, v in sorted(reported.items()) if v], "correct the company's books in gl, or its upload, and collect again")
translated: dict = {}
for r in repo.data.translated(group.group_id, period_id, gp.run_no):
if members.get(r["company_code"]) and members[r["company_code"]].method != "equity":
translated[r["company_code"]] = translated.get(r["company_code"], Decimal(0)) + r["ytd_group"]
for e in entries:
if e["document_type"] == "translation":
for ln in e["lines"]:
translated[ln["company_code"]] = translated.get(ln["company_code"], Decimal(0)) + ln["amount"]
gate("translated_nets_zero", [f"{c} nets to {v} after its translation entry" for c, v in sorted(translated.items()) if v],
"translate again (translate-currencies)")
def bal(company, accounts, partner):
return sum((v for (c, a, p, _m, _s), v in tb.items() if c == company and p == partner and a in accounts), Decimal(0))
full = [c for c in companies if members[c].method == "full"]
# a company sold in this period: its balance sheet leaves the group at the period's end, what it owed the group is external now (as runs.eliminate)
gone = {ch["company_code"] for ch in repo.data.effective_changes(group.group_id, ("divestiture",))
if C.period_of(ownership.day_of(ch["effective_date"])) == period_id}
ic = []
for rule in group.elimination_rules:
if rule.kind not in ("ic_balance", "ic_pl"):
continue
left, right, tol = C.accounts(rule.left_accounts), C.accounts(rule.right_accounts), Decimal(rule.tolerance or 0)
for x in full:
for y in full:
if x != y and not (rule.kind == "ic_balance" and (x in gone or y in gone)):
rest = bal(x, left, y) + bal(y, right, x)
if abs(rest) > tol:
ic.append(f"{x}/{y} still differ by {rest} on {rule.rule_id} (tolerance {tol})")
gate("intercompany_within_tolerance", ic, "run the eliminations again after fixing the pair in gl (GL-52)")
inv_problems = []
inv = role_account(group, "investment", period_id, need=False)
for c in full:
m = members[c]
if m.investor and inv:
left = bal(m.investor, [inv], c)
if left:
inv_problems.append(f"{m.investor}'s investment in {c} on {inv} still shows {left}")
gate("investments_eliminated", inv_problems, "run the eliminations (run-eliminations)")
waiting = repo.data.waiting(group.group_id, period_id)
gate("no_manual_entry_waiting", [f"entry {w} waits for approval" for w in waiting], "approve or reject it (approve-cons-entry, reject-cons-entry)")
total = sum(tb.values(), Decimal(0))
gate("consolidated_tb_nets_zero", [f"the consolidated trial balance nets to {total}"] if total else [], "find the entry or company that does not net to zero")
pl = {r.group_account_id for r in cfg_rules if r.rate_type == "average"}
result = sum((v for (_c, a, _p, _m, _s), v in tb.items() if a in pl), Decimal(0))
sheet = sum((v for (_c, a, _p, _m, _s), v in tb.items() if a not in pl), Decimal(0))
gate("balance_sheet_balances", [f"the balance sheet is out by {sheet + result} (balance sheet {sheet}, result {result})"] if sheet + result else [],
"find the entry or company that does not net to zero")
return {"group_id": group.group_id, "period_id": period_id, "status": gp.status, "met": all(g["met"] for g in gates), "gates": gates,
"tb_digest": C.tb_digest(tb)}
# ═══ CON-14 sign-off and reopen ═════════════════════════════════════════════════════════════════════════════════════
def sign_off(repo, group, period_id: str, *, by: str, sub: str):
gp = the_period(group, period_id)
if gp.status == "signed":
return None # a repeat changes nothing
if gp.status != "eliminated":
raise Refused(409, f"a {gp.status} group period cannot be signed off", gate="eliminated", fix="collect, translate and eliminate it first")
second = settings(repo)["signoff_second_person"]
me = {x for x in (by, sub) if x}
if second and me & gp.preparer_identities():
raise Refused(409, "a preparer cannot sign off the period they prepared — you ran a step of it (matched on email and sub)", gate="second_person",
fix="another group controller signs it")
ready = readiness(repo, group, period_id)
unmet = [g for g in ready["gates"] if not g["met"]]
if unmet:
raise Refused(409, "readiness is not met: " + " | ".join(f"{g['gate']}: {g['reason']}" for g in unmet), gate=unmet[0]["gate"], fix=unmet[0]["fix"])
tb, entries = reports.consolidated(repo, group, gp)
if second and any(me & {e.get("proposed_by"), e.get("proposed_sub")} for e in entries if e.get("status") == "posted"):
raise Refused(409, "a preparer cannot sign off the period they prepared — you proposed an entry of it (matched on email and sub)",
gate="second_person", fix="another group controller signs it")
gp.signed_at = work.now()
materialised = reports.materialise(repo, group, gp, tb) # gl's versions are read here; nothing is written before this has answered
gp.status, gp.signed_by, gp.signed_sub, gp.tb_digest = "signed", by, sub, ready["tb_digest"]
repo.data.add_signoff(group.group_id, period_id, repo.data.signoffs(group.group_id, period_id), action="signed", by_email=by, by_sub=sub,
tb_digest=gp.tb_digest, reason="", run_no=gp.run_no)
group._record(GroupPeriodSigned(group_id=group.group_id, period_id=period_id, by=_who(by, sub)))
carried = _carry_on(repo, group, period_id)
return {"period_id": period_id, "run_no": gp.run_no, "tb_digest": gp.tb_digest, "signed_by": by, "statements": materialised,
**({"carry_forward": carried} if carried else {})}
def _carry_on(repo, group, period_id: str):
"""CON-15, again: a year's last period signed (again) while the next year already has its first period: derive the carry-forward from it, and
the periods of the next year whose figures stood on the old one are stale."""
from brick_consolidation.application import carry
nxt = next_period(period_id)
if carry.first_period_of(group, nxt) != nxt or not any(x.period_id == nxt for x in group.group_periods):
return None
out = carry.derive(repo, group, nxt)
if out["changed"]:
for x in group.group_periods:
if x.period_id >= nxt and x.status in ("translated", "eliminated"):
x.status, x.run_no, x.stale_from_step = "collected", x.run_no + 1, "translate"
return out
def reopen(repo, group, period_id: str, *, reason: str, by: str, sub: str):
gp = the_period(group, period_id)
if gp.status == "collected" and gp.reopened_reason:
return None # a repeat changes nothing
if not (reason or "").strip():
raise Invalid("a reason is required to reopen a signed period")
if gp.status != "signed":
raise Refused(409, f"a {gp.status} group period is not signed", gate="signed", fix="only a signed period is reopened")
if not C.second_person(by, sub, gp.signed_by, gp.signed_sub):
raise Refused(409, "the person who signed the period cannot reopen it (matched on email and sub)", gate="second_person",
fix="another group controller reopens it")
later = sorted(x.period_id for x in group.group_periods if x.period_id > period_id and x.status == "signed")
if later:
raise Refused(409, f"{later[0]} is signed and follows this period; reopen it first", gate="next_period_signed", fix=f"reopen {later[-1]} first")
repo.data.add_signoff(group.group_id, period_id, repo.data.signoffs(group.group_id, period_id), action="reopened", by_email=by, by_sub=sub,
tb_digest=gp.tb_digest, reason=reason, run_no=gp.run_no)
gp.status, gp.reopened_reason, gp.run_no, gp.stale_from_step = "collected", reason, gp.run_no + 1, "translate"
group._record(GroupPeriodCollectedFromSigned(group_id=group.group_id, period_id=period_id, by=_who(by, sub)))
return {"period_id": period_id, "run_no": gp.run_no, "reason": reason, "provisional": True}