dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

src/brick_consolidation/application/migration.py

"""CON-18: what this brick holds for each migration reconcile check (module.yaml migration.odoo.reconcile), to set beside Odoo's side.

The loader (CON-34) writes the carried history read-only: the closed periods as signed group periods, their balances as cons_statement rows of the
report `migration` (node = group account), their journals as cons_journal rows of document type migration, the structure per period. Here those are
read back, in a number of statements that does not grow with the companies, and compared by the aggregate.
"""
import json
from decimal import Decimal

from sqlalchemy import select

from brick_consolidation.domain import consolidation as C
from brick_consolidation.infrastructure.tables import cons_journal, cons_journal_line, cons_statement


def ours(repo, run) -> dict:
    """{check: our side} for each check of the run. Journal totals are the sum of the debit lines (what Odoo's SUM(amount) of the positive side is)."""
    doors = json.loads(run.doors_json)
    group_id, periods = doors["group_id"], list(doors["carried_periods"])
    found = repo.groups.get(group_id)
    names = {c["check"] for c in run.checks()}
    out: dict = {}
    if "consolidated_balance" in names:
        rows = repo.conn.execute(select(cons_statement).where(cons_statement.c.group_id == group_id, cons_statement.c.report == "migration",
                                                                cons_statement.c.period_id.in_(periods))).mappings().all()
        out["consolidated_balance"] = {p: {r["node_id"]: str(r["amount"]) for r in rows if r["period_id"] == p} for p in periods}
    if "structure" in names:
        out["structure"] = {p: {c: str(m.ownership) for c, m in sorted(C.members_at(found[0], p).items())} for p in periods} if found else {}
    if "journals" in names:
        j, ln = cons_journal, cons_journal_line
        heads = repo.conn.execute(select(j.c.cons_journal_id, j.c.period_id).where(j.c.group_id == group_id, j.c.document_type == "migration",
                                                                                   j.c.period_id.in_(periods))).all()
        lines = repo.conn.execute(select(ln.c.cons_journal_id, ln.c.amount).where(ln.c.cons_journal_id.in_([h[0] for h in heads] or [""]))).all()
        by_period = {h[0]: h[1] for h in heads}
        debit: dict = {}
        for jid, amount in lines:
            if amount > 0:
                debit[by_period[jid]] = debit.get(by_period[jid], Decimal(0)) + amount
        out["journals"] = {p: {"count": sum(1 for h in heads if h[1] == p), "total": str(C.q2(debit.get(p, Decimal(0))))} for p in periods}
    if "proportional" in names:
        out["proportional"] = {c: d for c, d in sorted((doors.get("decisions") or {}).items())}
    return out