dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/odoo_migration.py
"""The one-off Odoo migration of stock (INV-26, milestone M8): warehouses and locations, the quant snapshot as OPENING moves
valued at Odoo's cost, lots, and the done moves as read-only history; then the reconciliation the customer signs.
Decisions it carries: the opening move is dated the cut-over (the one back-dated write); quantities are converted from the
product's Odoo uom to its base unit by the fraction product's id map returns; reserved quantities are NOT carried (re-made from
the migrated orders); a quant owned by a supplier comes over as that supplier's consignment stock at zero value; negative quants
stop the run unless the door says `carry_negative`. The run is idempotent (every Odoo row has a map row, a second run writes
nothing) and a SIGNED run is frozen: it refuses any further import for that cut-over."""
import json
import re
from datetime import datetime
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import qstr
from brick_inventory.application.inputs import Code, day
from brick_inventory.application.warehouses import _virtuals
from brick_inventory.domain import location as L
from brick_inventory.domain.errors import BadInput, GateBroken
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import NO_LOT, OURS, STOCK_TYPES, UNRESTRICTED, ZERO, money
from brick_inventory.infrastructure import odoo, seams
IMPORT, READ = "inv:migration:import", "inv:migration:read"
def loc_id(complete_name: str) -> str:
"""WH/Stock/Shelf 1 -> WH/STOCK/SHELF-1: path segments kept, upper case, spaces to hyphens."""
return "/".join(re.sub(r"\s+", "-", seg.strip()).upper() for seg in complete_name.split("/") if seg.strip())
def _read(rd, model: str, fields: list, domain=None) -> list:
try:
return rd.read(model, fields, domain)
except seams.Missing as m:
raise GateBroken("odoo_reader", m.text, "check the Odoo connection (ODOO_URL, ODOO_DB, ODOO_USER, ODOO_PASSWORD)", 409) from None
class Scope:
"""Everything one run reads, read once: the plan reads and maps and writes nothing."""
def __init__(self, db: str, warehouse_codes: Optional[list]):
rd = odoo.reader()
self.maps = odoo.maps(db)
whs = _read(rd, "stock.warehouse", ["id", "name", "code", "company_id", "active"])
self.warehouses = [w for w in whs if w.get("active", True) and (not warehouse_codes or w["code"] in warehouse_codes)]
self.locations = {l["id"]: l for l in _read(rd, "stock.location", ["id", "name", "complete_name", "location_id", "usage", "warehouse_id", "active"])}
self.quants = [q for q in _read(rd, "stock.quant", ["id", "product_id", "location_id", "lot_id", "quantity", "owner_id", "company_id"])
if q["quantity"] != 0 and (self.locations.get(q["location_id"]) or {}).get("usage") in ("internal", "transit")]
self.lots = {l["id"]: l for l in _read(rd, "stock.lot", ["id", "name", "product_id", "expiration_date"])}
self.cost = {p["id"]: Decimal(str(p["standard_price"])) for p in _read(rd, "product.product", ["id", "default_code", "standard_price"])}
self.layers = _read(rd, "stock.valuation.layer", ["product_id", "company_id", "quantity", "value"])
self.history = [m for m in _read(rd, "stock.move.line", ["id", "move_id", "product_id", "lot_id", "quantity", "location_id", "location_dest_id", "date"])]
self.moves = {m["id"]: m for m in _read(rd, "stock.move", ["id", "reference", "state", "origin"])}
def _wh_of(sc: Scope, location_id: int) -> str:
loc = sc.locations[location_id]
wid = loc.get("warehouse_id")
for w in sc.warehouses:
if w["id"] == wid:
return w["code"]
return loc_id(loc["complete_name"]).split("/")[0]
def _product(sc: Scope, odoo_id: int) -> dict:
p = sc.maps.product(odoo_id)
if not p:
raise GateBroken("product_map", f"Odoo product {odoo_id} has no product_id in product's id map", "run product's migration first", 409)
return p
def _run_row(repo, run_id: str):
return repo.get("inv_migration_run", run_id=run_id)
@INV.use_case(kind="command", route="POST /migration/odoo/import", permission=IMPORT, idempotent=True)
def import_odoo_stock(caller: dict, odoo_db: Code, cutover: str, company_code: Code, warehouse_codes: Optional[list[Code]] = None,
carry_negative: bool = False, history: bool = True) -> dict:
"""Read the customer's Odoo and write its stock: warehouses and locations, one OPENING move per quant dated the cut-over at
Odoo's cost, lots, and the done move lines as read-only history. Idempotent: a second run creates nothing. A signed run is
frozen."""
cut = day(cutover)
run_id = f"odoo:{odoo_db}:{cut.isoformat()}"
@U.decide
def body(unit):
run = _run_row(unit.repo, run_id)
if run and run["status"] == "signed":
raise GateBroken("run_signed", f"run {run_id} was signed by {run['signed_by']}: it is frozen", "start a new run for a later cut-over", 409)
sc = Scope(odoo_db, warehouse_codes)
negatives = [q for q in sc.quants if q["quantity"] < 0]
if negatives and not carry_negative:
raise GateBroken("negative_quants", f"{len(negatives)} negative quants (first: quant {negatives[0]['id']}): correct them in Odoo or carry them",
"set carry_negative, or fix Odoo first", 409)
created = {"warehouses": 0, "locations": 0, "opening_moves": 0, "lots": 0, "history": 0}
known = {(m["odoo_model"], m["odoo_id"]) for m in unit.repo.find("inv_migration_map", odoo_db=odoo_db)}
def mapped(model: str, odoo_id: int, key: str) -> None:
unit.put("inv_migration_map", {"odoo_db": odoo_db, "odoo_model": model, "odoo_id": odoo_id, "our_key": key, "run_id": run_id},
("odoo_db", "odoo_model", "odoo_id"))
_virtuals(unit, company_code)
for w in sc.warehouses:
if ("stock.warehouse", w["id"]) in known:
continue
if unit.repo.get("inv_warehouse", warehouse_id=w["code"]) is None:
unit.put("inv_warehouse", {"warehouse_id": w["code"], "name": w["name"], "company_code": company_code, "address": "", "status": "active"}, "warehouse_id")
unit.put("inv_location", {"location_id": w["code"], "warehouse_id": w["code"], "company_code": company_code, "parent_location_id": "",
"kind": "view", "name": w["name"], "stock_status": "unrestricted", "allowed_stock_types": ",".join(STOCK_TYPES),
"status": "active"}, "location_id")
created["warehouses"] += 1
mapped("stock.warehouse", w["id"], w["code"])
seen: dict = {}
wanted = {q["location_id"] for q in sc.quants}
for lid, l in sorted(sc.locations.items()):
if l["usage"] not in ("internal", "view") or not l.get("active", True) or (l["usage"] == "view" and lid not in wanted and not l.get("warehouse_id")):
continue
our = loc_id(l["complete_name"])
if our in seen and seen[our] != lid:
raise GateBroken("location_collision", f"Odoo locations {seen[our]} and {lid} both become {our}", "rename one in Odoo", 409)
seen[our] = lid
if ("stock.location", lid) in known or unit.repo.get("inv_location", location_id=our):
if ("stock.location", lid) not in known:
mapped("stock.location", lid, our)
continue
wh = _wh_of(sc, lid)
parent = "/".join(our.split("/")[:-1]) or wh
unit.put("inv_location", {"location_id": our, "warehouse_id": wh, "company_code": company_code, "parent_location_id": parent,
"kind": "internal" if l["usage"] == "internal" else "view", "name": l["name"], "stock_status": UNRESTRICTED,
"allowed_stock_types": ",".join(STOCK_TYPES), "status": "active"}, "location_id")
mapped("stock.location", lid, our)
created["locations"] += 1
for lot in sc.lots.values():
if ("stock.lot", lot["id"]) in known:
continue
p = _product(sc, lot["product_id"])
exp = day(str(lot["expiration_date"])[:10]) if lot.get("expiration_date") else None
unit.put("inv_lot", {"product_id": p["product_id"], "lot_code": lot["name"], "kind": "lot", "expires_on": exp, "received_on": cut,
"blocked": False, "supplier_bp_id": None}, ("product_id", "lot_code"))
mapped("stock.lot", lot["id"], f"{p['product_id']}:{lot['name']}")
created["lots"] += 1
for q in sorted(sc.quants, key=lambda x: x["id"]):
if ("stock.quant", q["id"]) in known:
continue
p = _product(sc, q["product_id"])
num, den = int(p.get("uom_numerator") or 1), int(p.get("uom_denominator") or 1)
qty_odoo = Decimal(str(q["quantity"]))
base = qty_odoo * num / den
usage = sc.locations[q["location_id"]]["usage"]
wh = _wh_of(sc, q["location_id"])
where = L.virtual_id(company_code, "TRANSIT") if usage == "transit" else loc_id(sc.locations[q["location_id"]]["complete_name"])
lot = sc.lots[q["lot_id"]]["name"] if q.get("lot_id") else NO_LOT
owner = OURS
if q.get("owner_id"):
bp = sc.maps.partner(q["owner_id"])
if not bp:
raise GateBroken("quant_owner", f"quant {q['id']} is owned by Odoo partner {q['owner_id']}, who has no bp_id", "migrate business-partner first", 409)
owner = str(bp)
value = money(qty_odoo * sc.cost.get(q["product_id"], ZERO)) if owner == OURS else ZERO
doc_id = f"opening:{odoo_db}:{wh}"
if unit.repo.get("inv_document", document_id=doc_id) is None and not any(d == doc_id for d in unit._opened):
docs.stage_document(unit, doc_id, "opening", wh, company_code, lifecycle="done", done_on=cut, done_by=U.who(caller),
source_kind="odoo", source_ref=run_id, number=f"OPN/{odoo_db}/{wh}")
unit._opened.add(doc_id)
opening = L.virtual_id(company_code, "OPENING")
neg = base < 0
spec = MoveSpec(doc_id, int(q["id"]), p["product_id"], abs(base), where if neg else opening, opening if neg else where, cut,
p.get("base_unit_code") or "", abs(base), 1, 1, lot, UNRESTRICTED, UNRESTRICTED,
owner if neg else OURS, OURS if neg else owner, value_in=abs(value) if not neg else None, zero_value=owner != OURS,
allow_negative=True, reason="odoo opening balance")
pm = ledger.post_move(unit, spec)
mapped("stock.quant", q["id"], pm.row["move_id"])
created["opening_moves"] += 1
if history:
have = {r["odoo_move_line_id"] for r in unit.repo.find("inv_odoo_history", odoo_db=odoo_db)}
for ml in sc.history:
mv = sc.moves.get(ml["move_id"]) or {}
if ml["id"] in have or mv.get("state") != "done":
continue
p = sc.maps.product(ml["product_id"]) or {}
unit.put("inv_odoo_history", {"odoo_db": odoo_db, "odoo_move_line_id": ml["id"], "reference": mv.get("reference"),
"product_id": p.get("product_id"), "lot_code": (sc.lots.get(ml.get("lot_id")) or {}).get("name"),
"quantity_base": Decimal(str(ml["quantity"])) * int(p.get("uom_numerator") or 1) / int(p.get("uom_denominator") or 1),
"from_location_id": loc_id(sc.locations[ml["location_id"]]["complete_name"]) if ml["location_id"] in sc.locations else None,
"to_location_id": loc_id(sc.locations[ml["location_dest_id"]]["complete_name"]) if ml["location_dest_id"] in sc.locations else None,
"done_at": datetime.fromisoformat(str(ml["date"])), "origin": mv.get("origin")}, ("odoo_db", "odoo_move_line_id"))
created["history"] += 1
unit.flush_races()
unchanged = sum(1 for v in created.values() if v) == 0
unit.put("inv_migration_run", {"run_id": run_id, "odoo_db": odoo_db, "cutover": cut, "status": "imported", "signed_by": None, "signed_at": None,
"summary": json.dumps(created)}, "run_id")
return {"run_id": run_id, "created": created, "unchanged": unchanged, "negative_quants_carried": len(negatives)}
return U.run(caller, body)
def _ours_opening(unit, odoo_db: str):
return [m for m in unit.repo.find("inv_move", reason="odoo opening balance")]
@INV.use_case(kind="query", route="GET /migration/odoo/reconcile", permission=READ)
def reconcile_inventory_migration(caller: dict, odoo_db: Code, cutover: str, company_code: Code) -> dict:
"""The reconciliation the customer signs: every check answers {check, odoo, ours, equal, explanation}. Nothing is re-computed
from Odoo's valuation layers; they are read to compare."""
cut = day(cutover)
with U.read() as unit:
sc = Scope(odoo_db, None)
rows = []
def row(check, o, ours, explanation=""):
rows.append({"check": check, "odoo": o, "ours": ours, "equal": o == ours, "explanation": explanation})
odoo_qty: dict = {}
ours_qty: dict = {}
wh_o: dict = {}
wh_u: dict = {}
by_prod: dict = {}
for q in sc.quants:
p = _product(sc, q["product_id"])
base = Decimal(str(q["quantity"])) * int(p.get("uom_numerator") or 1) / int(p.get("uom_denominator") or 1)
if sc.locations[q["location_id"]]["usage"] != "internal":
continue
lot = sc.lots[q["lot_id"]]["name"] if q.get("lot_id") else ""
k = (p["product_id"], loc_id(sc.locations[q["location_id"]]["complete_name"]), lot)
odoo_qty[k] = odoo_qty.get(k, ZERO) + base
w = (p["product_id"], _wh_of(sc, q["location_id"]))
wh_o[w] = wh_o.get(w, ZERO) + base
by_prod[p["product_id"]] = q["product_id"]
for pid in {k[0] for k in odoo_qty}:
for m in unit.repo.moves_of(product_id=pid):
if m["moved_on"] > cut:
continue # at the cut-over: what the moves dated up to it left
for side, sign in (("to", 1), ("from", -1)):
loc = unit.location(m[f"{side}_location_id"])
if loc["kind"] != "internal":
continue
k = (pid, loc["location_id"], m["lot_code"] or "")
ours_qty[k] = ours_qty.get(k, ZERO) + sign * m["quantity_base"]
w = (pid, loc["warehouse_id"])
wh_u[w] = wh_u.get(w, ZERO) + sign * m["quantity_base"]
ours_qty = {k: v for k, v in ours_qty.items() if v != 0}
fmt = lambda d: {f"{'|'.join(k)}": qstr(v) for k, v in sorted(d.items())} # noqa: E731
row("on hand per product per internal location (and lot)", fmt(odoo_qty), fmt(ours_qty))
row("on hand per product per warehouse", fmt(wh_o), fmt(wh_u))
value_o: dict = {}
for l in sc.layers:
value_o[l["product_id"]] = value_o.get(l["product_id"], ZERO) + Decimal(str(l["value"]))
value_u: dict = {}
for m in _ours_opening(unit, odoo_db):
value_u[m["product_id"]] = value_u.get(m["product_id"], ZERO) + (m["value_amount"] or ZERO)
v_o = {sc.maps.product(k)["product_id"]: f"{v:.2f}" for k, v in value_o.items() if sc.maps.product(k)}
v_u = {k: f"{v:.2f}" for k, v in value_u.items()}
n_moves = len(_ours_opening(unit, odoo_db))
diff = sum(abs(Decimal(v_o.get(k, "0")) - Decimal(v_u.get(k, "0"))) for k in set(v_o) | set(v_u))
rows.append({"check": "stock value per product per company", "odoo": v_o, "ours": v_u, "equal": v_o == v_u,
"explanation": "" if v_o == v_u else (f"rounding per move, at most 0.01 per opening move ({n_moves} moves)" if diff <= Decimal("0.01") * n_moves
else "differs by more than rounding: a FIFO or standard-cost category per the door?")})
lots_o = len({((sc.maps.product(l["product_id"]) or {}).get("product_id"), l["name"]) for l in sc.lots.values()})
lots_u = len({(m["product_id"], m["lot_code"]) for m in unit.repo.find("inv_move", reason="odoo opening balance") if m["lot_code"]})
row("lots and serials", lots_o, lots_u)
neg_o = sum(1 for q in sc.quants if q["quantity"] < 0)
neg_u = sum(1 for m in _ours_opening(unit, odoo_db) if m["to_location_id"].endswith(":OPENING"))
row("negative quants", neg_o, neg_u, "carried as negative opening moves, to be cleared by a count" if neg_o else "")
done = [m for m in sc.history if (sc.moves.get(m["move_id"]) or {}).get("state") == "done"]
row("history carried", len(done), len(unit.repo.find("inv_odoo_history", odoo_db=odoo_db)))
return {"run_id": f"odoo:{odoo_db}:{cut.isoformat()}", "checks": rows, "all_equal": all(r["equal"] for r in rows)}
@INV.use_case(kind="command", route="POST /migration/odoo/sign", permission=IMPORT, idempotent=True)
def sign_inventory_migration(caller: dict, odoo_db: Code, cutover: str) -> dict:
"""The customer signs the reconciliation: the run is frozen from then on (nothing more is imported for that cut-over)."""
run_id = f"odoo:{odoo_db}:{day(cutover).isoformat()}"
@U.decide
def body(unit):
run = _run_row(unit.repo, run_id)
if run is None:
raise Refused(404, f"no run {run_id}")
if run["status"] == "signed":
return {"run_id": run_id, "status": "signed", "signed_by": run["signed_by"], "replay": True}
unit.put("inv_migration_run", {**run, "status": "signed", "signed_by": U.who(caller), "signed_at": unit.now_utc.replace(tzinfo=None)}, "run_id")
return {"run_id": run_id, "status": "signed", "signed_by": U.who(caller), "replay": False}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /migration/odoo/history", permission=READ)
def list_inventory_migration_runs(caller: dict, odoo_db: Code, product_id: Optional[Code] = None, limit: int = 100) -> dict:
"""Odoo's done stock moves, copied read-only: never part of on hand, never edited."""
if limit < 1 or limit > 1000:
raise BadInput("limit is 1 to 1000")
with U.read() as unit:
rows = unit.repo.find("inv_odoo_history", order=("odoo_move_line_id",), odoo_db=odoo_db, **({"product_id": product_id} if product_id else {}))
return {"count": len(rows), "lines": [{"odoo_move_line_id": r["odoo_move_line_id"], "reference": r["reference"], "product_id": r["product_id"],
"lot_code": r["lot_code"], "quantity_base": qstr(r["quantity_base"]), "from_location_id": r["from_location_id"],
"to_location_id": r["to_location_id"], "origin": r["origin"]} for r in rows[:limit]]}