dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/domain/cons_migration_run.py
"""The ConsMigrationRun aggregate — the rules, and nothing else.
the cons migration run (no entity in the spec: TODO(rule) describe it)
No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored.
Generated by `erp-compose scaffold` from module.yaml; the TODO lines are the work left.
"""
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal, InvalidOperation
from typing import Optional
from brick_consolidation.domain.errors import RuleBroken
from brick_consolidation.domain.events import (
ConsMigrationRunDiscarded,
ConsMigrationRunLoaded,
ConsMigrationRunReconciled,
ConsMigrationRunSigned,
ConsMigrationRunUpdated,
ConsMigrationRunUpdatedFromLoaded,
)
from brick_consolidation.domain.values import ConsMigrationRunState
# migration.odoo.reconcile (module.yaml migration.odoo.reconcile): each check and what it must be. "equal" is never explained away; "explained" may
# differ when a person gives the reason.
CHECK_MUST = {"consolidated_balance": "equal", "structure": "equal", "journals": "equal", "proportional": "explained"}
def _plain(value):
"""A comparable form: numbers as their value (80 = 80.000000 = "80.00"), everything else as is."""
if isinstance(value, dict):
return {str(k): _plain(v) for k, v in sorted(value.items(), key=lambda kv: str(kv[0]))}
if isinstance(value, (list, tuple)):
return [_plain(v) for v in value]
if isinstance(value, (int, str, Decimal)) and not isinstance(value, bool):
try:
return str(Decimal(str(value)).normalize())
except InvalidOperation:
return value
return value
def same(odoo, ours) -> bool:
return _plain(odoo) == _plain(ours)
def parse_checks(text: str) -> list:
if not text:
return []
try:
checks = json.loads(text)
except ValueError:
raise RuleBroken("checks_json is not JSON") from None
if not isinstance(checks, list) or not all(isinstance(c, dict) and c.get("check") in CHECK_MUST and "odoo" in c for c in checks):
raise RuleBroken("checks_json is a list of {check, odoo}, the check being one of " + ", ".join(CHECK_MUST))
return checks
def parse_doors(text: str) -> dict:
try:
doors = json.loads(text) if text else None
except ValueError:
doors = None
if not isinstance(doors, dict) or not doors.get("group_id") or not doors.get("carried_periods"):
raise RuleBroken("the doors are not decided: doors_json names the group (group_id) and the closed periods to carry (carried_periods), and for each "
"company below 100% how it is consolidated from the cut-over (decisions)")
return doors
def checksum(check: dict) -> str:
return hashlib.sha256(json.dumps({"check": check["check"], "odoo": _plain(check["odoo"]), "ours": _plain(check.get("ours"))},
sort_keys=True, separators=(",", ":")).encode()).hexdigest()
@dataclass
class ConsMigrationRun:
"""the cons migration run (no entity in the spec: TODO(rule) describe it)"""
cons_migration_run_id: str
status: str = ConsMigrationRunState.LOADED.value
doors_json: str = ""
checks_json: str = ""
signed_by: str = ""
signed_sub: str = ""
signed_at: Optional[datetime] = None
checksums: str = ""
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
# ── the lifecycle (module.yaml workflows) ──────────────────────────────────────────────────────────────────
@classmethod
def load_cons_from_odoo(cls, cons_migration_run_id: str, *, by: str = "", doors_json: str = "", checks_json: str = "", **attrs) -> "ConsMigrationRun":
"""Load Odoo's consolidation: none → loaded. Gate: the doors decided; gl's migration signed (companies and accounts mapped). Effect: the
doors and Odoo's side of each check kept; the group, structure, rules and journals are written read-only by the loader (CON-34). Returns the
new aggregate."""
parse_doors(doors_json)
checks = [{"check": c["check"], "odoo": c["odoo"], "must": CHECK_MUST[c["check"]]} for c in parse_checks(checks_json)]
it = cls(cons_migration_run_id=cons_migration_run_id, doors_json=doors_json, checks_json=json.dumps(checks, sort_keys=True), **attrs)
it.status = ConsMigrationRunState.LOADED.value
it._record(ConsMigrationRunLoaded(cons_migration_run_id=it.cons_migration_run_id, by=by))
return it
@staticmethod
def parse_doors_of(text: str) -> dict:
return parse_doors(text)
def checks(self) -> list:
return parse_checks(self.checks_json)
def reconcile_cons_migration(self, ours: Optional[dict] = None, *, by: str = "") -> Optional[ConsMigrationRunReconciled]:
"""Reconcile against Odoo: loaded → reconciled. Effect: every check computed on both sides: ours is what this brick holds for it, and the
check is equal or differs."""
it = self
if it.status == ConsMigrationRunState.RECONCILED:
return None # already reconciled: a repeat changes nothing
if it.status not in (ConsMigrationRunState.LOADED,):
raise RuleBroken(f"a {it.status} cons migration run cannot move to reconciled")
checks = it.checks()
for c in checks:
c["ours"] = (ours or {}).get(c["check"])
c["status"] = "equal" if same(c["odoo"], c["ours"]) else "differs"
c["must"] = CHECK_MUST[c["check"]]
it.checks_json = json.dumps(checks, sort_keys=True)
it.status = ConsMigrationRunState.RECONCILED.value
return self._record(ConsMigrationRunReconciled(cons_migration_run_id=it.cons_migration_run_id, by=by))
def explain_cons_difference(self, reason: str, *, by: str = "", check: str = "", at: Optional[datetime] = None) -> Optional[ConsMigrationRunUpdated]:
"""Explain each difference: reconciled → reconciled. Gate: the check differs and may be explained (a check that must be equal never is)."""
it = self
if it.status not in (ConsMigrationRunState.RECONCILED,):
raise RuleBroken(f"a {it.status} cons migration run cannot move to reconciled")
if not (reason or "").strip():
raise RuleBroken("reason is required")
checks = it.checks()
target = [c for c in checks if c["status"] == "differs" and (not check or c["check"] == check)]
if check and not any(c["check"] == check for c in checks):
raise RuleBroken(f"the run has no check {check}")
if not target:
raise RuleBroken(f"{check or 'no check'} differs: there is nothing to explain" if check else "no check differs: there is nothing to explain")
refused = [c["check"] for c in target if c["must"] != "explained"]
if refused:
raise RuleBroken(f"{', '.join(refused)} must be equal to Odoo's and cannot be explained: fix the load and reload")
for c in target:
c["explanation"] = {"reason": reason, "by": by, "at": (at or datetime.min).isoformat()}
it.checks_json = json.dumps(checks, sort_keys=True)
return self._record(ConsMigrationRunUpdated(cons_migration_run_id=it.cons_migration_run_id, by=by))
def reload_cons_migration(self, *, by: str = "", odoo: Optional[list] = None) -> Optional[ConsMigrationRunUpdatedFromLoaded]:
"""Load again after a fix: loaded, reconciled → loaded. Gate: not signed. Odoo's side is read again when it is given (`odoo`, the
checks as Odoo reports them now: a fix in Odoo moves its side too); what was computed and explained is dropped."""
it = self
if it.status not in (ConsMigrationRunState.LOADED, ConsMigrationRunState.RECONCILED,):
raise RuleBroken(f"a {it.status} cons migration run cannot move to loaded")
now = {c["check"]: c["odoo"] for c in (odoo or [])}
it.checks_json = json.dumps([{"check": c["check"], "odoo": now.get(c["check"], c["odoo"]), "must": c["must"]} for c in it.checks()],
sort_keys=True)
it.status = ConsMigrationRunState.LOADED.value
return self._record(ConsMigrationRunUpdatedFromLoaded(cons_migration_run_id=it.cons_migration_run_id, by=by))
def sign_cons_reconciliation(self, *, by: str = "", sub: str = "", at: Optional[datetime] = None) -> Optional[ConsMigrationRunSigned]:
"""Sign the reconciliation: reconciled → signed. Gate: every check equal, or explained where explaining is allowed. Effect: who, when and
the checksums of the run recorded."""
it = self
if it.status == ConsMigrationRunState.SIGNED:
return None # already signed: a repeat changes nothing
if it.status not in (ConsMigrationRunState.RECONCILED,):
raise RuleBroken(f"a {it.status} cons migration run cannot move to signed")
checks = it.checks()
if not checks:
raise RuleBroken("nothing was compared: the run has no checks to sign")
open_ = [c["check"] for c in checks if c["status"] != "equal" and not (c["must"] == "explained" and c.get("explanation"))]
if open_:
raise RuleBroken("not every check is equal or explained: " + ", ".join(open_) + " (a check that must be equal is fixed and reloaded; one that "
"may differ is explained with a reason)")
sums = {c["check"]: checksum(c) for c in checks}
sums["run"] = hashlib.sha256(json.dumps(sums, sort_keys=True).encode()).hexdigest()
it.checksums, it.signed_by, it.signed_sub, it.signed_at = json.dumps(sums, sort_keys=True), by, sub, at
it.status = ConsMigrationRunState.SIGNED.value
return self._record(ConsMigrationRunSigned(cons_migration_run_id=it.cons_migration_run_id, by=by))
def discard_cons_migration(self, reason: str, *, by: str = "") -> Optional[ConsMigrationRunDiscarded]:
"""Discard the run: loaded, reconciled → discarded."""
it = self
if it.status == ConsMigrationRunState.DISCARDED:
return None # already discarded: a repeat changes nothing
if it.status not in (ConsMigrationRunState.LOADED, ConsMigrationRunState.RECONCILED,):
raise RuleBroken(f"a {it.status} cons migration run cannot move to discarded")
if not reason:
raise RuleBroken("reason is required")
it.status = ConsMigrationRunState.DISCARDED.value
return self._record(ConsMigrationRunDiscarded(cons_migration_run_id=it.cons_migration_run_id, by=by))