dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/migration.py
"""The Odoo migration slice (PUR-33, PUR-34): history is carried as Odoo stored it, keyed odoo:<db>:<model>:<id> and recorded in pur_migration_map,
so a second load changes nothing; a SIGNED run is frozen. Migrated receipts are NOT handed to inventory and migrated bills are NOT posted (the
ledger migrates its journals). Carried here: purchase orders and lines, receipts and returns (received quantity), vendor bills and refunds (with
what was paid, from amount_residual), supplier payments. The reconcile checks compare what the reader returned with what we hold.
Not carried yet (listed in the report): supplier prices, bank details, blanket orders, the per-tax summary and the series high-water marks."""
import json
from datetime import date, datetime
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import IsoDate, text
from brick_purchasing.application import base, bills, rules
from brick_purchasing.application.base import (
MIG_LOAD,
MIG_READ,
MIG_RECONCILE,
MIG_SIGN,
PURCHASING,
_decide,
_sub,
_who,
unit,
)
from brick_purchasing.application.inputs import Ref
from brick_purchasing.domain import bill as bl
from brick_purchasing.domain import order as od
from brick_purchasing.domain import payment as pm
from brick_purchasing.domain import receipt as rc
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.values import ZERO
from brick_purchasing.infrastructure.seams import ports
repository = lambda: base.repository() # noqa: E731
STATUS = {"purchase": "confirmed", "done": "closed", "cancel": "cancelled"}
def _d(v) -> Optional[date]:
if not v:
return None
return v if isinstance(v, date) and not isinstance(v, datetime) else date.fromisoformat(str(v)[:10])
def _m(v) -> Decimal:
return Decimal(str(v or 0))
def _key(db: str, model: str, oid) -> str:
return f"odoo:{db}:{model}:{oid}"
def _run_row(repo, run_id: str) -> Optional[dict]:
return repo.migration_runs.get(run_id)
def _put_map(repo, db: str, model: str, oid: int, natural: str, run_id: str) -> bool:
"""True when this row is new (not mapped before)."""
if repo.migration_map.get(db, model, int(oid)) is not None:
return False
repo.migration_map.put({"odoo_db": db, "odoo_model": model, "odoo_id": int(oid), "natural_key": natural, "run_id": run_id, "migrated_at": base._stamp()})
return True
def _raise_series(repo, named: dict, cut: date) -> None:
"""The highest migrated number per year raises number_series.high_water_mark, so the next number continues the carried series."""
import re
for kind, fmt_key in (("order", "order_number_format"), ("bill", "bill_number_format")):
fmt = repo.settings.value_on(fmt_key, base._today(repo)) # the format the next number is rendered in
pat = re.compile("^" + re.sub(r"\\\{seq(?::\d+)?\\\}", r"(?P<seq>\\d+)", re.sub(r"\\\{yyyy\\\}", r"(?P<yyyy>\\d{4})", re.escape(fmt))) + "$")
top: dict = {}
for name, day in named[kind]:
m = pat.match(name or "")
if not m or "seq" not in pat.groupindex:
continue
year = int(m.group("yyyy")) if "yyyy" in pat.groupindex else (day.year if day else cut.year)
top[year] = max(top.get(year, 0), int(m.group("seq")))
for year, mark in top.items():
sr, ver = repo.numbers.get(f"{kind}|{fmt}", year)
sr.format = fmt
sr.raise_to(mark)
repo.numbers.save(sr, ver)
@PURCHASING.use_case(kind="command", route="POST /migration/odoo/load", permission=MIG_LOAD, idempotent=True)
def load_odoo(caller: dict, cutover_date: IsoDate) -> dict:
"""Load the Odoo history on or before the cutover date through the reader port. Idempotent: every record is keyed odoo:<db>:<model>:<id>, so a
second load creates nothing; a signed run is never rewritten. Unmapped partners, products or units are listed (skipped), never guessed."""
cut = base._day(cutover_date, "cutover_date")
def run():
with repository() as repo, _decide():
odoo, maps = ports.odoo, ports.maps
missing = [n for n, port in (("an Odoo reader", odoo), ("business-partner/product maps", maps)) if port is None]
if missing:
raise Refused(409, json.dumps({"gate": "migration_adapters_installed", "met": False,
"reason": "not installed: " + " and ".join(missing) + " (the history is read through them)",
"fix": "install the Odoo reader and the migration maps adapters (ports.odoo, ports.maps), then load again"}))
named = {"order": [], "bill": []}
db = odoo.db
run_id = f"odoo:{db}:{cut.isoformat()}"
have = _run_row(repo, run_id)
if have and have["status"] == "signed":
raise RuleBroken("signed: this run is signed and frozen; history is never rewritten")
if have is None:
repo.migration_runs.put({"run_id": run_id, "odoo_db": db, "cutover_date": cut, "status": "loaded", "doors": "{}", "checksums": "{}", "version": 1})
created, skipped = {"orders": 0, "receipts": 0, "bills": 0, "payments": 0}, []
lines_of: dict = {}
for ln in sorted([x for x in odoo.read("purchase.order.line") if not x.get("display_type")], key=lambda x: (x.get("sequence") or 0, x["id"])):
lines_of.setdefault(ln["order_id"], []).append(ln)
line_no_of = {}
for o in odoo.read("purchase.order"):
if o.get("state") not in STATUS or _d(o.get("date_order")) is None or _d(o["date_order"]) > cut:
continue
bp = maps.partner(o["partner_id"])
if bp is None:
skipped.append({"model": "purchase.order", "id": o["id"], "reason": f"partner {o['partner_id']} is not in business-partner's map"})
continue
pid = _key(db, "purchase.order", o["id"])
for n, ln in enumerate(lines_of.get(o["id"], []), 1):
line_no_of[ln["id"]] = (pid, n)
if not _put_map(repo, db, "purchase.order", o["id"], pid, run_id):
continue
po = od.PurchaseOrder(pid, o["name"], 1, bp, o.get("currency_id") or "GBP", STATUS[o["state"]], source_kind="odoo", source_id=str(o["id"]),
supplier_ref=o.get("partner_ref") or "", confirmed_on=_d(o.get("date_approve")), net_total=_m(o.get("amount_untaxed")))
for n, ln in enumerate(lines_of.get(o["id"], []), 1):
product, unit_code = maps.product(ln["product_id"]) if ln.get("product_id") else "", maps.unit(ln["product_uom"])
if unit_code is None:
raise RuleBroken(f"unmapped: line {ln['id']} uses unit {ln['product_uom']}, which product's map does not know")
po.lines.append(od.OrderLine(pid, n, product or "", ln.get("name") or "", _m(ln["product_qty"]), unit_code, 1, 1, _m(ln["product_qty"]), unit_code, unit_code, 1, 1,
"unit", 1, _m(ln["price_unit"]), _m(ln["product_qty"]), _m(ln["price_subtotal"])))
repo.orders.save(po, 0)
repo.numbers.claim_number(o["name"], "migrated|order", pid)
named["order"].append((o["name"], _d(o.get("date_order"))))
created["orders"] += 1
pickings = {p["id"]: p for p in odoo.read("stock.picking") if p.get("state") == "done" and p.get("purchase_id") and _d(p.get("date_done")) and _d(p["date_done"]) <= cut}
moves_of: dict = {}
for mv in odoo.read("stock.move"):
if mv.get("state") == "done" and mv.get("picking_id") in pickings and mv.get("purchase_line_id") in line_no_of:
moves_of.setdefault(mv["picking_id"], []).append(mv)
for pk_id, p in pickings.items():
po_key = _key(db, "purchase.order", p["purchase_id"])
got = repo.orders.get(po_key)
if not got or not moves_of.get(pk_id):
continue
rid = _key(db, "stock.picking", pk_id)
if not _put_map(repo, db, "stock.picking", pk_id, rid, run_id):
continue
is_return = p.get("picking_type_code") == "outgoing"
r = rc.GoodsReceipt(rid, "return" if is_return else "receipt", po_key, p.get("name") or "", _d(p["date_done"]), "odoo migration", "odoo migration",
p.get("location_dest_id") and str(p["location_dest_id"]) or "", "migrated return" if is_return else "", "", "handed_over", got[0].bp_id)
for n, mv in enumerate(moves_of[pk_id], 1):
_po, ol = line_no_of[mv["purchase_line_id"]]
q = _m(mv.get("quantity", mv.get("quantity_done", mv.get("product_uom_qty")))) * (-1 if is_return else 1)
r.lines.append(rc.ReceiptLine(rid, n, ol, "", q, "", q, "", "", "", _d(p["date_done"]) and datetime.combine(_d(p["date_done"]), datetime.min.time()),
"migrated: inventory migrates its own moves"))
repo.receipts.save(r, 0)
created["receipts"] += 1
moves = [m for m in odoo.read("account.move") if m.get("move_type") in ("in_invoice", "in_refund") and m.get("state") in ("posted", "cancel")
and _d(m.get("invoice_date") or m.get("date")) and _d(m.get("invoice_date") or m.get("date")) <= cut]
mlines: dict = {}
for ml in odoo.read("account.move.line"):
if ml.get("display_type") in (None, False, "product"):
mlines.setdefault(ml["move_id"], []).append(ml)
for m in moves:
bp = maps.partner(m.get("commercial_partner_id") or m["partner_id"])
if bp is None:
skipped.append({"model": "account.move", "id": m["id"], "reason": f"partner {m['partner_id']} is not in business-partner's map"})
continue
bid = _key(db, "account.move", m["id"])
if not _put_map(repo, db, "account.move", m["id"], bid, run_id):
continue
kind = "credit" if m["move_type"] == "in_refund" else "bill"
idate = _d(m.get("invoice_date") or m.get("date"))
prod = [ml for ml in mlines.get(m["id"], []) if ml.get("product_id") or ml.get("purchase_line_id") or ml.get("price_subtotal")]
po_key = next((line_no_of[ml["purchase_line_id"]][0] for ml in prod if ml.get("purchase_line_id") in line_no_of), "")
b = bl.SupplierBill(bid, bp, (m.get("ref") or m["name"]), idate, kind, False, _key(db, "account.move", m["reversed_entry_id"]) if m.get("reversed_entry_id") else "", po_key,
"", m.get("currency_id") or "GBP", _m(m["amount_untaxed"]), _m(m["amount_tax"]), _m(m["amount_total"]), None, _d(m.get("invoice_date_due")), "",
"approved" if m["state"] == "posted" else "cancelled", m["name"], "odoo migration", "odoo migration", cut, "odoo migration", "odoo migration",
idate if m["state"] == "posted" else None, "odoo", str(m["id"]))
for n, ml in enumerate(prod, 1):
link = line_no_of.get(ml.get("purchase_line_id"))
unit_code = maps.unit(ml["product_uom_id"]) if ml.get("product_uom_id") else ""
b.lines.append(bl.BillLine(bid, n, link[1] if link else None, maps.product(ml["product_id"]) if ml.get("product_id") else "", ml.get("name") or "",
_m(ml["quantity"]), unit_code or "", 1, 1, _m(ml["quantity"]), unit_code or "", 1, 1, 1, _m(ml["price_unit"]),
_m(ml["price_subtotal"]), "", "", ZERO, _m(ml.get("price_total")) - _m(ml.get("price_subtotal")), ZERO))
if link and m["state"] == "posted":
b.claims.append(bl.BilledClaim(link[0], link[1], bid, _m(ml["quantity"]), None))
repo.bills.save(b, 0)
repo.numbers.claim_number(m["name"], "migrated|bill", bid)
named["bill"].append((m["name"], idate))
paid = _m(m["amount_total"]) - _m(m.get("amount_residual"))
if m["state"] == "posted" and paid > 0:
repo.allocations.put(pm.Allocation("odoo", _key(db, "residual", m["id"]), bid, 1, paid, cut, "odoo migration"))
created["bills"] += 1
for p in odoo.read("account.payment"):
if p.get("partner_type") != "supplier" or _d(p.get("date")) is None or _d(p["date"]) > cut:
continue
bp = maps.partner(p["partner_id"])
if bp is None:
skipped.append({"model": "account.payment", "id": p["id"], "reason": f"partner {p['partner_id']} is not in business-partner's map"})
continue
payid = _key(db, "account.payment", p["id"])
if not _put_map(repo, db, "account.payment", p["id"], payid, run_id):
continue
amt = _m(p["amount"]) * (-1 if p.get("payment_type") == "inbound" else 1)
repo.payments.save(pm.SupplierPayment(payid, "", bp, p.get("currency_id") or "GBP", amt, amt, paid_on=_d(p["date"]), status="paid", reference=p.get("name") or "",
source_kind="odoo"), 0)
created["payments"] += 1
_raise_series(repo, named, cut)
return {"run_id": run_id, "created": created, "skipped": skipped, "replay": not any(created.values())}
return unit(None, run)
def _checks(repo, odoo, cut: date, run_id: str) -> list:
out = []
def add(name: str, theirs, ours, explanation: str = "") -> None:
out.append({"check": name, "odoo": json.dumps(theirs, sort_keys=True, default=str), "ours": json.dumps(ours, sort_keys=True, default=str),
"equal": theirs == ours, "explanation": ""})
maps = {m["odoo_id"] for m in repo.migration_map.all(run_id=run_id, odoo_model="purchase.order")}
po = [o for o in odoo.read("purchase.order") if o["id"] in maps]
theirs, ours = {}, {}
for o in po:
k = f"{STATUS[o['state']]}:{str(o['date_order'])[:4]}"
theirs[k] = theirs.get(k, 0) + 1
for h in repo.orders.repo.all_heads():
if h.source_kind == "odoo" and int(h.source_id) in maps:
k = f"{h.status}:{h.confirmed_on.year if h.confirmed_on else ''}"
ours[k] = ours.get(k, 0) + 1
theirs = {k: v for k, v in theirs.items()}
add("order count by state and year", {k.split(":")[0] + ":" + (k.split(":")[1]): v for k, v in theirs.items()}, ours)
recv_t, recv_o, bill_t, bill_o = {}, {}, {}, {}
for o in po:
for n, ln in enumerate(sorted([x for x in odoo.read("purchase.order.line") if x["order_id"] == o["id"] and not x.get("display_type")], key=lambda x: (x.get("sequence") or 0, x["id"])), 1):
key = f"{_key(odoo.db, 'purchase.order', o['id'])}:{n}"
recv_t[key], bill_t[key] = str(_m(ln.get("qty_received"))), str(_m(ln.get("qty_invoiced")))
for h in repo.orders.repo.all_heads():
if h.source_kind == "odoo" and int(h.source_id) in maps:
rcv, bld = rules.received_by_line(repo, h.purchase_order_id), rules.billed_by_line(repo, h.purchase_order_id)
for ln in repo.orders.get(h.purchase_order_id)[0].lines:
recv_o[f"{h.purchase_order_id}:{ln.line_no}"] = str(Decimal(rcv.get(ln.line_no, Fraction(0)).numerator) / Decimal(rcv.get(ln.line_no, Fraction(1)).denominator))
bill_o[f"{h.purchase_order_id}:{ln.line_no}"] = str(Decimal(bld.get(ln.line_no, Fraction(0)).numerator) / Decimal(bld.get(ln.line_no, Fraction(1)).denominator))
norm = lambda d: {k: str(Decimal(v).normalize()) for k, v in d.items()} # noqa: E731
add("received quantity per order line", norm(recv_t), norm(recv_o), "equal, or explained where Odoo's qty_received was set by hand")
add("billed quantity per order line", norm(bill_t), norm(bill_o))
bmaps = {m["odoo_id"] for m in repo.migration_map.all(run_id=run_id, odoo_model="account.move")}
mv = [m for m in odoo.read("account.move") if m["id"] in bmaps]
bt, bo = {}, {}
for m in mv:
k = f"{'credit' if m['move_type'] == 'in_refund' else 'bill'}:{m['currency_id']}:{str(m.get('invoice_date') or m['date'])[:4]}"
c = bt.setdefault(k, [0, ZERO, ZERO, ZERO])
c[0] += 1
c[1], c[2], c[3] = c[1] + _m(m["amount_untaxed"]), c[2] + _m(m["amount_tax"]), c[3] + _m(m["amount_total"])
for m in mv:
b = repo.bills.get(_key(odoo.db, "account.move", m["id"]))[0]
k = f"{b.kind}:{b.currency}:{b.invoice_date.year}"
c = bo.setdefault(k, [0, ZERO, ZERO, ZERO])
c[0] += 1
c[1], c[2], c[3] = c[1] + b.subtotal, c[2] + b.tax_total, c[3] + b.total
add("bill count and totals by kind, currency and year", {k: [v[0]] + [str(x) for x in v[1:]] for k, v in bt.items()}, {k: [v[0]] + [str(x) for x in v[1:]] for k, v in bo.items()})
add("every bill number carried once per supplier", sorted(f"{ports.maps.partner(m.get('commercial_partner_id') or m['partner_id'])}|{m.get('ref') or m['name']}|{str(m.get('invoice_date') or m['date'])[:10]}" for m in mv),
sorted(f"{ports.maps.partner(m.get('commercial_partner_id') or m['partner_id'])}|{repo.bills.get(_key(odoo.db, 'account.move', m['id']))[0].supplier_invoice_no}|"
f"{repo.bills.get(_key(odoo.db, 'account.move', m['id']))[0].invoice_date}" for m in mv), "equal, or explained for bills that shared a reference")
open_t: dict = {}
for m in mv:
if m["state"] == "posted" and m.get("payment_state") in ("not_paid", "partial", "in_payment"):
k = f"{ports.maps.partner(m.get('commercial_partner_id') or m['partner_id'])}:{m['currency_id']}"
open_t[k] = open_t.get(k, ZERO) + _m(m["amount_residual"]) * (-1 if m["move_type"] == "in_refund" else 1)
open_o: dict = {}
for i in bills.open_items(repo, cut, None):
if i["bill_id"].startswith(f"odoo:{odoo.db}:"):
k = f"{i['bp_id']}:{i['currency']}"
open_o[k] = open_o.get(k, ZERO) + Decimal(str(i["outstanding"]))
add("open payables per supplier and currency at cutover", {k: str(v) for k, v in open_t.items() if v}, {k: str(v) for k, v in open_o.items() if v})
pt, po_ = {}, {}
for p in odoo.read("account.payment"):
if p["id"] in {m["odoo_id"] for m in repo.migration_map.all(run_id=run_id, odoo_model="account.payment")}:
k = f"{p['currency_id']}:{str(p['date'])[:4]}"
pt[k] = pt.get(k, ZERO) + _m(p["amount"]) * (-1 if p["payment_type"] == "inbound" else 1)
for m in repo.migration_map.all(run_id=run_id, odoo_model="account.payment"):
q = repo.payments.get(m["natural_key"])[0]
k = f"{q.currency}:{q.paid_on.year}"
po_[k] = po_.get(k, ZERO) + q.amount
add("supplier payments by currency and year", {k: str(v) for k, v in pt.items()}, {k: str(v) for k, v in po_.items()})
return out
@PURCHASING.use_case(kind="command", route="POST /migration/odoo/{run_id}/reconcile", permission=MIG_RECONCILE, idempotent=True)
def reconcile_purchasing_migration(caller: dict, run_id: Ref) -> dict:
"""Compare what Odoo holds with what this run carried: each check answers {check, odoo, ours, equal, explanation}. The results are stored on the
run; a signed run's results are returned as signed."""
with repository() as repo, _decide():
run = _run_row(repo, run_id)
if run is None:
raise Refused(404, f"no migration run {run_id}")
checks = _checks(repo, ports.odoo, run["cutover_date"], run_id) if run["status"] != "signed" else [
{k: c[k] for k in ("check", "odoo", "ours", "equal", "explanation")} for c in sorted(repo.migration_checks.all(run_id=run_id), key=lambda c: c["check_no"])]
if run["status"] != "signed":
for n, c in enumerate(checks, 1):
prior = repo.migration_checks.get(run_id, n)
c["explanation"] = (prior or {}).get("explanation") or "" if not c["equal"] else ""
repo.migration_checks.put({"run_id": run_id, "check_no": n, "name": c["check"], "odoo": c["odoo"], "ours": c["ours"], "equal": c["equal"], "must": "equal",
"explanation": c["explanation"], "explained_by_email": (prior or {}).get("explained_by_email") or ""})
repo.migration_runs.put({**run, "status": "reconciled", "reconciled_at": base._stamp()})
return {"run_id": run_id, "status": "signed" if run["status"] == "signed" else "reconciled", "checks": checks, "all_equal": all(c["equal"] for c in checks)}
@PURCHASING.use_case(kind="command", route="POST /migration/odoo/{run_id}/explain", permission=MIG_RECONCILE, idempotent=True)
def explain_difference(caller: dict, run_id: Ref, check_no: int, explanation: text(1000, min_length=3)) -> dict:
"""Explain a check that differs (for example Odoo's received quantity was set by hand), recorded with who; a run with an unexplained difference cannot be signed."""
with repository() as repo, _decide():
run, row = _run_row(repo, run_id), repo.migration_checks.get(run_id, int(check_no))
if run is None or row is None:
raise Refused(404, f"no check {check_no} on run {run_id}")
if run["status"] == "signed":
raise RuleBroken("signed: this run is signed and frozen")
if row["equal"]:
raise RuleBroken("this check is equal: there is nothing to explain")
repo.migration_checks.put({**row, "explanation": explanation, "explained_by_email": _who(caller)})
return {"run_id": run_id, "check_no": check_no, "check": row["name"], "explanation": explanation}
@PURCHASING.use_case(kind="command", route="POST /migration/odoo/{run_id}/sign", permission=MIG_SIGN, idempotent=True)
def sign_purchasing_migration(caller: dict, run_id: Ref) -> dict:
"""The owner signs a reconciled run: who (email and sub), when and its checksums. Refused while a check differs without an explanation. A signed
run is frozen."""
with repository() as repo, _decide():
run = _run_row(repo, run_id)
if run is None:
raise Refused(404, f"no migration run {run_id}")
if run["status"] == "signed":
return {"run_id": run_id, "status": "signed", "signed_by_email": run["signed_by_email"]}
if run["status"] != "reconciled":
raise RuleBroken("reconcile: reconcile the run before signing it")
bad = [c["name"] for c in repo.migration_checks.all(run_id=run_id) if not c["equal"] and not (c.get("explanation") or "").strip()]
if bad:
raise RuleBroken("differs: " + ", ".join(bad) + "; fix: correct the source or explain the difference")
sums = json.dumps({c["name"]: [c["odoo"], c["ours"]] for c in repo.migration_checks.all(run_id=run_id)}, sort_keys=True)
repo.migration_runs.put({**run, "status": "signed", "signed_by_email": _who(caller), "signed_by_sub": _sub(caller), "signed_at": base._stamp(), "checksums": sums})
return {"run_id": run_id, "status": "signed", "signed_by_email": _who(caller)}
@PURCHASING.use_case(kind="query", route="GET /migration-map", permission=MIG_READ)
def migration_map(caller: dict, odoo_model: Ref, odoo_id: int) -> dict:
"""What an Odoo record became here (other bricks' migrations resolve purchasing ids through this, never by names)."""
with repository() as repo:
row = repo.migration_map.get(ports.odoo.db, odoo_model, int(odoo_id))
if row is None:
raise Refused(404, f"{odoo_model} {odoo_id} was not migrated")
return {"odoo_model": odoo_model, "odoo_id": odoo_id, "natural_key": row["natural_key"], "run_id": row["run_id"]}