dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/migration.py
"""The Odoo slice (GL-34, GL-35): carry Odoo's chart, journals, posted entries, fiscal years, lock dates and rates, so the books continue
without re-keying and with history comparable; reconcile the trial balance per account and period to the penny; the owner signs.
load_gl_from_odoo (machine gl:migration:load) writes read-only history keyed on the Odoo record; a second run changes nothing
reconcile_gl_migration (machine gl:migration:reconcile) every check computed on BOTH sides, {check, odoo, ours, equal, explanation}
explain_gl_difference an accountant explains a check that may be explained (the trial balance per account and period cannot be)
reload_gl_migration load again after a fix: the same keys; explanations on unchanged checks kept; refused once signed
sign_gl_reconciliation (gl:migration:sign, the owner) every check equal, or explained where allowed; who, when and checksums recorded
discard_gl_migration the owner discards an unsigned run (its rows are removed; nothing was posted on them)
History is READ-ONLY: a migrated journal equals Odoo's move to the penny per line, is never recomputed, re-posted or rewritten, and
nothing of a SIGNED run is rewritten by a later one - a later run only adds what Odoo posted after the signed cut-over. Odoo is read
through one port (infrastructure/odoo.py); nothing is ever written to it. The id map (gl_migration_map) is read by later bricks through
GET /migration-map/{db}/{model}?ids=.
"""
import hashlib
import json
from datetime import date, timedelta
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from pydantic import Field
from sqlalchemy import and_, func, select
from brick_gl.application import carry, ledgerops
from brick_gl.application.brick import (
GL,
MIGRATION_LOAD,
MIGRATION_READ,
MIGRATION_RECONCILE,
MIGRATION_SIGN,
SETTINGS_WRITE,
decide,
real_now,
repository,
sub_of,
who,
)
from brick_gl.application.inputs import Code, IsoDate, LedgerCode, Reason, _In
from brick_gl.domain import dates, hashchain, odoo_map
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO, largest_remainder, q2
from brick_gl.domain.readiness import Gate, met, refusal, require, unmet
from brick_gl.domain.values import BALANCE_SHEET
from brick_gl.infrastructure import odoo, seams
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
CHART = "MAIN"
RUN = lambda repo: _Keyed(repo.conn, t.gl_migration_run, ("run_id",)) # noqa: E731
MAP = lambda repo: _Keyed(repo.conn, t.gl_migration_map, ("odoo_db", "odoo_model", "odoo_id")) # noqa: E731
class DoorCompany(_In):
odoo_id: int
company_code: Annotated[str, Field(pattern=r"^[A-Z0-9]{1,4}$")]
country: Optional[Annotated[str, Field(max_length=2)]] = None
class DoorsModel(_In):
"""The migration's doors: the decisions the customer makes before anything is written."""
db: Annotated[str, Field(max_length=100)]
cutover_date: IsoDate
companies: Annotated[list[DoorCompany], Field(max_length=50)]
ledger_id: LedgerCode
accounting_principle: Code = "other"
history: Code = "two_years" # two_years | all | opening_only
profit_centre_plan: Optional[Code] = None
cost_centre_plan: Optional[Code] = None
segments: dict = {} # profit centre code -> segment; "*" = one segment for all
document_splitting: Code = "off"
recurring: Code = "proposals" # proposals | skip
lock_months: Code = "close" # close | open (months closed by period_lock_date only)
analytic_distribution: Code = "profit_centre" # profit_centre | drop
numbering: Code = "continue" # continue | new
group_chart_id: Code = "GROUP"
def _doors_gate(d: Optional[dict]) -> Gate:
need = ("db", "cutover_date", "companies", "ledger_id")
missing = [k for k in need if not (d or {}).get(k)]
return unmet("doors_decided", f"a migration door is undecided: {', '.join(missing)}", "Migration: answer it") if missing else met("doors_decided")
def _history_start(repo, doors: dict, company: dict, cutover: date) -> Optional[date]:
h = doors.get("history", "two_years")
if h == "all":
return None
if h == "opening_only":
return cutover + timedelta(days=1)
endm = int(company["financial_year_end_month"])
year = dates.fiscal_year_of(cutover, endm)
return dates.fy_bounds(year - 1, endm)[0]
def _run_id(doors: dict) -> str:
return f"odoo:{doors['db']}:{doors['cutover_date']}"
def _chunks(seq: list, n: int):
for i in range(0, len(seq), n):
yield seq[i:i + n]
def _map_put(repo, db: str, model: str, odoo_id, ours, run_id: str) -> bool:
"""True when new. A plain INSERT of the key: the map IS the record that a thing was loaded."""
k = str(odoo_id)
if MAP(repo).get(db, model, k) is not None:
return False
MAP(repo).add({"odoo_db": db, "odoo_model": model, "odoo_id": k, "natural_key": str(ours), "run_id": run_id})
return True
def _map_get(repo, db: str, model: str, odoo_id) -> Optional[str]:
r = MAP(repo).get(db, model, str(odoo_id))
return r["natural_key"] if r else None
class Counts(dict):
def bump(self, model: str, what: str, n: int = 1) -> None:
d = self.setdefault(model, {"created": 0, "skipped": 0, "updated": 0})
d[what] = d.get(what, 0) + n
# ═══ the load ═══════════════════════════════════════════════════════════════════════════════════════════════════════════
def _prior_signed_cutover(repo, db: str, below: date) -> Optional[date]:
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_migration_run.select().where(
t.gl_migration_run.c.odoo_db == db, t.gl_migration_run.c.status == "signed", t.gl_migration_run.c.cutover_date < below))]
return max((r["cutover_date"] for r in rows), default=None)
def _gather_mappings(reader, doors: dict, company_ids: list, start_by_company: dict, cutover: date, floor: Optional[date]) -> tuple:
"""Pass 1: the partners on receivable/payable lines and the taxes on tax lines (to check the gates before anything is written)."""
partners, taxes, per_company_accounts = set(), set(), {}
for cid in company_ids:
accounts = {a["id"]: a for a in reader.accounts(cid)}
per_company_accounts[cid] = accounts
lo = start_by_company[cid]
if floor and (lo is None or lo <= floor):
lo = floor + timedelta(days=1)
moves = reader.moves(cid, lo, cutover)
for chunk in _chunks([m["id"] for m in moves], 200):
for ln in reader.lines(chunk):
if ln.get("display_type") in ("line_section", "line_note"):
continue
acc = accounts.get(ln["account_id"])
if ln.get("partner_id") and acc and odoo_map.account_type(acc)[1] in ("receivable", "payable"):
partners.add(ln["partner_id"])
if ln.get("tax_line_id"):
taxes.add(ln["tax_line_id"])
return partners, taxes, per_company_accounts
def _resolve_map(answer, ids: set) -> dict:
rows = answer.data.get("body", []) if answer.ok else []
return {int(r["odoo_id"]): r["ours"] for r in rows if int(r["odoo_id"]) in ids}
def _first_balance(f: Decimal) -> tuple:
return (f if f > 0 else ZERO, -f if f < 0 else ZERO)
def _do_load(caller: dict, doors: dict, *, reload: bool) -> dict:
reader = odoo.ports.reader
db = doors["db"]
try:
cutover = dates.parse(doors["cutover_date"])
except ValueError:
raise Refused(422, f"cutover_date is a date like 2026-09-30, not {doors['cutover_date']!r}") from None
run_id = _run_id(doors)
actor, actor_sub = who(caller), sub_of(caller)
at = real_now()
counts = Counts()
with repository() as repo, decide():
require("load the Odoo books", [_doors_gate(doors)], status=409)
prior = RUN(repo).get(run_id)
if prior and prior["status"] == "signed":
raise refusal("not_signed", f"the run {run_id} is signed: a signed run is never rewritten", "a later cut-over only adds what Odoo posted after it")
if prior and prior["status"] == "discarded":
raise RuleBroken(f"the run {run_id} was discarded", status=409)
floor = _prior_signed_cutover(repo, db, cutover)
codes = {c["odoo_id"]: c["company_code"] for c in doors["companies"]}
odoo_companies = {c["id"]: c for c in reader.companies()}
for cid in codes:
if cid not in odoo_companies:
raise Refused(422, f"Odoo has no company {cid}")
# ── gates that read the neighbours' maps ──
with repository() as repo:
start_by_company = {}
for cid, code in codes.items():
oc = odoo_companies[cid]
fake = {"financial_year_end_month": int(oc.get("fiscalyear_last_month") or 12)}
start_by_company[cid] = _history_start(repo, doors, fake, cutover)
partners, taxes, accounts_by_company = _gather_mappings(reader, doors, list(codes), start_by_company, cutover, floor)
pmap = {}
for chunk in _chunks(sorted(partners), 200):
pmap.update(_resolve_map(seams.neighbours.partner_map(db, chunk, who(caller)), set(chunk)))
tmap = {}
for chunk in _chunks(sorted(taxes), 200):
tmap.update(_resolve_map(seams.neighbours.tax_map(db, "account.tax", chunk, who(caller)), set(chunk)))
gates = [unmet("partners_mapped", f"{len(partners - set(pmap))} Odoo partner(s) on receivable or payable lines have no bp_id "
f"({', '.join(map(str, sorted(partners - set(pmap))[:8]))})", "run the business-partner migration first")
if partners - set(pmap) else met("partners_mapped"),
unmet("taxes_mapped", f"{len(taxes - set(tmap))} Odoo tax(es) on tax lines have no tax_code ({', '.join(map(str, sorted(taxes - set(tmap))[:8]))})",
"run the tax migration first") if taxes - set(tmap) else met("taxes_mapped")]
require("load the Odoo books", gates, status=409)
# ── companies, chart, ledger ──
with repository() as repo, decide():
if repo.charts.get(CHART) is None:
repo.charts.put({"chart_id": CHART, "name": "Chart of accounts (Odoo)", "kind": "operational", "country": None})
lead = repo.ledgers.leading()
if lead is None:
repo.ledgers.put({"ledger_id": doors["ledger_id"], "name": doors["ledger_id"], "kind": "leading",
"accounting_principle": doors.get("accounting_principle", "other"), "reval_method": "closing_rate"})
has_default = repo.companies.default() is not None
for i, (cid, code) in enumerate(codes.items()):
oc = odoo_companies[cid]
if repo.companies.get(code) is None:
bp = None
if oc.get("partner_id"):
ans = seams.neighbours.partner_map(db, [oc["partner_id"]], who(caller))
bp = _resolve_map(ans, {oc["partner_id"]}).get(oc["partner_id"])
repo.companies.put({"company_code": code, "name": oc["name"], "country": next((c.get("country") for c in doors["companies"] if c["company_code"] == code), None),
"functional_currency": oc["currency"], "chart_id": CHART, "financial_year_end_month": int(oc.get("fiscalyear_last_month") or 12),
"bp_id": int(bp) if bp else None, "default_profit_centre": None, "splitting_from_fiscal_year": None,
"is_default": (not has_default and i == 0), "active_until": None})
counts.bump("company", "created")
else:
counts.bump("company", "skipped")
_map_put(repo, db, "res.company", cid, code, run_id)
lg = repo.ledgers.leading()["ledger_id"]
if repo.company_ledgers.get(code, lg) is None:
repo.company_ledgers.put({"company_code": code, "ledger_id": lg, "from_fiscal_year": None, "until_fiscal_year": None})
# ── accounts, journals' books, dimensions ──
analytic = {a["id"]: a for a in reader.analytic_accounts()}
pc_plan, cc_plan = doors.get("profit_centre_plan"), doors.get("cost_centre_plan")
pc_of = {aid: (a.get("code") or a["name"]) for aid, a in analytic.items() if pc_plan and a.get("plan") == pc_plan}
cc_of = {aid: (a.get("code") or a["name"]) for aid, a in analytic.items() if cc_plan and a.get("plan") == cc_plan}
with repository() as repo, decide():
seen = set()
for cid, accs in accounts_by_company.items():
for a in accs.values():
if a["id"] in seen:
continue
seen.add(a["id"])
aid = derived_id(f"odoo:{db}:account.account:{a['id']}")
typ, sub = odoo_map.account_type(a)
row = {"account_id": aid, "chart_id": CHART, "code": a["code"], "name": a["name"], "type": typ, "subtype": sub,
"normal_balance": "debit" if typ in ("asset", "expense") else "credit", "currency": a.get("currency") or None,
"parent_account_id": None, "monetary": bool(a.get("currency")), "source_only": sub in ("receivable", "payable"),
"fs_cost_centre": "optional", "fs_profit_centre": "optional", "fs_bp_id": "optional", "fs_tax_code": "optional",
"default_profit_centre": None, "movement_tracked": False, "active_until": None}
if _map_put(repo, db, "account.account", a["id"], aid, run_id):
if repo.accounts.get(aid) is None:
repo.accounts.put(row)
counts.bump("account", "created")
else:
cur = repo.accounts.get(aid)
if cur and cur["name"] != a["name"] and not repo.accounts.used(aid):
repo.accounts.set((aid,), name=a["name"])
counts.bump("account", "updated")
else:
counts.bump("account", "skipped")
for cid, code in codes.items():
for j in reader.journals(cid):
_map_put(repo, db, "account.journal", j["id"], odoo_map.book_for_journal_type(j.get("type")), run_id)
segs = doors.get("segments") or {}
for aid, pc in {**pc_of, **cc_of}.items():
a = analytic[aid]
if aid in pc_of:
if repo.dims.profit_centres.get(pc) is None:
repo.dims.profit_centres.put({"profit_centre": pc, "name": a["name"], "parent_profit_centre": None, "company_codes": None, "valid_from": None, "valid_until": None})
counts.bump("profit_centre", "created")
seg = segs.get(pc) or segs.get("*")
if seg:
if repo.dims.segments.get(seg) is None:
repo.dims.segments.put({"segment": seg, "name": seg, "active_until": None})
repo.dims.pc_segments.put({"profit_centre": pc, "valid_from": date(1900, 1, 1), "segment": seg})
_map_put(repo, db, "account.analytic.account", aid, f"profit_centre:{pc}", run_id)
else:
if repo.dims.cost_centres.get(pc) is None:
repo.dims.cost_centres.put({"cost_centre": pc, "name": a["name"], "company_code": None, "active_until": None})
counts.bump("cost_centre", "created")
_map_put(repo, db, "account.analytic.account", aid, f"cost_centre:{pc}", run_id)
# ── fiscal years and periods ──
horizon_by_company = {}
with repository() as repo, decide():
for cid, code in codes.items():
oc = odoo_companies[cid]
comp = repo.companies.get(code)
start = start_by_company[cid]
first_day = (start - timedelta(days=1)) if start else None
moves0 = reader.moves(cid, start, cutover)
earliest = min([m["date"] for m in moves0], default=cutover)
from_year = dates.fiscal_year_of(first_day if (first_day and first_day < earliest) else earliest, int(comp["financial_year_end_month"]))
to_year = dates.fiscal_year_of(cutover + timedelta(days=1), int(comp["financial_year_end_month"]))
for y in range(from_year, to_year + 1):
from brick_gl.application.periods import _open_year_rows
if _open_year_rows(repo, comp, y, at):
counts.bump("fiscal_year", "created")
horizon_by_company[cid] = (from_year, to_year)
_map_put(repo, db, "res.company.fiscal", cid, f"{code}:FY{from_year}-FY{to_year}", run_id)
# ── rates ──
with repository() as repo, decide():
for cid, code in codes.items():
comp = repo.companies.get(code)
for r in reader.rates(cid, start_by_company[cid], cutover):
ours = odoo_map.rate_from_odoo(r)
if ours is None or r["currency"] == comp["functional_currency"]:
continue
day = r["date"] if isinstance(r["date"], date) else dates.parse(str(r["date"]))
if repo.fx.get(r["currency"], comp["functional_currency"], "spot", day) is None:
repo.fx.add({"from_ccy": r["currency"], "to_ccy": comp["functional_currency"], "rate_type": "spot", "rate_date": day, "rate": ours,
"source": "odoo", "entered_by": actor, "entered_sub": actor_sub, "entered_at": utc(at)})
counts.bump("fx_rate", "created")
else:
counts.bump("fx_rate", "skipped")
# ── the account map: receivable / payable / revenue defaults and tax purposes ──
with repository() as repo, decide():
for cid, code in codes.items():
defaults = reader.company_defaults(cid) or {}
first = dates.period_bounds(dates.period_id(start_by_company[cid] or cutover))[0] if start_by_company[cid] else date(1900, 1, 1)
for purpose, key in (("receivable", "property_account_receivable_id"), ("payable", "property_account_payable_id"),
("revenue", "property_account_income_categ_id")):
oid = defaults.get(key)
ours = _map_get(repo, db, "account.account", oid) if oid else None
if ours and not repo.account_map.has_any(CHART, purpose, ""):
repo.account_map.add({"chart_id": CHART, "purpose": purpose, "qualifier": "", "valid_from": first, "account_id": int(ours),
"set_by": actor, "set_at": utc(at)})
counts.bump("account_map", "created")
for aid_odoo, acc in accounts_by_company[cid].items():
if odoo_map.account_type(acc)[1] == "retained_earnings" and not repo.account_map.has_any(CHART, "retained_earnings", ""):
repo.account_map.add({"chart_id": CHART, "purpose": "retained_earnings", "qualifier": "", "valid_from": first,
"account_id": int(_map_get(repo, db, "account.account", aid_odoo)), "set_by": actor, "set_at": utc(at)})
counts.bump("account_map", "created")
tids = sorted(taxes)
for rl in reader.repartition_lines(tids):
if rl.get("repartition_type") != "tax" or rl["tax_id"] not in tmap:
continue
ours = _map_get(repo, db, "account.account", rl["account_id"]) if rl.get("account_id") else None
q = str(tmap[rl["tax_id"]])
if ours and not repo.account_map.has_any(CHART, "tax", q):
repo.account_map.add({"chart_id": CHART, "purpose": "tax", "qualifier": q, "valid_from": first, "account_id": int(ours),
"set_by": actor, "set_at": utc(at)})
counts.bump("account_map", "created")
# ── the journals ──
touched: dict = {}
for cid, code in codes.items():
touched.update(_load_journals(caller, doors, cid, code, start_by_company[cid], cutover, floor, run_id, pmap, tmap, pc_of, cc_of, counts, at))
# ── recurring, deferrals, budgets, group mapping ──
_load_planning(caller, doors, codes, cutover, run_id, counts, at)
# ── lock dates close months; balances derived; the run recorded ──
with repository() as repo, decide():
for cid, code in codes.items():
oc = odoo_companies[cid]
lock = max([d for d in (oc.get("fiscalyear_lock_date"), oc.get("hard_lock_date")) if d] or [None], default=None) if any(
oc.get(k) for k in ("fiscalyear_lock_date", "hard_lock_date")) else None
period_lock = oc.get("period_lock_date")
if doors.get("lock_months", "close") == "close" and period_lock and (lock is None or period_lock > lock):
lock = period_lock
if lock:
for p in repo.periods.of_company(code):
if p["ends_on"] <= lock and p["status"] != "closed":
repo.conn.execute(t.gl_period.update().where(t.gl_period.c.company_code == code, t.gl_period.c.period_id == p["period_id"]).values(
status="closed", closed_by="migration", closed_sub="migration", closed_at=utc(at)))
repo.periods.log(code, p["period_id"], "close", f"Odoo lock date {lock}", "migration", "migration", at)
counts.bump("period", "closed")
for (code, period_id), accounts in sorted(touched.items()):
with repository() as repo:
ledgerops.derive_balances(repo, code, period_id, sorted(accounts))
with repository() as repo, decide():
checks = (json.loads(prior["checks_json"]) if prior and prior["checks_json"] else [])
row = {"run_id": run_id, "odoo_db": db, "cutover_date": cutover, "status": "loaded", "doors_json": json.dumps(doors, default=str),
"checks_json": json.dumps(checks, default=str) if reload else "[]", "counts_json": json.dumps(counts, default=str), "checksums": None,
"loaded_at": utc(at), "signed_by": None, "signed_at": None}
RUN(repo).put(row)
return _run_view(row)
def _run_view(r: dict) -> dict:
return {"run_id": r["run_id"], "odoo_db": r["odoo_db"], "cutover_date": r["cutover_date"], "status": r["status"],
"doors": json.loads(r["doors_json"] or "{}"), "counts": json.loads(r["counts_json"] or "{}"), "checks": json.loads(r["checks_json"] or "[]"),
"signed_by": r["signed_by"], "signed_at": r["signed_at"], "checksums": r["checksums"]}
class _Chain:
"""The next chain link and entry numbers of a company during a batch load: read once from the committed rows, then counted locally
(a batch is the only writer of the company while it loads, and every write is a plain INSERT, so a stray writer still collides)."""
def __init__(self, repo, code: str):
top = repo.conn.execute(select(func.max(t.gl_chain_claim.c.chain_seq)).where(t.gl_chain_claim.c.company_code == code)).scalar() or 0
self.seq = int(top)
r = repo.conn.execute(select(t.gl_journal.c.row_hash).where(t.gl_journal.c.company_code == code, t.gl_journal.c.chain_seq == self.seq)).first() if top else None
self.prev = r[0] if r and r[0] else hashchain.GENESIS
self.numbers: dict = {}
def used(self, repo, code: str, book: str, fy: str) -> set:
k = (book, fy)
if k not in self.numbers:
rows = repo.conn.execute(select(t.gl_number_series.c.entry_no).where(
t.gl_number_series.c.company_code == code, t.gl_number_series.c.book == book, t.gl_number_series.c.fiscal_year == fy))
self.numbers[k] = {int(r[0]) for r in rows}
return self.numbers[k]
def _load_journals(caller, doors, cid, code, start, cutover, floor, run_id, pmap, tmap, pc_of, cc_of, counts, at) -> dict:
reader = odoo.ports.reader
db = doors["db"]
touched: dict = {}
lo = start
if floor and (lo is None or lo <= floor):
lo = floor + timedelta(days=1)
with repository() as repo:
company = repo.companies.get(code)
accounts = {a["id"]: a for a in reader.accounts(cid)}
journals = {j["id"]: j for j in reader.journals(cid)}
moves = sorted(reader.moves(cid, lo, cutover), key=lambda m: (m["date"], m["id"]))
# one opening journal for everything before the door's start
if start and not floor and doors.get("history") != "all":
_load_opening(caller, doors, cid, code, company, start, run_id, counts, at, accounts, touched)
batch_no = 0
for chunk in _chunks(moves, 200):
batch_no += 1
lines_by_move: dict = {}
for ln in reader.lines([m["id"] for m in chunk]):
if ln.get("display_type") in ("line_section", "line_note"):
continue
lines_by_move.setdefault(ln["move_id"], []).append(ln)
with repository() as repo, decide():
chain = _Chain(repo, code)
for m in chunk:
jid = derived_id(f"odoo:{db}:account.move:{m['id']}")
if repo.journals.get(jid) is not None or MAP(repo).get(db, "account.move", str(m["id"])) is not None:
counts.bump("journal", "skipped")
continue
if m.get("state") != "posted":
continue
_write_move(repo, chain, company, db, m, sorted(lines_by_move.get(m["id"], []), key=lambda x: x["id"]), accounts, journals, pmap, tmap, pc_of,
cc_of, doors, run_id, at, touched)
counts.bump("journal", "created")
# a reversed move's original is 'reversed' once both are carried
with repository() as repo:
for m in moves:
if m.get("reversed_entry_id"):
orig = derived_id(f"odoo:{db}:account.move:{m['reversed_entry_id']}")
o = repo.journals.get(orig)
if o and o["status"] == "posted":
repo.journals.set((orig,), status="reversed", reversal_status="posted")
return touched
def _load_opening(caller, doors, cid, code, company, start, run_id, counts, at, accounts, touched) -> None:
reader = odoo.ports.reader
db = doors["db"]
day = start - timedelta(days=1)
key = derived_id(f"odoo:{db}:opening:{day}", code)
bal = [b for b in reader.opening_balances(cid, start) if Decimal(str(b["balance"])) != 0 or Decimal(str(b.get("amount_currency") or 0)) != 0]
if not bal:
return
with repository() as repo, decide():
if repo.journals.get(key) is not None:
counts.bump("journal", "skipped")
return
chain = _Chain(repo, company["company_code"])
lines = []
for i, b in enumerate(sorted(bal, key=lambda x: (x["account_id"], x.get("currency") or "")), 1):
f = q2(Decimal(str(b["balance"])))
aid = int(_map_get(repo, db, "account.account", b["account_id"]) or 0)
fd, fc = _first_balance(f)
lines.append({"line_no": i, "account_id": aid, "debit": fd, "credit": fc, "f_debit": fd, "f_credit": fc,
"currency": b.get("currency") or None, "amount_currency": Decimal(str(b["amount_currency"])) if b.get("amount_currency") is not None else None})
move = {"id": f"opening:{day}", "name": f"Opening {day}", "date": day, "journal_type": "general", "ref": "opening balances",
"currency": company["functional_currency"], "reversed_entry_id": None, "company_id": cid}
_write_raw(repo, chain, company, db, key, move, lines, "opening", "opening", run_id, at, touched,
source_document=f"odoo:{db}:opening:{day}:{company['company_code']}")
counts.bump("journal", "created")
def _write_move(repo, chain, company, db, m, lines, accounts, journals, pmap, tmap, pc_of, cc_of, doors, run_id, at, touched) -> None:
jtype = (journals.get(m.get("journal_id")) or {}).get("type") or "general"
book = odoo_map.book_for_journal_type(jtype)
fn = company["functional_currency"]
mcur = m.get("currency") or fn
rows = []
for i, ln in enumerate(lines, 1):
acc = accounts.get(ln["account_id"])
sub = odoo_map.account_type(acc)[1] if acc else "other"
deb, cred = q2(Decimal(str(ln.get("debit") or 0))), q2(Decimal(str(ln.get("credit") or 0)))
lc = ln.get("currency") or None
ac = Decimal(str(ln["amount_currency"])) if ln.get("amount_currency") is not None else None
if lc and lc == mcur and mcur != fn and ac is not None:
ddeb, dcred = (q2(ac) if ac > 0 else ZERO), (q2(-ac) if ac < 0 else ZERO)
else:
ddeb, dcred = deb, cred
bp = pmap.get(ln.get("partner_id")) if ln.get("partner_id") and sub in ("receivable", "payable") else None
tax = tmap.get(ln.get("tax_line_id")) if ln.get("tax_line_id") else None
dist = odoo_map.share_pcts(ln.get("analytic_distribution")) if doors.get("analytic_distribution", "profit_centre") == "profit_centre" else []
pcs = [(pc_of[a], s) for a, s in dist if a in pc_of]
ccs = [(cc_of[a], s) for a, s in dist if a in cc_of]
rows.append({"line_no": i, "account_id": int(_map_get(repo, db, "account.account", ln["account_id"]) or 0), "debit": ddeb, "credit": dcred,
"f_debit": deb, "f_credit": cred, "currency": lc if lc and lc != fn else None, "amount_currency": q2(ac) if (ac is not None and lc and lc != fn) else None,
"bp_id": int(bp) if bp else None, "tax_code": str(tax) if tax else None, "cost_centre": ccs[0][0] if ccs else None,
"profit_centre": pcs[0][0] if pcs else None, "pcs": pcs, "memo": ln.get("name"), "odoo_line_id": str(ln["id"]),
"partner_company": None})
jid = derived_id(f"odoo:{db}:account.move:{m['id']}")
m2 = {**m, "journal_type": jtype, "currency": mcur}
_write_raw(repo, chain, company, db, jid, m2, rows, book, "posted", run_id, at, touched, source_document=f"odoo:{db}:account.move:{m['id']}")
_map_put(repo, db, "account.move", m["id"], jid, run_id)
def _write_raw(repo, chain, company, db, jid, move, rows, book, event, run_id, at, touched, *, source_document) -> None:
"""Write one migrated journal exactly as Odoo had it: no period guard (history), explicit entry number, chained. Never recomputed."""
code = company["company_code"]
on = move["date"] if isinstance(move["date"], date) else dates.parse(str(move["date"]))
endm = int(company["financial_year_end_month"])
fy = dates.fy_name(dates.fiscal_year_of(on, endm))
parsed = odoo_map.parse_entry_number(move.get("name") or "")
used = chain.used(repo, code, book, fy)
n = parsed[1] if parsed and parsed[0] == on.year or (parsed and parsed[0] == dates.fiscal_year_of(on, endm)) else None
if n is None or n in used:
n = (max(used) if used else 0) + 1
used.add(n)
chain.seq += 1
seq = chain.seq
f_total = sum((r["f_debit"] for r in rows), ZERO)
d_total = sum((r["debit"] for r in rows), ZERO)
fx = ZERO
foreign = [r for r in rows if r["currency"] and r["debit"] + r["credit"] > 0 and r["f_debit"] + r["f_credit"] > 0]
rate = (foreign[0]["f_debit"] + foreign[0]["f_credit"]) / (foreign[0]["debit"] + foreign[0]["credit"]) if foreign else Decimal(1)
fx = rate.quantize(Decimal("0.000001"))
cur = move.get("currency") or company["functional_currency"]
rev = move.get("reversed_entry_id")
header = {"journal_id": jid, "company_code": code, "ledger_id": None, "status": "posted", "journal_date": on, "period_id": dates.period_id(on), "fiscal_year": fy,
"book": book, "entry_no": n, "source_module": "odoo", "source_document": source_document, "event": event, "currency": cur, "fx_rate": fx,
"memo": (move.get("ref") or move.get("name") or "")[:2000], "total": d_total, "functional_total": f_total, "adjustment": False,
"reverse_next_period": False, "reversal_status": "none", "reversal_reason": None, "ic_document_id": None, "recurring_id": None, "accrual_id": None,
"allocation_run_id": None, "reval_run_id": None, "proposed_by": "odoo", "proposed_sub": "odoo", "approved_by": "odoo", "approved_sub": "odoo",
"rejected_by": None, "reject_reason": None, "posted_at": utc(at), "reverses_journal_id": derived_id(f"odoo:{db}:account.move:{rev}") if rev else None,
"body_digest": "", "chain_seq": seq, "prev_hash": chain.prev, "migration_run_id": run_id}
line_rows = [{"journal_id": jid, "line_no": r["line_no"], "account_id": r["account_id"], "purpose": None, "qualifier": None, "debit": r["debit"], "credit": r["credit"],
"functional_debit": r["f_debit"], "functional_credit": r["f_credit"], "currency": r["currency"], "amount_currency": r["amount_currency"],
"bp_id": r.get("bp_id"), "tax_code": r.get("tax_code"), "cost_centre": r.get("cost_centre"), "profit_centre": r.get("profit_centre"),
"partner_company": r.get("partner_company"), "movement_type": "opening" if event == "opening" else None, "against_document": None,
"ic_reference": None, "asset_id": None, "line_memo": r.get("memo"), "odoo_line_id": r.get("odoo_line_id")} for r in rows]
splits = []
for r in rows:
shares = r.get("pcs") or []
if len(shares) > 1:
parts_d = largest_remainder(r["debit"], [s for _, s in shares]) if r["debit"] else [ZERO] * len(shares)
parts_c = largest_remainder(r["credit"], [s for _, s in shares]) if r["credit"] else [ZERO] * len(shares)
parts_fd = largest_remainder(r["f_debit"], [s for _, s in shares]) if r["f_debit"] else [ZERO] * len(shares)
parts_fc = largest_remainder(r["f_credit"], [s for _, s in shares]) if r["f_credit"] else [ZERO] * len(shares)
for k, (pc, _s) in enumerate(shares, 1):
splits.append((r, k, pc, parts_d[k - 1], parts_c[k - 1], parts_fd[k - 1], parts_fc[k - 1]))
else:
splits.append((r, 1, r.get("profit_centre"), r["debit"], r["credit"], r["f_debit"], r["f_credit"]))
split_rows = [{
"journal_id": jid, "line_no": r["line_no"], "split_no": k, "company_code": code, "ledger_id": None, "account_id": r["account_id"],
"period_id": header["period_id"], "journal_date": on, "debit": d, "credit": c, "functional_debit": fd, "functional_credit": fc, "currency": cur,
"profit_centre": pc, "segment": repo.dims.segment_of(pc, on) if pc else None, "cost_centre": r.get("cost_centre"), "partner_company": None,
"movement_type": "opening" if event == "opening" else None, "bp_id": r.get("bp_id"), "asset_id": None, "is_clearing": False, "adjustment": False}
for (r, k, pc, d, c, fd, fc) in splits]
header["row_hash"] = hashchain.row_hash(chain.prev, header, line_rows, split_rows)
chain.prev = header["row_hash"]
repo.chain.add({"company_code": code, "chain_seq": seq, "journal_id": jid, "claimed_at": utc(at)})
repo.numbers.add({"company_code": code, "book": book, "fiscal_year": fy, "entry_no": n, "journal_id": jid, "claimed_at": utc(at)})
repo.journals.add(header)
repo.journals.add_lines(line_rows)
repo.journals.add_splits(split_rows)
touched.setdefault((code, header["period_id"]), set()).update(r["account_id"] for r in rows)
def _load_planning(caller, doors, codes, cutover, run_id, counts, at) -> None:
"""Recurring templates and deferrals still running become PROPOSALS (never auto-approved); the current year's budgets an approved v1;
consolidation accounts the group chart and its mapping."""
reader = odoo.ports.reader
db = doors["db"]
after = cutover
with repository() as repo, decide():
for cid, code in codes.items():
comp = repo.companies.get(code)
first = dates.period_id(cutover + timedelta(days=1))
if doors.get("recurring", "proposals") == "proposals":
for r in reader.recurring_moves(cid, after):
freq = odoo_map.recurring_frequency(r.get("auto_post"))
rid = f"ODOO-{r['id']}"
if not freq or repo.conn.execute(t.gl_recurring_journal.select().where(t.gl_recurring_journal.c.recurring_id == rid)).first():
counts.bump("recurring", "skipped")
continue
lines = []
for ln in r.get("lines", []):
aid = _map_get(repo, db, "account.account", ln["account_id"])
if not aid:
continue
d = {"account_id": str(aid)}
if q2(Decimal(str(ln.get("debit") or 0))) > 0:
d["debit"] = f"{q2(Decimal(str(ln['debit']))):.2f}"
else:
d["credit"] = f"{q2(Decimal(str(ln.get('credit') or 0))):.2f}"
lines.append(d)
until = r.get("auto_post_until")
last = dates.period_id(until) if until else None
_Keyed(repo.conn, t.gl_recurring_journal, ("recurring_id", "version")).add({
"recurring_id": rid, "version": 1, "company_code": code, "ledger_id": None, "lines_json": json.dumps(lines), "memo": r.get("name") or rid,
"frequency": freq, "first_period": first, "last_period": last, "posting_day": "last", "status": "proposed", "proposed_by": "odoo",
"proposed_sub": "odoo", "approved_by": None, "approved_sub": None, "ended_at": None, "end_reason": None})
_map_put(repo, db, "account.move.recurring", r["id"], f"{rid}:1", run_id)
counts.bump("recurring", "created")
for d in reader.deferral_lines(cid, after):
acode = f"ODOO-{d['id']}"
if repo.conn.execute(t.gl_accrual.select().where(t.gl_accrual.c.company_code == code, t.gl_accrual.c.accrual_code == acode)).first():
counts.bump("accrual", "skipped")
continue
balance_id = _map_get(repo, db, "account.account", d["account_id"])
pl_id = _map_get(repo, db, "account.account", d.get("counter_account_id")) if d.get("counter_account_id") else None
amount = q2(Decimal(str(d["balance"])) - Decimal(str(d.get("released") or 0)))
if not balance_id or not pl_id or amount <= 0:
counts.bump("accrual", "skipped")
continue
acc = repo.accounts.get(int(balance_id))
kind = "deferred_revenue" if acc["type"] == "liability" else "prepaid_expense"
end = dates.period_id(d["deferred_end_date"])
if end < first:
counts.bump("accrual", "skipped")
continue
_Keyed(repo.conn, t.gl_accrual, ("company_code", "accrual_code")).add({
"company_code": code, "accrual_code": acode, "kind": kind, "ledger_id": None, "amount": amount, "currency": comp["functional_currency"],
"pl_account_id": int(pl_id), "balance_account_id": int(balance_id), "cost_centre": None, "profit_centre": None, "start_period": first,
"end_period": end, "method": "straight_line", "source_journal_id": None, "status": "proposed", "registered_by": "odoo",
"registered_sub": "odoo", "approved_by": None, "approved_sub": None, "terminated_period": None, "reject_reason": None})
_map_put(repo, db, "account.move.line.deferral", d["id"], f"{code}:{acode}", run_id)
counts.bump("accrual", "created")
for b in reader.budgets(cid):
bid = f"ODOO-{b['id']}"
if repo.conn.execute(t.gl_budget.select().where(t.gl_budget.c.budget_id == bid)).first():
counts.bump("budget", "skipped")
continue
d0, d1 = b["date_from"], b["date_to"]
fy = dates.fy_name(dates.fiscal_year_of(d0, int(comp["financial_year_end_month"])))
if repo.years.get(code, fy) is None:
counts.bump("budget", "skipped")
continue
periods = []
p = dates.period_id(d0)
while p <= dates.period_id(d1):
periods.append(p)
p = dates.next_period(p)
lg = repo.ledgers.leading()["ledger_id"]
_Keyed(repo.conn, t.gl_budget, ("budget_id", "version")).add({
"budget_id": bid, "version": 1, "company_code": code, "ledger_id": lg, "fiscal_year": fy, "status": "approved", "submitted_by": "odoo",
"submitted_sub": "odoo", "approved_by": "migration", "approved_sub": "migration", "return_reason": None})
K = _Keyed(repo.conn, t.gl_budget_line, ("budget_id", "version", "account_id", "profit_centre", "cost_centre", "period_id"))
for ln in b.get("lines", []):
aid = _map_get(repo, db, "account.account", ln["account_id"])
if not aid:
continue
# spread evenly over the months, the remainder in the last
each = (Decimal(str(ln["planned_amount"])) / len(periods)).quantize(Decimal("0.01"), rounding="ROUND_DOWN")
amounts = [each] * (len(periods) - 1) + [q2(Decimal(str(ln["planned_amount"]))) - each * (len(periods) - 1)]
pc = _map_get(repo, db, "account.analytic.account", ln.get("analytic_account_id")) if ln.get("analytic_account_id") else None
pc = pc.split(":", 1)[1] if pc and pc.startswith("profit_centre:") else ""
for per, amt in zip(periods, amounts):
K.add({"budget_id": bid, "version": 1, "account_id": int(aid), "profit_centre": pc, "cost_centre": "", "period_id": per, "amount": amt})
_map_put(repo, db, "crossovered.budget", b["id"], f"{bid}:1", run_id)
counts.bump("budget", "created")
cons = reader.consolidation_accounts()
if cons:
gid = doors.get("group_chart_id", "GROUP")
if repo.charts.get(gid) is None:
repo.charts.put({"chart_id": gid, "name": "Group chart (Odoo consolidation)", "kind": "group", "country": None})
for c in cons:
ga = derived_id("account", gid, c["code"])
if repo.accounts.get(ga) is None:
repo.accounts.put({"account_id": ga, "chart_id": gid, "code": c["code"], "name": c["name"], "type": c.get("type", "asset"), "subtype": "other",
"normal_balance": "debit", "currency": None, "parent_account_id": None, "monetary": False, "source_only": False,
"fs_cost_centre": "optional", "fs_profit_centre": "optional", "fs_bp_id": "optional", "fs_tax_code": "optional",
"default_profit_centre": None, "movement_tracked": False, "active_until": None})
counts.bump("group_account", "created")
for oid in c.get("account_ids", []):
ours = _map_get(repo, db, "account.account", oid)
if ours and not repo.group_map.history(int(ours)):
repo.group_map.add({"account_id": int(ours), "valid_from": date(1900, 1, 1), "group_account_id": ga, "set_by": "migration", "set_at": utc(at)})
counts.bump("group_mapping", "created")
@GL.use_case(kind="command", route="POST /migration/odoo/load", permission=MIGRATION_LOAD, idempotent=True)
def load_gl_from_odoo(caller: dict, doors: DoorsModel) -> dict:
"""Load the Odoo books (machine). Gates: the doors decided, every Odoo partner on a receivable or payable line mapped to a bp_id
(business-partner's migration ran first) and every tax on a tax line to a tax_code (tax's ran first). Writes read-only history keyed
on the Odoo record - companies, accounts and the account map, fiscal years and periods (closed up to Odoo's lock date), rates, dimensions,
posted journals of every move type with their Odoo entry numbers, recurring entries and deferrals as PROPOSALS, the current year's
budgets, the group mapping - and answers per kind what it created and skipped. A second run changes nothing."""
with decide():
return _do_load(caller, doors.model_dump(mode="json"), reload=False)
@GL.use_case(kind="command", route="POST /migration/odoo/reload", permission=MIGRATION_LOAD, idempotent=True)
def reload_gl_migration(caller: dict, doors: DoorsModel) -> dict:
"""Load again after a fix (machine): the same keys, explanations on unchanged checks kept; refused once the run is signed."""
with decide():
return _do_load(caller, doors.model_dump(mode="json"), reload=True)
# ═══ the reconciliation ═════════════════════════════════════════════════════════════════════════════════════════════════
def _s(v) -> str:
return f"{Decimal(str(v)):.2f}"
def _checks(repo, doors: dict, run_id: str, caller: dict) -> list:
reader = odoo.ports.reader
db = doors["db"]
cutover = dates.parse(doors["cutover_date"])
out = []
codes = {c["odoo_id"]: c["company_code"] for c in doors["companies"]}
ccodes = {c["company_code"] for c in doors["companies"]}
def add(name: str, odoo_v, ours_v, must: str) -> None:
o, u = json.loads(json.dumps(odoo_v, default=str, sort_keys=True)), json.loads(json.dumps(ours_v, default=str, sort_keys=True))
out.append({"check": name, "odoo": o, "ours": u, "equal": o == u, "must": must, "explanation": ""})
s = t.gl_split_line.c
j = t.gl_journal.c
mine = j.migration_run_id == run_id
# 1. trial balance per company, account and profit centre at cutover
o1, u1 = {}, {}
for cid, code in codes.items():
for r in reader.tb_by_account_pc(cid, cutover):
ours = _map_get(repo, db, "account.account", r["account_id"])
key = f"{code}|{ours}|{r.get('profit_centre') or ''}"
o1[key] = _s(Decimal(str(r["debit"])) - Decimal(str(r["credit"])))
q = (select(s.account_id, s.profit_centre, func.sum(s.functional_debit - s.functional_credit)).select_from(
t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(s.company_code == code, s.journal_date <= cutover, mine)
.group_by(s.account_id, s.profit_centre))
for aid, pc, net in repo.conn.execute(q):
if net and q2(net) != 0:
u1[f"{code}|{aid}|{pc or ''}"] = _s(net)
o1 = {k: v for k, v in o1.items() if Decimal(v) != 0}
add("trial balance per company, account and profit centre at cutover", o1, u1, "equal")
# 2. intercompany balance per pair
o2, u2 = {}, {}
pcode = {}
for cid, code in codes.items():
for r in reader.partner_balances(cid, cutover):
if r.get("partner_company_id") in codes:
o2[f"{code}|{codes[r['partner_company_id']]}"] = _s(r["balance"])
q = (select(s.partner_company, func.sum(s.functional_debit - s.functional_credit)).select_from(
t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(
s.company_code == code, s.journal_date <= cutover, mine, s.partner_company.isnot(None)).group_by(s.partner_company))
for p, net in repo.conn.execute(q):
if net and q2(net) != 0:
u2[f"{code}|{p}"] = _s(net)
_ = pcode
add("intercompany balance per pair of companies at cutover", {k: v for k, v in o2.items() if Decimal(v) != 0}, u2, "equal")
# 3. trial balance per account and period (debit, credit, cumulative closing)
o3, u3 = {}, {}
for cid, code in codes.items():
cum: dict = {}
opening = {}
for b in reader.opening_balances(cid, _history_start_for(repo, doors, cid, code, cutover) or date(1900, 1, 1)):
opening[b["account_id"]] = opening.get(b["account_id"], ZERO) + Decimal(str(b["balance"]))
cum.update(opening)
for r in sorted(reader.tb_by_account_period(cid, cutover), key=lambda x: (x["period"], x["account_id"])):
ours = _map_get(repo, db, "account.account", r["account_id"])
cum[r["account_id"]] = cum.get(r["account_id"], ZERO) + Decimal(str(r["debit"])) - Decimal(str(r["credit"]))
o3[f"{code}|{ours}|{r['period']}"] = [_s(r["debit"]), _s(r["credit"]), _s(cum[r["account_id"]])]
cumu: dict = {}
q = (select(s.account_id, s.period_id, func.sum(s.functional_debit), func.sum(s.functional_credit), func.min(s.movement_type)).select_from(
t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(s.company_code == code, mine, j.event != "opening").group_by(
s.account_id, s.period_id).order_by(s.period_id))
op_rows = repo.conn.execute(select(s.account_id, func.sum(s.functional_debit - s.functional_credit)).select_from(
t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(s.company_code == code, mine, j.event == "opening").group_by(s.account_id))
for aid, net in op_rows:
cumu[aid] = q2(net)
for aid, per, d, c, _mt in repo.conn.execute(q):
cumu[aid] = cumu.get(aid, ZERO) + q2(d) - q2(c)
u3[f"{code}|{aid}|{per}"] = [_s(d), _s(c), _s(cumu[aid])]
add("trial balance per account and period (debit, credit, closing)", o3, u3, "equal")
# 4. nets to zero on both sides
o4 = ZERO
u4 = ZERO
for cid, code in codes.items():
o4 += sum((Decimal(str(r["debit"])) - Decimal(str(r["credit"])) for r in reader.tb_by_account_pc(cid, cutover)), ZERO)
q = select(func.sum(s.functional_debit - s.functional_credit)).select_from(t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(
s.company_code == code, s.journal_date <= cutover, mine)
u4 += q2(repo.conn.execute(q).scalar() or ZERO)
add("trial balance at cutover nets to zero on both sides", _s(o4), _s(u4), "equal")
# 5. foreign-currency balance per account and currency
o5, u5 = {}, {}
for cid, code in codes.items():
for r in reader.foreign_balances(cid, cutover):
if Decimal(str(r["amount_currency"])) != 0:
o5[f"{code}|{_map_get(repo, db, 'account.account', r['account_id'])}|{r['currency']}"] = _s(r["amount_currency"])
q = (select(t.gl_journal_line.c.account_id, t.gl_journal_line.c.currency, func.sum(t.gl_journal_line.c.amount_currency)).select_from(
t.gl_journal_line.join(t.gl_journal, j.journal_id == t.gl_journal_line.c.journal_id)).where(
j.company_code == code, mine, j.journal_date <= cutover, t.gl_journal_line.c.currency.isnot(None)).group_by(
t.gl_journal_line.c.account_id, t.gl_journal_line.c.currency))
for aid, cur, net in repo.conn.execute(q):
if net and q2(net) != 0:
u5[f"{code}|{aid}|{cur}"] = _s(net)
add("foreign-currency balance per account and currency at cutover", o5, u5, "equal")
# 6. posted move count per journal book and period
o6, u6 = {}, {}
for cid, code in codes.items():
for r in reader.move_counts(cid, cutover):
book = _map_get(repo, db, "account.journal", r["journal_id"]) or "misc"
k = f"{code}|{book}|{r['period']}"
o6[k] = o6.get(k, 0) + int(r["count"])
q = select(j.book, j.period_id, func.count()).where(j.company_code == code, mine, j.event == "posted", j.journal_date <= cutover).group_by(j.book, j.period_id)
for b, per, n in repo.conn.execute(q):
u6[f"{code}|{b}|{per}"] = int(n)
add("posted move count per journal and period", o6, u6, "equal")
# 7. tax per tax code and quarter
o7, u7 = {}, {}
for cid, code in codes.items():
for r in reader.tax_by_quarter(cid, cutover):
tc = seams.neighbours.tax_map(db, "account.tax", [r["tax_line_id"]], who(caller))
tcode = _resolve_map(tc, {r["tax_line_id"]}).get(r["tax_line_id"], str(r["tax_line_id"]))
k = f"{code}|{tcode}|{r['quarter']}"
o7[k] = _s(Decimal(o7.get(k, "0")) + Decimal(str(r["balance"])))
q = (select(t.gl_journal_line.c.tax_code, j.journal_date, t.gl_journal_line.c.functional_debit, t.gl_journal_line.c.functional_credit).select_from(
t.gl_journal_line.join(t.gl_journal, j.journal_id == t.gl_journal_line.c.journal_id)).where(
j.company_code == code, mine, t.gl_journal_line.c.tax_code.isnot(None), j.journal_date <= cutover))
for tc, d, fd, fc in repo.conn.execute(q):
k = f"{code}|{tc}|{odoo_map.quarter(d)}"
u7[k] = _s(Decimal(u7.get(k, "0")) + q2(fd) - q2(fc))
add("tax per tax code and quarter", {k: v for k, v in o7.items() if Decimal(v) != 0}, {k: v for k, v in u7.items() if Decimal(v) != 0}, "equal")
# 8. account count by type
o8, u8 = {}, {}
for cid, code in codes.items():
for r in reader.account_counts(cid):
typ, sub = odoo_map.account_type(r)
k = f"{typ}/{sub}"
o8[k] = o8.get(k, 0) + int(r["count"])
rows = repo.conn.execute(select(t.gl_account.c.type, t.gl_account.c.subtype, func.count()).select_from(
t.gl_account.join(t.gl_migration_map, (t.gl_migration_map.c.natural_key == func.cast(t.gl_account.c.account_id, t.gl_migration_map.c.natural_key.type)) &
(t.gl_migration_map.c.odoo_model == "account.account") & (t.gl_migration_map.c.odoo_db == db))).group_by(
t.gl_account.c.type, t.gl_account.c.subtype))
for typ, sub, n in rows:
u8[f"{typ}/{sub}"] = int(n)
add("account count by type", o8, u8, "equal")
# 9. closed periods
o9, u9 = [], []
for cid, code in codes.items():
oc = next(c for c in reader.companies() if c["id"] == cid)
lock = max([d for d in (oc.get("fiscalyear_lock_date"), oc.get("hard_lock_date"), oc.get("period_lock_date") if doors.get("lock_months", "close") == "close" else None) if d] or [None]) \
if any(oc.get(k) for k in ("fiscalyear_lock_date", "hard_lock_date", "period_lock_date")) else None
ps = repo.periods.of_company(code)
o9 += [f"{code}|{p['period_id']}" for p in ps if lock and p["ends_on"] <= lock]
u9 += [f"{code}|{p['period_id']}" for p in ps if p["status"] == "closed"]
add("closed periods", sorted(o9), sorted(u9), "equal")
# 10. fx rate count per currency and year
o10, u10 = {}, {}
for cid, code in codes.items():
for r in reader.rate_counts(cid, cutover):
o10[f"{r['currency']}|{r['year']}"] = o10.get(f"{r['currency']}|{r['year']}", 0) + int(r["count"])
for rr in repo.fx.rows(t.gl_fx_rate.c.source == "odoo"):
k = f"{rr['from_ccy']}|{rr['rate_date'].year}"
u10[k] = u10.get(k, 0) + 1
add("fx rate count per currency and year", o10, u10, "equal")
# 11. the receivable control account against payments' open balance
o11 = _s(sum((Decimal(str(reader.receivable_balance(cid, cutover))) for cid in codes), ZERO))
ours11 = None
if seams.PAYMENTS.installed:
docs = [r[0] for r in repo.conn.execute(select(j.source_document).where(j.migration_run_id == run_id, j.source_document.like(f"odoo:{db}:account.move:%")).distinct())]
a = seams.neighbours.settlement(docs, cutover, who(caller)) if docs else None
ours11 = _s(sum((Decimal(str(i["outstanding"])) for i in a.data["invoices"]), ZERO)) if a is not None and a.ok else None
add("receivable control account against payments' open balance at cutover", o11, ours11, "explained")
# 12. draft moves
o12 = sum(int(reader.draft_move_count(cid, cutover)) for cid in codes)
add("draft moves dated on or before cutover", o12, 0, "explained")
_ = ccodes
return out
def _history_start_for(repo, doors: dict, cid: int, code: str, cutover: date) -> Optional[date]:
comp = repo.companies.get(code)
return _history_start(repo, doors, comp, cutover)
def _fingerprint(c: dict) -> str:
return hashlib.sha256(json.dumps([c["check"], c["odoo"], c["ours"]], sort_keys=True, default=str).encode()).hexdigest()
def _get_run(repo, run_id: str) -> dict:
r = RUN(repo).get(run_id)
if r is None:
raise Refused(404, f"no migration run {run_id}")
return r
@GL.use_case(kind="command", route="POST /migration/odoo/reconcile", permission=MIGRATION_RECONCILE, idempotent=True)
def reconcile_gl_migration(caller: dict, run_id: Annotated[str, Field(max_length=200)]) -> dict:
"""Compute every reconciliation check on BOTH sides (machine): the trial balance per company, account and profit centre at cutover; the
intercompany balance per pair; the trial balance per account and period (debit, credit, closing); that it nets to zero; foreign-
currency balances; posted move counts per journal and period; tax per code and quarter; account counts; closed periods; rate counts;
the receivable against payments' open balance (explained); draft moves (explained). Each is {check, odoo, ours, equal, must, explanation};
an explanation recorded on an unchanged check is kept."""
with repository() as repo, decide():
r = _get_run(repo, run_id)
if r["status"] in ("signed", "discarded"):
raise refusal("not_signed", f"the run {run_id} is {r['status']}", "a signed run is frozen", status=409)
doors = json.loads(r["doors_json"])
old = {_fingerprint(c): c for c in json.loads(r["checks_json"] or "[]")}
checks = _checks(repo, doors, run_id, caller)
for c in checks:
prev = old.get(_fingerprint(c))
if prev and prev.get("explanation"):
c["explanation"] = prev["explanation"]
RUN(repo).set((run_id,), status="reconciled", checks_json=json.dumps(checks, default=str), reconciled_at=utc(real_now()))
return _run_view(_get_run(repo, run_id))
@GL.use_case(kind="command", route="POST /migration/odoo/explain-difference", permission=SETTINGS_WRITE)
def explain_gl_difference(caller: dict, run_id: Annotated[str, Field(max_length=200)], check: Annotated[str, Field(max_length=300)], reason: Reason) -> dict:
"""Explain a check that differs and may be explained, with a reason recorded beside it. The trial balance per company, account and
period (every check marked must=equal) differs only if the mapping is wrong: it cannot be explained away - fix and reload."""
with repository() as repo, decide():
r = _get_run(repo, run_id)
if r["status"] != "reconciled":
raise RuleBroken(f"the run is {r['status']}: reconcile it first", status=409)
checks = json.loads(r["checks_json"])
c = next((x for x in checks if x["check"] == check), None)
if c is None:
raise Refused(404, f"no check {check!r} in {run_id}")
if c["equal"]:
raise RuleBroken("that check is equal: nothing to explain", status=409)
if c["must"] == "equal":
raise refusal("must_equal", f"{check!r} must be equal and cannot be explained away", "fix the mapping and reload")
c["explanation"] = reason
c["explained_by"] = who(caller)
RUN(repo).set((run_id,), checks_json=json.dumps(checks, default=str))
return _run_view(_get_run(repo, run_id))
def sign_gates(checks: list) -> list:
must_equal = [c["check"] for c in checks if c["must"] == "equal" and not c["equal"]]
unexplained = [c["check"] for c in checks if c["must"] == "explained" and not c["equal"] and not c["explanation"]]
return [unmet("must_equal", "a check that must be equal differs: " + "; ".join(must_equal), "fix the mapping and reload; it cannot be explained away")
if must_equal else met("must_equal"),
unmet("zero_unexplained", "checks neither equal nor explained: " + "; ".join(unexplained), "explain each, or fix and reload")
if unexplained else met("zero_unexplained")]
@GL.use_case(kind="command", route="POST /migration/odoo/sign", permission=MIGRATION_SIGN)
def sign_gl_reconciliation(caller: dict, run_id: Annotated[str, Field(max_length=200)]) -> dict:
"""The OWNER signs the reconciliation: every check equal, or explained where explaining is allowed (the trial balance per account and
period must be equal and cannot be explained away). Records who, when and the run's checksums. From then on the run is frozen, and
postings from the composed modules are accepted from the day after the cut-over; a later run only adds what Odoo posted after it."""
with repository() as repo, decide():
r = _get_run(repo, run_id)
if r["status"] == "signed":
return _run_view(r)
if r["status"] != "reconciled":
raise RuleBroken(f"the run is {r['status']}: reconcile it first", status=409)
checks = json.loads(r["checks_json"])
require("sign the reconciliation", sign_gates(checks), status=409)
checksums = hashlib.sha256(json.dumps([[c["check"], c["odoo"], c["ours"], c["explanation"]] for c in checks], sort_keys=True, default=str).encode()).hexdigest()
n = repo.conn.execute(select(func.count()).select_from(t.gl_journal).where(t.gl_journal.c.migration_run_id == run_id)).scalar()
RUN(repo).set((run_id,), status="signed", signed_by=who(caller), signed_at=utc(real_now()), checksums=f"{checksums}:{n}")
return _run_view(_get_run(repo, run_id))
@GL.use_case(kind="command", route="POST /migration/odoo/discard", permission=MIGRATION_SIGN, destructive=True)
def discard_gl_migration(caller: dict, run_id: Annotated[str, Field(max_length=200)], reason: Reason) -> dict:
"""The owner discards an UNSIGNED run, with a reason: its journals, lines and split lines are removed (nothing was posted on them) and
the id-map rows of its journals, so a corrected load starts clean. A signed run is never discarded."""
with repository() as repo, decide():
r = _get_run(repo, run_id)
if r["status"] == "signed":
raise refusal("not_signed", "a signed run is never discarded", "it is frozen history", status=409)
if r["status"] == "discarded":
return _run_view(r)
ids = [x[0] for x in repo.conn.execute(select(t.gl_journal.c.journal_id).where(t.gl_journal.c.migration_run_id == run_id))]
later = repo.conn.execute(select(func.count()).select_from(t.gl_journal).where(
t.gl_journal.c.reverses_journal_id.in_(ids), t.gl_journal.c.migration_run_id.is_(None))).scalar() if ids else 0
if later:
raise RuleBroken("a journal of this run was reversed by a posting: it cannot be discarded", status=409)
for jid in ids:
for table in (t.gl_split_line, t.gl_journal_line):
repo.conn.execute(table.delete().where(table.c.journal_id == jid))
repo.conn.execute(t.gl_chain_claim.delete().where(t.gl_chain_claim.c.journal_id == jid))
repo.conn.execute(t.gl_number_series.delete().where(t.gl_number_series.c.journal_id == jid))
repo.conn.execute(t.gl_journal.delete().where(t.gl_journal.c.journal_id == jid))
repo.conn.execute(t.gl_migration_map.delete().where(t.gl_migration_map.c.run_id == run_id, t.gl_migration_map.c.odoo_model == "account.move"))
RUN(repo).set((run_id,), status="discarded", checks_json="[]")
return {**_run_view(_get_run(repo, run_id)), "reason": reason, "journals_removed": len(ids)}
@GL.use_case(kind="query", route="GET /migration/odoo/runs", permission=MIGRATION_READ)
def list_gl_migration_runs(caller: dict) -> dict:
"""The migration runs with their status."""
with repository() as repo:
return {"runs": [{k: r[k] for k in ("run_id", "odoo_db", "cutover_date", "status", "signed_by", "signed_at")}
for r in (dict(x._mapping) for x in repo.conn.execute(t.gl_migration_run.select().order_by(t.gl_migration_run.c.run_id)))]}
@GL.use_case(kind="query", route="GET /migration/odoo/runs/{run_id}", permission=MIGRATION_READ)
def get_gl_migration_run(caller: dict, run_id: Annotated[str, Field(max_length=200)]) -> dict:
"""One run: the doors, what it created and skipped, every check as {check, odoo, ours, equal, must, explanation}, and the sign-off
readiness."""
with repository() as repo:
r = _get_run(repo, run_id)
v = _run_view(r)
v["readiness"] = [g.as_dict() for g in sign_gates(v["checks"])] if v["checks"] else []
return v
@GL.use_case(kind="query", route="GET /migration/status", permission=MIGRATION_READ)
def get_gl_migration_status(caller: dict, db: Annotated[str, Field(min_length=1, max_length=100)]) -> dict:
"""Where the migration of one Odoo database stands, for a loader that reads migrated gl history (consolidation CON-34): the
latest run of that database - {db, run_id, status (none | loaded | reconciled | signed), cutover_date, reconciled_at, signed_by,
signed_at}. No run (or only a discarded one) is status none with null fields, never a 404. Only status signed may be read."""
with repository() as repo:
rows = [dict(x._mapping) for x in repo.conn.execute(t.gl_migration_run.select().where(
and_(t.gl_migration_run.c.odoo_db == db, t.gl_migration_run.c.status != "discarded")).order_by(t.gl_migration_run.c.loaded_at))]
if not rows:
return {"db": db, "run_id": None, "status": "none", "cutover_date": None, "reconciled_at": None, "signed_by": None,
"signed_at": None}
r = rows[-1]
return {"db": db, "run_id": r["run_id"], "status": r["status"], "cutover_date": r["cutover_date"],
"reconciled_at": r["reconciled_at"], "signed_by": r["signed_by"], "signed_at": r["signed_at"]}
@GL.use_case(kind="query", route="GET /migration-map/{db}/{model}", permission=MIGRATION_READ)
def get_gl_migration_map(caller: dict, db: Annotated[str, Field(max_length=100)], model: Annotated[str, Field(max_length=100)],
ids: Annotated[str, Field(max_length=8000)]) -> list:
"""The one map route (shared.yaml conventions.odoo_migration.map_route): for Odoo ids (comma-separated) of a model (account.account,
account.journal, res.company ...), [{odoo_model, odoo_id, ours}] where ours is this brick's natural key (an account_id, a book, a
company_code); an id not migrated is omitted - never null, never an error."""
want = [i.strip() for i in ids.split(",") if i.strip()] # never truncated: the 8000-character bound on `ids` refuses (422) what is too long
with repository() as repo:
mapped: dict = {}
col = t.gl_migration_map.c
for at in range(0, len(want), 500): # one IN query per 500 ids, however many are asked
for r in MAP(repo).rows(col.odoo_db == db, col.odoo_model == model, col.odoo_id.in_(want[at:at + 500])):
mapped[r["odoo_id"]] = r["natural_key"]
return [{"odoo_model": model, "odoo_id": i, "ours": mapped[i]} for i in want if i in mapped]
_ = (BALANCE_SHEET, carry, Gate)