dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/domain/reconcile.py
"""The reconciliation (migration.odoo.reconcile): every check computed on both sides, with the difference.
Pure: it is handed the Odoo rows the run read and the documents/settlement this brick holds, and returns MigrationChecks.
`must: equal` checks block signing when they differ and cannot be explained away; `must: explained` checks (gaps in the
numbering, the receivable account, in_payment invoices) may differ if a person says why.
"""
import hashlib
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_ar.domain.migration_run import MigrationCheck
from brick_ar.domain.money import ZERO, money
from brick_ar.domain.numbering import parse_number
from brick_ar.domain.odoo import D, MapContext, as_date, m2o, m2o_name
OURS_STATE = {"issued": "posted", "approved": "posted", "void": "cancel", "cancelled": "cancel", "draft": "draft"}
BUCKETS = ("current", "1-30", "31-60", "61-90", "90+")
@dataclass
class OdooData:
moves: list # the moves in scope
lines: dict # move id -> account.move.line rows
ctx: MapContext
cutover: date
receivable: dict = field(default_factory=dict) # currency -> trial balance of the receivable account(s) at cutover
@dataclass
class OursData:
invoices: list
credits: list
outstanding: dict = field(default_factory=dict) # invoice_id -> outstanding from payments, None when unknown
credit_remaining: dict = field(default_factory=dict) # credit_id -> unallocated credit from payments
def _q(d: Optional[date]) -> str:
return f"{d.year}-Q{(d.month - 1) // 3 + 1}" if d else "none"
def _bucket(due: Optional[date], at: date) -> str:
days = (at - due).days if due else 0
return "current" if days <= 0 else "1-30" if days <= 30 else "31-60" if days <= 60 else "61-90" if days <= 90 else "90+"
def _fmt(v) -> str:
return str(v) if not isinstance(v, Decimal) else f"{money(v):.2f}"
def _gaps(numbers: list) -> dict:
"""series -> the counters missing between the lowest and highest carried."""
by = defaultdict(set)
for n in numbers:
p = parse_number(n)
if p:
by[p[0]].add(p[1])
def missing(v: set) -> str: # the numbers between the lowest and the highest carried that no document holds
top = max(v)
return ",".join(str(i) for i in range(min(v), top + 1) if i not in v)
return {s: missing(v) for s, v in by.items()}
def _sign(move_type: str) -> int:
return -1 if move_type == "out_refund" else 1
def _check(run_id, key, must, odoo: dict, ours: dict, zero="0") -> list:
out = []
for sl in sorted(set(odoo) | set(ours)):
o, u = odoo.get(sl, zero), ours.get(sl, zero)
try:
diff = _fmt(Decimal(u) - Decimal(o))
except Exception: # a list or a hash: the difference is "differs"
diff = "" if o == u else "differs"
out.append(MigrationCheck(run_id=run_id, check_key=key, slice=sl, odoo_value=o, ours_value=u, difference=diff,
equal=(o == u), must=must))
return out or [MigrationCheck(run_id=run_id, check_key=key, slice="all", odoo_value=zero, ours_value=zero,
difference="0", equal=True, must=must)]
def compute_checks(run_id: str, odoo: OdooData, ours: OursData) -> list:
ctx, checks = odoo.ctx, []
docs = [(i, i.issue_date, OURS_STATE[i.lifecycle], "out_invoice") for i in ours.invoices] + \
[(c, c.credit_date, OURS_STATE[c.status], "out_refund") for c in ours.credits]
# 1. document count by move_type, state and year
o = defaultdict(int)
for m in odoo.moves:
d = as_date(m.get("invoice_date")) or as_date(m.get("date"))
o[f"{m['move_type']}|{m['state']}|{d.year if d else 'none'}"] += 1
u = defaultdict(int)
for doc, d, st, mt in docs:
u[f"{mt}|{st}|{d.year if d else 'none'}"] += 1
checks += _check(run_id, "document_count", "equal", {k: str(v) for k, v in o.items()}, {k: str(v) for k, v in u.items()})
# 2. totals by move_type, currency and year (untaxed, tax, total)
for key, field_, attr in (("total_untaxed", "amount_untaxed", "subtotal"), ("total_tax", "amount_tax", "tax_total"),
("total_amount", "amount_total", "total")):
o = defaultdict(lambda: ZERO)
for m in odoo.moves:
d = as_date(m.get("invoice_date")) or as_date(m.get("date"))
cur = (m2o_name(m.get("currency_id")) or ctx.company_currency).upper()
o[f"{m['move_type']}|{cur}|{d.year if d else 'none'}"] += money(D(m.get(field_)))
u = defaultdict(lambda: ZERO)
for doc, d, _st, mt in docs:
u[f"{mt}|{doc.currency}|{d.year if d else 'none'}"] += getattr(doc, attr)
checks += _check(run_id, key, "equal", {k: _fmt(v) for k, v in o.items()}, {k: _fmt(v) for k, v in u.items()}, "0.00")
# 3. tax by tax code and VAT period (invoices less credit notes)
for key, idx in (("tax_net", 0), ("tax_amount", 1)):
o = defaultdict(lambda: ZERO)
for m in odoo.moves:
if m["state"] == "cancel" or m["state"] == "draft":
continue
rows = defaultdict(lambda: [ZERO, ZERO])
for ln in odoo.lines.get(m["id"], []):
if ln.get("display_type") == "tax" or ln.get("tax_line_id"):
code = ctx.tax.get(m2o(ln.get("tax_line_id")), "")
rows[code][0] += D(ln.get("tax_base_amount"))
rows[code][1] += -D(ln.get("amount_currency") or ln.get("balance"))
for code, v in rows.items():
o[f"{code}|{_q(as_date(m.get('date')))}"] += _sign(m["move_type"]) * money(abs(v[idx]))
u = defaultdict(lambda: ZERO)
for doc, d, st, mt in docs:
if st != "posted":
continue
point = getattr(doc, "tax_point_date", None) or d
for r in doc.tax:
u[f"{r.tax_code}|{_q(point)}"] += _sign(mt) * (r.net_amount if idx == 0 else r.tax_amount)
checks += _check(run_id, key, "equal", {k: _fmt(v) for k, v in o.items()}, {k: _fmt(v) for k, v in u.items()}, "0.00")
# 4. lines sum to the document
oc = sum(1 for m in odoo.moves if m["state"] != "draft" and
sum((money(D(ln.get("price_subtotal"))) for ln in odoo.lines.get(m["id"], [])
if ln.get("display_type") in ("product", False, None) and not ln.get("tax_line_id")), ZERO)
== money(D(m.get("amount_untaxed"))))
uc = sum(1 for doc, _d, st, _mt in docs if st != "draft" and sum((ln.net_amount for ln in doc.lines), ZERO) == doc.subtotal)
checks += _check(run_id, "lines_sum_to_document", "equal", {"consistent": str(oc)}, {"consistent": str(uc)})
# 5. every number carried once (count and a digest of the set, per series)
def digest(names):
by = defaultdict(list)
for n in names:
p = parse_number(n)
by[p[0] if p else "unnumbered"].append(n)
return {s: f"{len(v)}:{hashlib.sha1(chr(10).join(sorted(v)).encode()).hexdigest()[:12]}" for s, v in by.items()}
on = [m["name"] for m in odoo.moves if m.get("name") not in (None, False, "", "/")]
un = [getattr(doc, "invoice_no", None) or getattr(doc, "credit_no", "") for doc, *_ in docs]
checks += _check(run_id, "numbers_carried", "equal", digest(on), digest([n for n in un if n]), "0:")
# 6. numbering gaps: a gap in the sequence is never "equal": a person says why (credit notes deleted, test documents)
gaps = {s: v for s, v in _gaps(on).items() if v}
for c in _check(run_id, "numbering_gaps", "explained", gaps, gaps, ""):
if c.odoo_value:
c.equal, c.difference = False, "missing: " + c.odoo_value
checks.append(c)
# 7-8. open balance and its aging: Odoo's residual against payments' settlement
o_open, u_open = defaultdict(lambda: ZERO), defaultdict(lambda: ZERO)
o_age, u_age = defaultdict(lambda: ZERO), defaultdict(lambda: ZERO)
for m in odoo.moves:
if m["state"] != "posted" or m.get("payment_state") not in ("not_paid", "partial", "in_payment"):
continue
bp = ctx.bp.get(m2o(m.get("commercial_partner_id")) or m2o(m.get("partner_id")), 0)
cur = (m2o_name(m.get("currency_id")) or ctx.company_currency).upper()
amt = _sign(m["move_type"]) * money(D(m.get("amount_residual")))
o_open[f"{bp}|{cur}"] += amt
o_age[f"{bp}|{cur}|{_bucket(as_date(m.get('invoice_date_due')), odoo.cutover)}"] += amt
unknown = False
for inv in ours.invoices:
out = ours.outstanding.get(inv.invoice_id)
if out is None:
unknown = unknown or inv.lifecycle == "issued"
continue
if inv.lifecycle != "issued" or out == ZERO:
continue
u_open[f"{inv.bp_id}|{inv.currency}"] += out
u_age[f"{inv.bp_id}|{inv.currency}|{_bucket(inv.due_date, odoo.cutover)}"] += out
for cn in ours.credits:
rem = ours.credit_remaining.get(cn.credit_id)
if cn.status == "approved" and rem:
u_open[f"{cn.bp_id}|{cn.currency}"] -= rem
checks += _check(run_id, "open_balance", "equal", {k: _fmt(v) for k, v in o_open.items() if v},
{k: _fmt(v) for k, v in u_open.items() if v}, "0.00")
checks += _check(run_id, "aged_open_balance", "equal", {k: _fmt(v) for k, v in o_age.items() if v},
{k: _fmt(v) for k, v in u_age.items() if v}, "0.00")
# 9. receivable account balance at cutover (explained): the trial balance against the open balance above
by_cur = defaultdict(lambda: ZERO)
for k, v in u_open.items():
by_cur[k.split("|")[1]] += v
checks += _check(run_id, "receivable_balance", "explained", {c: _fmt(money(v)) for c, v in odoo.receivable.items()},
{c: _fmt(v) for c, v in by_cur.items()}, "0.00")
# 10. in_payment invoices (explained): open on our side until payments carries the receipt
ip = [m for m in odoo.moves if m.get("payment_state") == "in_payment" and m["state"] == "posted"]
still = [m for m in ip if (ours.outstanding.get(invoice_id_of(odoo.ctx, m)) or ZERO) > ZERO]
checks += _check(run_id, "in_payment_invoices", "explained", {"count": str(len(ip))}, {"count": str(len(still))})
for c in checks:
c.slice = c.slice or "all"
return _note_unknown(checks, unknown) if unknown else checks
def invoice_id_of(ctx: MapContext, m: dict) -> str:
return f"odoo:{ctx.db}:{m['id']}"
def _note_unknown(checks: list, unknown: bool) -> list:
"""Payments did not answer for some invoice: the open-balance checks cannot pass on a guess."""
for c in checks:
if c.check_key in ("open_balance", "aged_open_balance") and c.equal:
c.equal, c.difference = False, "payments has not settled these invoices yet"
return checks