dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/domain/odoo.py
"""Odoo account.move rows -> this brick's documents (migration.odoo.maps). Pure: rows in, documents out, nothing read or written.
History is copied as Odoo stored it: amounts to the cent, never recomputed; an issued document keeps Odoo's number and dates;
nothing here asks tax for a rate or the ledger for a journal (the ledger's own migration carries Odoo's entries).
"""
import json
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_ar.domain.credit_note import CreditNote
from brick_ar.domain.errors import RuleBroken
from brick_ar.domain.invoice import Invoice, Line, TaxRow
from brick_ar.domain.money import ZERO, money
from brick_ar.domain.numbering import parse_number
from brick_ar.domain.values import CreditState, InvoiceKind, InvoiceState, SourceKind
def D(v) -> Decimal:
return Decimal(str(v or 0))
def m2o(v):
"""The id of an Odoo many2one ([id, name] or False)."""
return v[0] if isinstance(v, (list, tuple)) and v else (v or None)
def m2o_name(v) -> str:
return v[1] if isinstance(v, (list, tuple)) and len(v) > 1 else ""
def as_date(v) -> Optional[date]:
if not v:
return None
return v if isinstance(v, date) else date.fromisoformat(str(v)[:10])
@dataclass
class MapContext:
"""What earlier bricks' migrations mapped (shared.yaml conventions.odoo_migration): Odoo id -> our natural key."""
db: str
run_id: str
company_currency: str = ""
bp: dict = field(default_factory=dict) # res.partner id -> bp_id
tax: dict = field(default_factory=dict) # account.tax id -> tax_code
product: dict = field(default_factory=dict) # product.product id -> product_id
unit: dict = field(default_factory=dict) # uom id -> UN/ECE Rec 20 code
so_line: dict = field(default_factory=dict) # sale.order.line id -> (sales_order_id, line_no)
partners: dict = field(default_factory=dict) # res.partner id -> the partner row (bill-to details)
multi_tax_lines: int = 0
def invoice_id(db: str, move_id) -> str:
return f"odoo:{db}:{move_id}"
def in_scope(move: dict, cutover: date, years: Optional[int], carry_drafts: bool) -> bool:
"""The filter of migration.odoo.reads: posted, or a numbered cancel, plus drafts only where the door carries them;
dated on or before the cutover and within the years the door keeps (older years stay as the Odoo PDFs)."""
state, name = move.get("state"), move.get("name")
if move.get("move_type") not in ("out_invoice", "out_refund"):
return False
if not (state == "posted" or (state == "cancel" and name not in (None, "", "/", False)) or (state == "draft" and carry_drafts)):
return False
d = as_date(move.get("invoice_date")) or as_date(move.get("date"))
if d and d > cutover:
return False
if years and d and d < date(cutover.year - years, cutover.month, min(cutover.day, 28)):
return False
return True
def _lines(lines: list, ctx: MapContext) -> list:
product_lines = sorted((ln for ln in lines if ln.get("display_type") in ("product", False, None)
and not ln.get("tax_line_id")), key=lambda ln: (ln.get("sequence") or 0, ln["id"]))
out = []
for i, ln in enumerate(product_lines, start=1):
ids = ln.get("tax_ids") or []
if len(ids) > 1:
ctx.multi_tax_lines += 1
net = money(D(ln.get("price_subtotal")))
so = ctx.so_line.get((ln.get("sale_line_ids") or [None])[0]) if ln.get("sale_line_ids") else None
out.append(Line(
line_no=i, description=(ln.get("name") or "").strip(), quantity=D(ln.get("quantity")),
unit_price=D(ln.get("price_unit")), discount_pct=D(ln.get("discount")),
unit_code=ctx.unit.get(m2o(ln.get("product_uom_id")), ""), product_id=ctx.product.get(m2o(ln.get("product_id")), ""),
tax_code=ctx.tax.get(ids[0], "") if ids else "", net_amount=net,
tax_amount=money(D(ln.get("price_total")) - D(ln.get("price_subtotal"))),
sales_order_id=str(so[0]) if so else "", so_line_no=so[1] if so else None, odoo_line_id=str(ln["id"])))
return out
def _tax_rows(lines: list, ctx: MapContext) -> list:
by: dict = {}
for ln in lines:
if ln.get("display_type") != "tax" and not ln.get("tax_line_id"):
continue
code = ctx.tax.get(m2o(ln.get("tax_line_id")), "") or ""
row = by.setdefault(code, [ZERO, ZERO])
row[0] += D(ln.get("tax_base_amount"))
row[1] += -D(ln.get("amount_currency") or ln.get("balance"))
return [TaxRow(code, (abs(v[1]) / abs(v[0]) * 100).quantize(Decimal("0.0001")) if v[0] else Decimal(0),
money(abs(v[0])), money(abs(v[1])), "odoo") for code, v in sorted(by.items())]
def map_move(move: dict, lines: list, ctx: MapContext):
"""An Invoice (out_invoice) or a CreditNote (out_refund), exactly as Odoo stored it. A failing invariant stops the load."""
partner = m2o(move.get("commercial_partner_id")) or m2o(move.get("partner_id"))
bp_id = ctx.bp.get(partner)
if bp_id is None:
raise RuleBroken(f"Odoo partner {partner} (move {move['id']}) has no bp_id: run the business-partner migration first")
sub, tax, tot = (money(D(move.get(k))) for k in ("amount_untaxed", "amount_tax", "amount_total"))
if sub + tax != tot:
raise RuleBroken(f"Odoo move {move['id']} ({move.get('name')}): untaxed {sub} + tax {tax} is not total {tot}")
currency = (m2o_name(move.get("currency_id")) or ctx.company_currency).upper()
signed = abs(money(D(move.get("amount_total_signed"))))
fx = functional = None
if currency != ctx.company_currency.upper() and tot:
fx, functional = (signed / tot).quantize(Decimal("0.0001")), signed
elif signed:
functional = signed
name = move.get("name") if move.get("name") not in (None, False, "/") else ""
series = (parse_number(name) or ("", 0))[0]
mapped = _lines(lines, ctx)
partner_row = ctx.partners.get(partner, {})
bill_to = json.dumps({"bill_to_name": partner_row.get("name", ""), "bill_to_address": ", ".join(
str(partner_row.get(k)) for k in ("street", "street2", "zip", "city") if partner_row.get(k)),
"invoice_email": partner_row.get("email") or ""}, sort_keys=True)
state = move.get("state")
if move["move_type"] == "out_invoice":
life = {"posted": InvoiceState.ISSUED, "cancel": InvoiceState.VOID, "draft": InvoiceState.DRAFT}[state].value
inv = Invoice(
invoice_id=invoice_id(ctx.db, move["id"]), bp_id=bp_id, lifecycle=life, kind=InvoiceKind.REGULAR.value,
source_kind=SourceKind.ODOO.value, source_id=(move.get("invoice_origin") or "") or "", invoice_no=name,
series_id=series, currency=currency, fx_rate=fx, issue_date=as_date(move.get("invoice_date")),
tax_point_date=as_date(move.get("date")), due_date=as_date(move.get("invoice_date_due")),
terms_id=f"odoo:{m2o(move.get('invoice_payment_term_id'))}" if m2o(move.get("invoice_payment_term_id")) else "",
customer_po_ref=(move.get("ref") or "") or "", bill_to_json=bill_to, subtotal=sub, tax_total=tax, total=tot,
total_functional=functional, migration_run_id=ctx.run_id, created_by=f"odoo:{ctx.db}")
inv.lines, inv.tax = mapped, _tax_rows(lines, ctx)
if state == "cancel":
inv.void_reason, inv.voided_on = "cancelled in Odoo", as_date(move.get("write_date")) or inv.issue_date
return inv
rev = move.get("reversed_entry_id")
cn = CreditNote(
credit_id=invoice_id(ctx.db, move["id"]), bp_id=bp_id, invoice_id=invoice_id(ctx.db, m2o(rev)) if m2o(rev) else None,
credit_no=name, series_id=series, credit_date=as_date(move.get("invoice_date")), currency=currency,
reason=("cancelled in Odoo" if state == "cancel" else "carried from Odoo"),
status={"posted": CreditState.APPROVED, "cancel": CreditState.CANCELLED, "draft": CreditState.DRAFT}[state].value,
subtotal=sub, tax_total=tax, total=tot, source_kind="odoo", drafted_by=f"odoo:{ctx.db}",
approved_by=f"odoo:{ctx.db}" if state == "posted" else "", migration_run_id=ctx.run_id)
cn.lines, cn.tax = mapped, _tax_rows(lines, ctx)
return cn
# ── payment terms ────────────────────────────────────────────────────────────────────────────────
def map_term(term: dict, term_lines: list) -> dict:
"""ar_terms row from account.payment.term (+ lines): one line -> days and basis; several -> instalments kept in the description."""
def days(ln):
return int(ln.get("nb_days") if ln.get("nb_days") is not None else ln.get("days") or 0)
def end(ln):
return ln.get("delay_type") == "days_after_end_of_month" or bool(ln.get("end_month")) or \
(ln.get("days_next_month") or 0) > 0
last = max(term_lines, key=days) if term_lines else {}
desc = term.get("name") or ""
if len(term_lines) > 1:
desc += " (instalments: " + ", ".join(f"{ln.get('value_amount', '')}{'%' if ln.get('value') == 'percent' else ''} "
f"at {days(ln)} days" for ln in term_lines) + ")"
disc = max((D(ln.get("discount_percentage")) for ln in term_lines), default=Decimal(0))
return {"terms_id": f"odoo:{term['id']}", "days": days(last), "basis": "end_of_month" if end(last) else "invoice_date",
"description": desc, "early_discount_pct": disc, "active": bool(term.get("active", True))}