dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/loader.py
"""CON-34: carry Odoo 16 Enterprise `account_consolidation` into this brick, read-only, through the CON-18 migration-run doors.
What is written (module.yaml migration.odoo.maps), all keyed on the Odoo record so a second run changes nothing:
consolidation.chart -> cons_group (group_id named at the door; currency; the leading ledger of the top company from gl)
consolidation.account -> cons_account_rule [group, group account, first month] (end/avg/hist -> closing/average/historical)
consolidation.company_period -> cons_group_structure [group, company, valid_from_period], a row only where the ownership changes;
then, from the cut-over, the door's decision for every company below 100
consolidation.period (closed) -> cons_group_period, status signed, signed_by `migration`, prepared_by the run, tb_digest of the carried balances
consolidation.journal(.line) -> cons_journal (document_type migration), cons_journal_line; cons_journal_id = derived_id(odoo:<db>:consolidation.journal:<id>)
the consolidated balance -> cons_statement (report `migration`, node = group account): the SUM of the carried lines, never re-translated
every Odoo record -> cons_migration_map [odoo_db, odoo_model, odoo_id] -> our key
Preconditions are gl's: companies and accounts mapped by gl's migration, read with ONE GET /migration-map/{db}/{model}?ids= per model (and
MAP_CHUNK ids), never one call per id. Journals load in CHUNK-sized keyset chunks, each its own transaction: a load that stops resumes where it
stopped, because a journal already written is skipped by its key. A period carried by ANOTHER run is never read or rewritten.
"""
import hashlib
import json
from datetime import date
from decimal import Decimal
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import RuleBroken
from brick_consolidation.domain.group import AccountRule, Group, GroupPeriod, GroupStructure
from brick_consolidation.infrastructure.seams import Missing, OdooUnavailable, ports
CHUNK = 200
RATE_TYPE = {"end": "closing", "avg": "average", "hist": "historical"}
CHART_MODEL, ACCOUNT_MODEL, PERIOD_MODEL, JOURNAL_MODEL, COMPANY_MODEL, ODOO_ACCOUNT = (
"consolidation.chart", "consolidation.account", "consolidation.period", "consolidation.journal", "res.company", "account.account")
def wired() -> bool:
return bool(getattr(ports.odoo, "wired", False))
def db_of(run_id: str) -> str:
"""run_id is odoo:<db>:<cutover_date>."""
parts = run_id.split(":")
if len(parts) < 3 or parts[0] != "odoo":
raise RuleBroken(f"the run id {run_id} is not odoo:<db>:<cutover_date>")
return ":".join(parts[1:-1])
def cutover_of(run_id: str) -> str:
try:
d = date.fromisoformat(run_id.split(":")[-1])
except ValueError:
raise RuleBroken(f"the run id {run_id} does not end in the cut-over date (YYYY-MM-DD)") from None
return f"{d.year:04d}-{d.month:02d}"
def month_after(period_id: str) -> str:
y, m = int(period_id[:4]), int(period_id[5:7])
return f"{y + (m == 12):04d}-{1 if m == 12 else m + 1:02d}"
def month_of(d) -> str:
d = d if isinstance(d, date) else date.fromisoformat(str(d))
return f"{d.year:04d}-{d.month:02d}"
def journal_key(db: str, odoo_id) -> str:
return str(derived_id(f"odoo:{db}:consolidation.journal:{odoo_id}"))
class Plan:
"""Everything the load needs, read and checked BEFORE the first write: one read of Odoo's structure, one map lookup per model."""
def __init__(self):
self.problems: list = []
def _chunks(items: list, n: int):
for i in range(0, len(items), n):
yield items[i:i + n]
def preflight(run_id: str, doors: dict) -> Plan:
db, odoo, gl = db_of(run_id), ports.odoo, ports.gl
chart_id = doors.get("odoo_chart_id")
if not chart_id:
raise RuleBroken("the doors do not name the Odoo consolidation chart: add odoo_chart_id (which consolidation chart is the group) to doors_json")
plan = Plan()
plan.db, plan.run_id, plan.group_id, plan.carried = db, run_id, doors["group_id"], sorted(doors["carried_periods"])
try: # v3 (gl GL-34): the loader proceeds only when gl's migration of THIS database is signed
status = gl.migration_status(db)
except Missing as m:
raise RuleBroken(f"gl cannot answer the migration status of {db}: {m.reason}") from None
if status.get("status") != "signed":
raise Refused(409, f"gl's migration of {db} is {status.get('status') or 'none'} (none | loaded | reconciled), not signed", gate="gl_signed",
fix=f"sign gl's reconciliation of {db} first (gl GL-34)")
if status.get("cutover_date"):
cut = str(status["cutover_date"])[:7]
late = [p for p in plan.carried if p > cut]
if late:
raise RuleBroken(f"{', '.join(late)} falls after gl's cutover ({status['cutover_date']}): nothing after it is loaded from Odoo")
try:
chart = odoo.chart(chart_id)
if not chart:
raise RuleBroken(f"Odoo has no consolidation chart {chart_id} (is account_consolidation installed? Odoo 17 and later: start the group at the cut-over)")
accounts = odoo.accounts(chart_id)
periods = [p for p in odoo.periods(chart_id) if p.get("state") == "closed"]
by_period = {month_of(p["date_analysis_end"]): p for p in periods}
missing = [p for p in plan.carried if p not in by_period]
if missing:
raise RuleBroken(f"Odoo has no closed consolidation period for {', '.join(missing)}: carry only closed periods")
plan.periods = {p: by_period[p] for p in plan.carried}
cps = odoo.company_periods([p["id"] for p in plan.periods.values()])
except OdooUnavailable as e:
raise RuleBroken(f"Odoo cannot be read: {e}") from None
plan.chart, plan.accounts, plan.company_periods = chart, accounts, cps
# gl's map: one lookup per model (companies, accounts), then the group chart's accounts by code
company_ids = sorted({str(i) for i in chart.get("company_ids", [])} | {str(cp["company_id"]) for cp in cps})
account_ids = sorted({str(i) for a in accounts for i in a.get("account_ids", [])})
try:
plan.companies = gl.migration_map(db, COMPANY_MODEL, company_ids)
plan.account_map = gl.migration_map(db, ODOO_ACCOUNT, account_ids)
known = {c["company_code"]: c for c in gl.companies()}
except Missing as m:
raise RuleBroken(f"gl cannot answer the migration map: {m.reason}") from None
gaps = []
if [i for i in company_ids if i not in plan.companies]:
gaps.append(f"companies {', '.join(i for i in company_ids if i not in plan.companies)}")
if [i for i in account_ids if i not in plan.account_map]:
gaps.append(f"accounts {', '.join(i for i in account_ids if i not in plan.account_map)}")
if gaps:
raise RuleBroken(f"gl's migration is not signed for this group: not mapped in gl: {'; '.join(gaps)} — load, reconcile and sign gl's Odoo "
"migration first (GET /migration-map answers only what it carried)")
group_chart = doors.get("group_chart_id", "GROUP")
try:
by_code = {a["code"]: str(a["account_id"]) for a in gl.chart_accounts(group_chart)}
except Missing as m:
raise RuleBroken(f"gl has no group chart {group_chart}: {m.reason}") from None
plan.group_chart, plan.group_account = group_chart, {}
absent = []
for a in accounts:
if a["code"] in by_code:
plan.group_account[a["id"]] = by_code[a["code"]]
else:
absent.append(a["code"])
if absent:
raise RuleBroken(f"gl's group chart {group_chart} has no account {', '.join(absent)}: gl's migration carries the consolidation accounts")
top = doors.get("top_company_code") or None
if top is None:
raise RuleBroken("the doors do not name the top company: add top_company_code to doors_json")
if top not in known:
raise RuleBroken(f"top_company_code {top} is not a company in gl")
plan.top = top
plan.ledger = doors.get("ledger_id") or (known[top].get("ledgers") or ["IFRS"])[0]
plan.fye = str(known[top].get("financial_year_end_month") or 12)
plan.decisions = doors.get("decisions") or {}
return plan
def _group(repo, plan) -> tuple:
found = repo.groups.get(plan.group_id)
if found is None:
return Group(group_id=plan.group_id, name=plan.chart.get("name", plan.group_id), group_currency=plan.chart["currency"],
ledger_id=plan.ledger, group_chart_id=plan.group_chart, top_company_code=plan.top, fiscal_year_end_month=plan.fye), 0
group, version = found
if group.group_currency != plan.chart["currency"]:
raise RuleBroken(f"group {plan.group_id} is in {group.group_currency} here and the Odoo chart is in {plan.chart['currency']}")
return group, version
def _structure_and_rules(group, plan, run_id: str) -> bool:
"""Add what is missing, by key; True when something was added."""
changed = False
first_month = month_of(plan.periods[plan.carried[0]]["date_analysis_begin"])
have = {(r.group_account_id, r.valid_from_period) for r in group.account_rules}
for a in sorted(plan.accounts, key=lambda a: a["id"]):
key = (plan.group_account[a["id"]], first_month)
if key not in have:
group.account_rules.append(AccountRule(group_id=group.group_id, group_account_id=key[0], valid_from_period=first_month,
rate_type=RATE_TYPE.get(a.get("currency_mode"), "closing")))
have.add(key)
changed = True
rows = {(s.company_code, s.valid_from_period) for s in group.group_structures}
last: dict = {}
by_period: dict = {}
for cp in plan.company_periods:
by_period.setdefault(cp["period_id"], []).append(cp)
for period_id in plan.carried:
odoo_period = plan.periods[period_id]
start = month_of(odoo_period["date_analysis_begin"])
for cp in sorted(by_period.get(odoo_period["id"], []), key=lambda c: str(c["company_id"])):
code = plan.companies[str(cp["company_id"])]
rate = Decimal(str(cp["rate_consolidation"]))
if code == plan.top or last.get(code) == rate or (code, start) in rows:
last[code] = rate
continue
group.group_structures.append(GroupStructure(group_id=group.group_id, company_code=code, valid_from_period=start, investor_company_code=plan.top,
ownership_pct=rate, voting_pct=rate, method="full", change_id=run_id))
rows.add((code, start))
last[code] = rate
changed = True
cut = max(cutover_of(run_id), month_after(plan.carried[-1]))
for code, rate in sorted(last.items()):
decision = plan.decisions.get(code)
if rate < 100 and decision in ("full", "equity", "none") and (code, cut) not in rows:
group.group_structures.append(GroupStructure(group_id=group.group_id, company_code=code, valid_from_period=cut, investor_company_code=plan.top,
ownership_pct=rate, voting_pct=rate, method=decision, change_id=run_id))
rows.add((code, cut))
changed = True
return changed
def _owned_elsewhere(group, run_id: str) -> set:
"""Periods a DIFFERENT migration run carried and signed: not read, not rewritten."""
return {p.period_id for p in group.group_periods if p.status == "signed" and p.signed_by == "migration" and p.prepared_by not in ("", run_id)}
def odoo_checks(plan: Plan, doors: dict) -> list:
"""Odoo's side of each reconcile check, from ODOO's own aggregates (read_group), in this brick's keys."""
agg = ports.odoo.aggregates(doors["odoo_chart_id"], [v["id"] for v in plan.periods.values()])
by_odoo = {v["id"]: k for k, v in plan.periods.items()}
def per(name):
return {by_odoo.get(p, p): v for p, v in (agg.get(name) or {}).items() if by_odoo.get(p, p) in plan.carried}
balance: dict = {}
for p, accounts in per("balance").items():
for odoo_account, amount in accounts.items():
ga = plan.group_account[odoo_account]
balance.setdefault(p, {})[ga] = str(C.q2(Decimal(balance.get(p, {}).get(ga, 0)) + Decimal(str(amount))))
structure = {p: {plan.companies[str(c)]: format(Decimal(str(r)).normalize(), "f") for c, r in sorted(rows.items(), key=lambda kv: str(kv[0]))}
for p, rows in per("structure").items()}
journals = {p: {"count": int(v["count"]), "total": str(C.q2(Decimal(str(v["total"]))))} for p, v in per("journals").items()}
last = structure.get(plan.carried[-1], {})
return [{"check": "consolidated_balance", "odoo": balance}, {"check": "structure", "odoo": structure}, {"check": "journals", "odoo": journals},
{"check": "proportional", "odoo": {c: r for c, r in last.items() if Decimal(r) < 100}}]
def load(repository, run_id: str, doors: dict, plan: "Plan" = None, *, refresh: bool = False) -> dict:
"""Preflight, then group/rules/structure, then journals in chunks, then statements and the signed group periods. Returns what it wrote.
refresh (a reload after a fix in Odoo, the run not signed): a journal already written whose Odoo amounts changed is superseded — its next
version carries Odoo's lines now — so the reconcile compares Odoo's corrected figures with ours, never the old ones on both sides."""
plan = plan or preflight(run_id, doors)
out = {"group": 0, "journals": 0, "skipped_journals": 0, "superseded_journals": 0, "statements": 0, "periods": 0, "maps": 0, "chunks": 0, "skipped_periods": []}
with repository() as repo:
group, version = _group(repo, plan)
elsewhere = _owned_elsewhere(group, run_id)
plan.carried = [p for p in plan.carried if p not in elsewhere]
out["skipped_periods"] = sorted(elsewhere & set(doors["carried_periods"]))
if not plan.carried:
return out
plan.periods = {p: plan.periods[p] for p in plan.carried}
odoo_ids = {v["id"]: k for k, v in plan.periods.items()}
period_of_cp = {cp["id"]: cp for cp in plan.company_periods}
with repository() as repo: # 1. the group, its rules, its structure, the records' map
group, version = _group(repo, plan)
added = _structure_and_rules(group, plan, run_id)
if added or version == 0:
repo.groups.save(group, version)
out["group"] = 1
out["maps"] += _write_maps(repo, plan, [(CHART_MODEL, doors["odoo_chart_id"], plan.group_id)]
+ [(ACCOUNT_MODEL, a["id"], plan.group_account[a["id"]]) for a in plan.accounts]
+ [(PERIOD_MODEL, v["id"], k) for k, v in plan.periods.items()])
after = 0 # 2. journals, chunk by chunk, each its own transaction
while True:
rows = ports.odoo.journals(list(odoo_ids), after, CHUNK)
if not rows:
break
after = max(r["id"] for r in rows)
with repository() as repo:
keys = {r["id"]: journal_key(plan.db, r["id"]) for r in rows}
have = repo.migration.existing_journals(list(keys.values()))
fresh = [r for r in rows if keys[r["id"]] not in have]
out["skipped_journals"] += len(rows) - len(fresh)
lines: dict = {}
if fresh or refresh:
for ln in sorted(ports.odoo.journal_lines([r["id"] for r in (rows if refresh else fresh)]), key=lambda x: x["id"]):
lines.setdefault(ln["journal_id"], []).append(ln)
if refresh and len(fresh) < len(rows):
held = repo.migration.journal_versions([keys[r["id"]] for r in rows if keys[r["id"]] in have])
changed = []
for r in rows:
if keys[r["id"]] not in held:
continue
version, old = held[keys[r["id"]]]
cp = period_of_cp.get(r.get("company_period_id"))
company = plan.companies[str(cp["company_id"])] if cp else plan.top
now_lines = [{"group_account_id": plan.group_account[ln["account_id"]], "company_code": company, "partner_company": "",
"movement_type": "", "segment": "", "amount": C.q2(Decimal(str(ln["amount"])))} for ln in lines.get(r["id"], [])]
padded = now_lines + [{**o, "amount": C.ZERO} for o in old[len(now_lines):]]
if [(x["group_account_id"], x["company_code"], C.q2(x["amount"])) for x in padded] != \
[(o["group_account_id"], o["company_code"], C.q2(o["amount"])) for o in old]:
changed.append({"cons_journal_id": keys[r["id"]], "version": version, "group_id": plan.group_id,
"period_id": odoo_ids[r["period_id"]], "lines": now_lines, "held_lines": len(old), "old": old})
repo.migration.supersede_journals(changed)
out["superseded_journals"] += len(changed)
if fresh:
entries = []
for r in fresh:
cp = period_of_cp.get(r.get("company_period_id"))
company = plan.companies[str(cp["company_id"])] if cp else plan.top
entries.append({"cons_journal_id": keys[r["id"]], "group_id": plan.group_id, "period_id": odoo_ids[r["period_id"]],
"document_type": "migration", "task": "migration", "run_no": 1, "entry_no": 0, "reverses_id": "",
"period_specific": "", "memo": str(r.get("name") or "")[:200], "migration_run_id": run_id,
"lines": [{"group_account_id": plan.group_account[ln["account_id"]], "company_code": company, "partner_company": "",
"movement_type": "", "segment": "", "amount": C.q2(Decimal(str(ln["amount"])))}
for ln in lines.get(r["id"], [])]})
repo.data.post_entries(entries)
out["journals"] += len(entries)
out["maps"] += _write_maps(repo, plan, [(JOURNAL_MODEL, r["id"], keys[r["id"]]) for r in fresh])
out["chunks"] += 1
with repository() as repo: # 3. the carried balances and the signed group periods
balances = repo.migration.balances(plan.group_id, plan.carried)
existing = {(r["period_id"], r["node_id"]): r for p in plan.carried for r in repo.data.statements(plan.group_id, p, ["migration"])}
now = _now()
fresh_rows, digests = [], {}
for p in plan.carried:
mine = {a: v for (pp, a), v in balances.items() if pp == p}
digests[p] = hashlib.sha256(json.dumps({a: str(v) for a, v in sorted(mine.items())}, sort_keys=True).encode()).hexdigest()
for a, v in sorted(mine.items()):
old = existing.get((p, a))
if old is None or Decimal(str(old["amount"])) != v or old["digest"] != digests[p]:
fresh_rows.append({"report": "migration", "group_id": plan.group_id, "period_id": p, "node_id": a, "amount": v, "prior_year": "",
"produced_at": now, "digest": digests[p]})
if fresh_rows:
repo.data.write_statements(fresh_rows)
out["statements"] = len(fresh_rows)
group, version = _group(repo, plan)
have = {gp.period_id: gp for gp in group.group_periods}
touched = False
for p in plan.carried:
gp = have.get(p)
if gp is None:
group.group_periods.append(GroupPeriod(group_id=plan.group_id, period_id=p, run_no=1, prepared_by=run_id, signed_by="migration",
signed_sub="migration", signed_at=now, tb_digest=digests[p], status="signed"))
touched = True
elif gp.status == "signed" and gp.prepared_by == run_id and gp.tb_digest != digests[p]:
gp.tb_digest, touched = digests[p], True
elif gp.prepared_by != run_id:
raise RuleBroken(f"{p} was closed here by a real close, not carried: the group's own period is never replaced by history")
if touched:
repo.groups.save(group, version)
out["periods"] = len(plan.carried)
return out
def _now():
from erp_platform import work
return work.now()
def _write_maps(repo, plan, entries: list) -> int:
"""Write the missing/different map rows of one model batch; one read and at most one write per model."""
written = 0
for model in sorted({e[0] for e in entries}):
want = {str(i): str(k) for m, i, k in entries if m == model}
have = repo.migration.maps(plan.db, model, list(want))
rows = [{"odoo_db": plan.db, "odoo_model": model, "odoo_id": i, "natural_key": k, "run_id": plan.run_id}
for i, k in want.items() if have.get(i) != k]
repo.migration.write_maps(rows)
written += len(rows)
return written