dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/odoo_migration.py
"""The Odoo migration slice (module.yaml migration.odoo, TAX-15): the customer's Odoo taxes, fiscal positions and customer VAT
data become this module's FIRST rows, so the first invoice after cutover is taxed exactly as Odoo would have taxed it.
Three steps, each testable alone:
read `read_odoo` pulls the rows the spec names through the reader port (read-only) and filters them again here
plan `plan` is a PURE function of (rows, the door's answers, how partners resolve): what to write and the map entries
apply `apply` writes the plan idempotently — first rows only, keyed in tax_migration_map; a second run writes nothing,
a row that exists with different content is reported (`differs`) and left alone
reconcile `reconcile` compares Odoo with what is now here: {check, odoo, ours, equal, explanation}
History is never touched: no issued document's tax is read or written by the plan or the apply; the move lines are read only
to reconcile.
"""
import re
from dataclasses import dataclass, field
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from brick_tax.domain import rules as rules_mod
from brick_tax.domain import uk_pack
from brick_tax.domain.errors import RuleBroken
from brick_tax.domain.rules import Rule
from brick_tax.domain.tax_code import TaxCode
from brick_tax.domain.values import CODE, FOREVER, ChangeState, CodeState, Treatment, clean_vat_number
from brick_tax.infrastructure import odoo
from brick_tax.infrastructure.tables import tax_document, tax_document_line
# The l10n_uk external ids the proposal knows (confirmed at the door); everything else is proposed from its rate and name.
XMLID_CODES = {"ST11": "S", "ST2": "Z", "ST0": "E", "ST1": "R", "ST4": "OS", "ST5": "NV"}
CATEGORY_OF = {Treatment.STANDARD.value: "standard", Treatment.REDUCED.value: "reduced", Treatment.ZERO.value: "zero",
Treatment.EXEMPT.value: "exempt", Treatment.REVERSE_CHARGE.value: "construction_service",
Treatment.OUTSIDE_SCOPE.value: "outside_scope"}
DEFAULT_TREATMENTS = ("uk_business", "uk_consumer")
MODELS = {"company": "res.company", "tax": "account.tax", "repartition": "account.tax.repartition.line", "tag": "account.account.tag",
"group": "account.tax.group", "position": "account.fiscal.position", "position_tax": "account.fiscal.position.tax",
"template": "product.template", "partner": "res.partner", "move_line": "account.move.line"}
@dataclass
class Rows:
company: dict = field(default_factory=dict)
taxes: list = field(default_factory=list)
repartition: list = field(default_factory=list)
tags: list = field(default_factory=list)
positions: list = field(default_factory=list)
position_taxes: list = field(default_factory=list)
templates: list = field(default_factory=list)
partners: list = field(default_factory=list)
move_lines: list = field(default_factory=list)
def read_odoo(company_id: int, window: tuple) -> Rows:
"""Pull the rows migration.odoo.reads names, through the read-only reader, and filter them to this company again."""
r = odoo.reader
company = next((c for c in r.read(MODELS["company"], ["id", "vat", "country_id", "account_fiscal_country_id",
"tax_calculation_rounding_method"]) if c.get("id") == company_id), None)
if company is None:
raise RuleBroken(f"Odoo has no company {company_id}")
mine = lambda row: row.get("company_id") in (None, company_id) # noqa: E731
taxes = [t for t in r.read(MODELS["tax"], ["id", "name", "description", "invoice_label", "amount", "amount_type", "type_tax_use",
"tax_scope", "price_include", "include_base_amount", "tax_exigibility", "tax_group_id",
"country_id", "active", "company_id", "xmlid"], include_archived=True)
if mine(t) and t.get("type_tax_use") == "sale"]
ids = [t["id"] for t in taxes]
return Rows(
company=company, taxes=taxes,
repartition=[x for x in r.read(MODELS["repartition"], ["tax_id", "document_type", "repartition_type", "factor_percent", "tag_ids"],
{"tax_id": ("in", ids)}) if x.get("tax_id") in ids],
tags=r.read(MODELS["tag"], ["id", "name", "applicability", "country_id"], {"applicability": "taxes"}),
positions=[p for p in r.read(MODELS["position"], ["id", "name", "country_id", "country_group_id", "vat_required", "auto_apply",
"active", "company_id"]) if mine(p) and p.get("active", True)],
position_taxes=r.read(MODELS["position_tax"], ["id", "position_id", "tax_src_id", "tax_dest_id"]),
templates=r.read(MODELS["template"], ["id", "taxes_id", "type", "sale_ok"], {"sale_ok": True}),
partners=r.read(MODELS["partner"], ["id", "vat", "country_id", "is_company", "property_account_position_id", "customer_rank"]),
move_lines=[m for m in r.read(MODELS["move_line"], ["id", "move_id", "date", "move_type", "state", "tax_ids", "tax_line_id",
"tax_tag_ids", "balance", "tax_base_amount", "product_tmpl_id", "partner_id"],
{"state": "posted"}) if m.get("state", "posted") == "posted"
and m.get("move_type") in ("out_invoice", "out_refund") and window[0] <= _d(m.get("date")) <= window[1]])
def _d(v) -> date:
return v if isinstance(v, date) else date.fromisoformat(str(v)[:10])
def _ids(v) -> list:
return [int(x) for x in v] if isinstance(v, (list, tuple)) else ([int(v)] if v else [])
@dataclass
class PlannedCode:
odoo_id: int
tax_code: str
treatment: str
description: str
wording: str
reporting_box: str
rate_pct: Decimal
charged_pct: Decimal
lifecycle: str # active | retired
creates: bool = True # False: maps onto a code another Odoo tax already created in this plan
odoo_name: str = ""
odoo_amount: Decimal = Decimal(0)
@dataclass
class Plan:
codes: list = field(default_factory=list)
treatments: list = field(default_factory=list) # (treatment_code, description, needs_vat_number, odoo_position_id)
categories: list = field(default_factory=list) # (category_code, description)
rules: list = field(default_factory=list) # Rule
rule_maps: list = field(default_factory=list) # (odoo_position_tax_id, rule scope key text)
parties: list = field(default_factory=list) # (odoo_partner_id, bp_id, treatment_code, vat_number)
registration: Optional[dict] = None
rounding: Optional[tuple] = None
handed_to_product: dict = field(default_factory=dict) # product.template id -> category_code
unresolved_partners: list = field(default_factory=list)
notes: list = field(default_factory=list)
skipped_taxes: list = field(default_factory=list)
code_of_tax: dict = field(default_factory=dict) # odoo tax id -> tax code
treatment_of_position: dict = field(default_factory=dict)
proposals: dict = field(default_factory=dict)
def _tag_names(rows: Rows) -> dict:
return {t["id"]: str(t.get("name") or "") for t in rows.tags}
def _box(rows: Rows, tax_id: int) -> str:
names, boxes = _tag_names(rows), set()
for rp in rows.repartition:
if rp.get("tax_id") == tax_id and rp.get("document_type") == "invoice":
for tag in _ids(rp.get("tag_ids")):
m = re.search(r"(\d+)", names.get(tag, ""))
if m:
boxes.add(int(m.group(1)))
return ",".join(str(b) for b in sorted(boxes))
def _is_reverse_charge(rows: Rows, tax_id: int) -> bool:
factors = sorted(Decimal(str(rp.get("factor_percent") or 0)) for rp in rows.repartition
if rp.get("tax_id") == tax_id and rp.get("document_type") == "invoice" and rp.get("repartition_type") == "tax")
return Decimal(-100) in factors and Decimal(100) in factors
def _treatment_of(rows: Rows, t: dict, amount: Decimal) -> str:
if _is_reverse_charge(rows, t["id"]):
return Treatment.REVERSE_CHARGE.value
name = f"{t.get('name') or ''} {t.get('description') or ''}".lower()
if amount == 0:
if "exempt" in name:
return Treatment.EXEMPT.value
if "outside" in name or "out of scope" in name or "no vat" in name:
return Treatment.OUTSIDE_SCOPE.value
return Treatment.ZERO.value
return Treatment.REDUCED.value if amount < 20 else Treatment.STANDARD.value
def _propose_code(treatment: str, t: dict, taken: set) -> str:
code = XMLID_CODES.get(str(t.get("xmlid") or "").split(".")[-1].split("_")[-1].upper()) or \
{"standard": "S", "reduced": "R", "zero": "Z", "exempt": "E", "outside_scope": "OS", "reverse_charge": "RC"}[treatment]
return code if code not in taken else f"{code}{t['id']}"
def _supply(scope) -> str:
return {"service": "services", "consu": "goods", "goods": "goods", "services": "services"}.get(str(scope or ""), "any")
def _classify_position(name: str) -> Optional[str]:
n = (name or "").lower()
if any(w in n for w in ("cis", "reverse charge", "construction")):
return "uk_cis_contractor"
if n.strip() in ("uk", "united kingdom", "domestic", "gb") or n.startswith("uk b2b"):
return "uk_business"
if "consumer" in n or "b2c" in n:
return "uk_consumer" if n.startswith("uk") else "overseas_consumer"
if any(w in n for w in ("b2b", "eu", "world", "export", "overseas", "international", "non-eu", "rest of")):
return "overseas_business"
return None
def plan(rows: Rows, door, resolve_partner, existing_treatments: dict) -> Plan:
"""Everything the migration would write, from the rows and the door's answers. Pure: no storage."""
p = Plan()
cutover = _d(door.cutover_date)
used = {_ for m in rows.move_lines for _ in [m.get("tax_line_id")] if _} | {x for m in rows.move_lines for x in _ids(m.get("tax_ids"))}
# ── the company: registration number and the rounding method ──────────────────────────────
vat = (rows.company.get("vat") or "").strip()
if vat:
number = clean_vat_number(vat)
if not number:
raise RuleBroken(f"Odoo's company VAT number '{vat}' is not in the GB format (GB and 9 or 12 digits): correct it in Odoo, "
"or record the registration here by hand")
p.registration = {"status": "registered", "vat_number": number, "effective_from": _d(door.registration_from)}
else:
p.registration = {"status": "not_registered", "vat_number": "", "effective_from": _d(door.registration_from)}
method = rows.company.get("tax_calculation_rounding_method")
if method == "round_globally":
if not door.rounding_method:
raise RuleBroken("Odoo rounds the total half-up (round_globally), which is no HMRC method: choose per_line_nearest, "
"per_line_down_0_1p or per_invoice_round_down from the cutover (rounding_method)")
p.rounding = (door.rounding_method, cutover)
else:
p.rounding = (door.rounding_method or "per_line_nearest", cutover)
# ── the taxes ─────────────────────────────────────────────────────────────────────────────
taken, unsupported = set(), []
for t in sorted(rows.taxes, key=lambda x: x["id"]):
active = t.get("active", True)
if not active and t["id"] not in used:
p.skipped_taxes.append({"odoo_id": t["id"], "name": t.get("name"), "why": "archived and no posted invoice used it"})
continue
amount = Decimal(str(t.get("amount") or 0))
mapped = door.code_map.get(str(t["id"]))
if t.get("amount_type") not in (None, "percent") or t.get("include_base_amount"):
if not mapped:
unsupported.append(f"{t.get('name')} (id {t['id']}, {t.get('amount_type')}"
f"{', tax on tax' if t.get('include_base_amount') else ''})")
continue
p.code_of_tax[t["id"]] = mapped # replaced by a percent code the door named
p.notes.append(f"Odoo tax {t['id']} ({t.get('name')}) is not a percent tax: replaced by {mapped} at the door")
continue
treatment = _treatment_of(rows, t, amount)
code = mapped or _propose_code(treatment, t, taken)
if not CODE.match(code):
raise RuleBroken(f"'{code}' is not a usable tax code for Odoo tax {t['id']}")
rc = treatment == Treatment.REVERSE_CHARGE.value
wording = (t.get("invoice_label") or t.get("description") or "").strip()
if rc and not wording:
wording = "Reverse charge: Customer to pay the VAT to HMRC"
p.notes.append(f"{code}: Odoo has no reverse-charge legend; the standard one is used — check it at the door")
planned = PlannedCode(t["id"], code, treatment, (t.get("description") or t.get("name") or code).strip(), wording,
_box(rows, t["id"]), amount, Decimal(0) if rc else amount,
CodeState.ACTIVE.value if active else CodeState.RETIRED.value, odoo_name=str(t.get("name") or ""),
odoo_amount=amount)
clash = next((c for c in p.codes if c.tax_code == code), None)
if clash:
if (clash.rate_pct, clash.charged_pct, clash.treatment) != (planned.rate_pct, planned.charged_pct, planned.treatment):
raise RuleBroken(f"Odoo taxes {clash.odoo_id} and {t['id']} both map to {code} but differ in rate or treatment: "
"give one of them another code (code_map)")
planned.creates = False
taken.add(code)
p.codes.append(planned)
p.code_of_tax[t["id"]] = code
p.proposals[str(t["id"])] = code
if unsupported:
raise RuleBroken("this module cannot carry these Odoo taxes (fixed, division, group, python code or tax on tax) — replace "
"them with a percent code (code_map) or stop: " + "; ".join(unsupported))
treatment_of_code = {c.tax_code: c.treatment for c in p.codes}
# ── treatments (one per fiscal position) ──────────────────────────────────────────────────
undecided = []
for pos in sorted(rows.positions, key=lambda x: x["id"]):
code = door.treatment_map.get(str(pos["id"])) or _classify_position(pos.get("name"))
if not code:
undecided.append(f"{pos.get('name')} (id {pos['id']})")
continue
p.treatment_of_position[pos["id"]] = code
p.proposals[f"position:{pos['id']}"] = code
if code not in existing_treatments and not any(t[0] == code for t in p.treatments):
p.treatments.append((code, str(pos.get("name") or code), bool(pos.get("vat_required")), pos["id"]))
if undecided:
raise RuleBroken("which customer treatment is each of these fiscal positions? " + "; ".join(undecided) +
" — answer in treatment_map (Odoo position id -> treatment code)")
for tr, description, needs_vat in uk_pack.UK_TREATMENTS: # the company default (no position) is uk_business or uk_consumer
if tr in DEFAULT_TREATMENTS and tr not in existing_treatments and not any(t[0] == tr for t in p.treatments):
p.treatments.append((tr, description, needs_vat, 0))
# ── categories: one per distinct default sale tax, handed to product ──────────────────────
def category_of_code(code: str) -> Optional[str]:
return CATEGORY_OF.get(treatment_of_code.get(code, ""))
cats = {}
for tpl in sorted(rows.templates, key=lambda x: x["id"]):
first = next((t for t in _ids(tpl.get("taxes_id")) if t in p.code_of_tax), None)
cat = category_of_code(p.code_of_tax[first]) if first else None
if cat:
p.handed_to_product[tpl["id"]] = cat
cats.setdefault(cat, p.code_of_tax[first])
own_code = {}
for c in sorted(p.codes, key=lambda c: c.odoo_id):
if c.lifecycle == CodeState.ACTIVE.value:
own_code.setdefault(CATEGORY_OF.get(c.treatment), c.tax_code)
p.categories = [(cat, f"migrated from Odoo: {cat.replace('_', ' ')}") for cat in sorted(set(cats) | {k for k in own_code if k})]
# ── rules: the fiscal positions' mappings, then the default treatments' own codes ─────────
wanted: dict = {}
for fpt in sorted(rows.position_taxes, key=lambda x: x.get("id") or 0):
pos = fpt.get("position_id")
if pos not in p.treatment_of_position:
continue
src, dest = fpt.get("tax_src_id"), fpt.get("tax_dest_id")
if src not in p.code_of_tax:
continue
if not dest or dest not in p.code_of_tax:
p.notes.append(f"fiscal position {pos}: tax {src} maps to no tax (or one not carried): that pair stays uncovered and is refused")
continue
cat = category_of_code(p.code_of_tax[src])
if not cat:
continue
key = (cat, p.treatment_of_position[pos], _supply(next((t.get("tax_scope") for t in rows.taxes if t["id"] == src), None)))
if key in wanted and wanted[key][0] != p.code_of_tax[dest]:
p.notes.append(f"{key}: two fiscal-position mappings disagree; the first (lowest id) is kept")
continue
wanted.setdefault(key, (p.code_of_tax[dest], fpt))
for cat, code in own_code.items():
for tr in DEFAULT_TREATMENTS:
wanted.setdefault((cat, tr, "any"), (code, None))
for (cat, tr, supply), (code, fpt) in sorted(wanted.items()):
p.rules.append(Rule(cat, tr, supply, cutover, code, 0, None, ChangeState.APPROVED.value, "odoo-migration"))
if fpt is not None and fpt.get("id") is not None:
p.rule_maps.append((fpt["id"], f"{cat}|{tr}|{supply}|{cutover}"))
# ── customers ─────────────────────────────────────────────────────────────────────────────
needs = {code: needs for code, _, needs, _ in p.treatments}
needs |= {k: v["needs_vat_number"] for k, v in existing_treatments.items()}
stuck = []
for part in sorted(rows.partners, key=lambda x: x["id"]):
bp = resolve_partner(part["id"])
if bp is None:
p.unresolved_partners.append(part["id"])
continue
tr = p.treatment_of_position.get(part.get("property_account_position_id")) or ("uk_business" if part.get("is_company") else "uk_consumer")
number = (part.get("vat") or "").strip().upper().replace(" ", "")
if needs.get(tr) and not number:
if door.missing_vat_to and not needs.get(door.missing_vat_to):
tr = door.missing_vat_to
else:
stuck.append(f"{part.get('id')}")
continue
p.parties.append((part["id"], bp, tr, number))
if stuck:
raise RuleBroken("these Odoo customers are on a treatment that needs a VAT number but have none: " + ", ".join(stuck) +
" — record the number in Odoo, or give missing_vat_to (a consumer treatment)")
return p
# ── apply ────────────────────────────────────────────────────────────────────────────────────
def apply(repo, p: Plan, door, by: str) -> dict:
"""Write the plan as FIRST rows. Idempotent: what exists is left alone (and listed if it differs)."""
db, cutover = door.odoo_db, _d(door.cutover_date)
out = {k: 0 for k in ("codes", "rates", "treatments", "categories", "rules", "parties", "registration", "settings", "map")}
same = {k: 0 for k in out}
differs = []
def mapped(model, oid, table, key):
if repo.migration_map.get(db, model, oid) is None:
repo.migration_map.put(db, model, oid, table, key, by)
out["map"] += 1
else:
same["map"] += 1
made = set()
for c in p.codes:
if c.creates and c.tax_code not in made:
made.add(c.tax_code)
have = repo.codes.get(c.tax_code)
if have is None:
t = TaxCode.from_history(c.tax_code, c.treatment, description=c.description, document_wording=c.wording,
reporting_box=c.reporting_box, rate_pct=c.rate_pct, charged_pct=c.charged_pct,
valid_from=cutover, lifecycle=c.lifecycle, by=by)
repo.codes.save(t, 0)
out["codes"] += 1
out["rates"] += 1
else:
existing = have[0]
r = existing.rate_on(cutover)
if existing.treatment != c.treatment or r is None or (r.rate_pct, r.charged_pct) != (c.rate_pct, c.charged_pct):
differs.append(f"tax code {c.tax_code} exists with another treatment or rate than Odoo tax {c.odoo_id}")
same["codes"] += 1
mapped("account.tax", c.odoo_id, "tax_code", c.tax_code)
for tid, code in p.code_of_tax.items():
if not any(c.odoo_id == tid for c in p.codes):
mapped("account.tax", tid, "tax_code", code)
for code, description, needs, pos in p.treatments:
if repo.treatments.get(code) is None:
repo.treatments.put(code, description, needs)
out["treatments"] += 1
else:
same["treatments"] += 1
for pos, code in p.treatment_of_position.items():
mapped("account.fiscal.position", pos, "tax_treatment", code)
for cat, description in p.categories:
if repo.categories.get(cat) is None:
repo.categories.put(cat, description, by)
out["categories"] += 1
else:
same["categories"] += 1
stored: dict = {}
for x in rules_mod.live(repo.rules.all()):
stored.setdefault(x.scope, []).append(x)
for r in p.rules:
# A migrated rule is written only where this scope has no rule at all: a pair already in use is never back-dated
# to the cutover (documents taxed since would change). It is listed for a person to change from tomorrow.
have = stored.get(r.scope)
if not have:
repo.rules.put(r)
out["rules"] += 1
else:
same["rules"] += 1
latest = max(have, key=lambda x: x.valid_from)
if latest.tax_code != r.tax_code:
differs.append(f"rule {r.describe()} already chooses {latest.tax_code or 'nothing'} (from {latest.valid_from}), "
f"Odoo says {r.tax_code}: left as it is; change it from tomorrow if Odoo is right")
for oid, key in p.rule_maps:
mapped("account.fiscal.position.tax", oid, "tax_rule", key)
for odoo_partner, bp, tr, number in p.parties:
rows_ = repo.parties.rows_of(bp)
if not rows_:
repo.parties.put(bp, FOREVER, tr, number, by)
out["parties"] += 1
else:
same["parties"] += 1
if rows_[0]["treatment_code"] != tr or (rows_[0]["vat_number"] or "") != number:
differs.append(f"customer {bp} already has treatment {rows_[0]['treatment_code']}, Odoo says {tr}")
mapped("res.partner", odoo_partner, "tax_party", f"{bp}|{FOREVER}")
reg = p.registration
if not repo.registrations.rows():
from brick_tax.domain.registration import Registration
repo.registrations.put(Registration("GB", reg["effective_from"], reg["status"], reg["vat_number"], "standard", None,
ChangeState.APPROVED.value, by))
out["registration"] += 1
else:
same["registration"] += 1
mapped("res.company", door.company_id, "tax_registration", f"GB|{reg['effective_from']}")
method, since = p.rounding
if not [r for r in repo.settings.rows() if r.setting_key == "rounding_method"]:
repo.settings.put("rounding_method", since, method, by)
out["settings"] += 1
else:
same["settings"] += 1
return {"created": out, "unchanged": same, "differs": differs}
# ── reconcile ────────────────────────────────────────────────────────────────────────────────
def _check(name, odoo_v, ours_v, equal, why) -> dict:
return {"check": name, "odoo": odoo_v, "ours": ours_v, "equal": bool(equal), "explanation": why}
def reconcile(repo, rows: Rows, p: Plan, door) -> list:
"""Compare Odoo with what is here now. Run after the apply has committed (no read-your-writes inside a transaction)."""
db, cutover = door.odoo_db, _d(door.cutover_date)
codes = {c.tax_code: c for c in repo.codes.all()}
facts = repo.codes.facts_on(cutover)
maps = {m["odoo_id"]: m["our_key"] for m in repo.migration_map.all(db) if m["odoo_model"] == "account.tax"}
out = []
active = [t for t in rows.taxes if t.get("active", True) and t["id"] in p.code_of_tax]
ours_active = [t for t in active if maps.get(str(t["id"])) in codes and codes[maps[str(t["id"])]].lifecycle == CodeState.ACTIVE.value]
out.append(_check("active sale taxes", len(active), len(ours_active), len(active) == len(ours_active),
"every active Odoo sale tax maps to an active code" if len(active) == len(ours_active)
else "an active Odoo sale tax has no active code here"))
bad_rate = []
for t in active:
f = facts.get(maps.get(str(t["id"]), ""))
if f is None or f.rate_pct != Decimal(str(t.get("amount") or 0)):
bad_rate.append(f"{t.get('name')}: Odoo {t.get('amount')}, here {None if f is None else f.rate_pct}")
out.append(_check("rate per code", {str(t["id"]): str(t.get("amount")) for t in active},
{maps.get(str(t["id"]), "?"): None if facts.get(maps.get(str(t["id"]), "")) is None
else str(facts[maps[str(t["id"])]].rate_pct) for t in active}, not bad_rate,
"every rate (and the charged rate under reverse charge) equals Odoo's on the cutover date" if not bad_rate else "; ".join(bad_rate)))
bad_box = [f"{t.get('name')}: Odoo {_box(rows, t['id'])!r}, here {codes[maps[str(t['id'])]].reporting_box!r}"
for t in active if maps.get(str(t["id"])) in codes and codes[maps[str(t["id"])]].reporting_box != _box(rows, t["id"])]
out.append(_check("reporting box per code", {str(t["id"]): _box(rows, t["id"]) for t in active},
{maps.get(str(t["id"]), "?"): codes[maps[str(t["id"])]].reporting_box for t in active if maps.get(str(t["id"])) in codes},
not bad_box, "the invoice-side tax tags name the same boxes" if not bad_box else "; ".join(bad_box)))
migrated = {odoo_partner: (bp, tr, number) for odoo_partner, bp, tr, number in p.parties}
parties = repo.parties.all_current(date.max)
odoo_vat = sum(1 for part in rows.partners if part["id"] in migrated and (part.get("vat") or "").strip())
ours_vat = sum(1 for bp, _, _ in migrated.values() if parties.get(bp, {}).get("vat_number"))
out.append(_check("customers with a VAT number", odoo_vat, ours_vat, odoo_vat == ours_vat,
"same count" if odoo_vat == ours_vat else f"{len(p.unresolved_partners)} Odoo customer(s) have no business partner yet"))
wrong = [f"customer {bp}: expected {tr}, here {parties.get(bp, {}).get('treatment_code')}" for bp, tr, _ in migrated.values()
if parties.get(bp, {}).get("treatment_code") != tr]
out.append(_check("treatment per customer", {str(bp): tr for bp, tr, _ in migrated.values()},
{str(bp): parties.get(bp, {}).get("treatment_code") for bp, _, _ in migrated.values()}, not wrong,
"every migrated customer holds the treatment their fiscal position maps to" if not wrong else "; ".join(wrong)))
# determination parity: Odoo's tax on each posted line against what our rules would choose for the same (product, partner)
all_rules = repo.rules.all()
category_of_template = {tpl["id"]: p.handed_to_product.get(tpl["id"]) for tpl in rows.templates}
hand_edited, compared = [], 0
for m in rows.move_lines:
tax_ids = [t for t in _ids(m.get("tax_ids")) if t in p.code_of_tax]
if not tax_ids or m.get("partner_id") not in migrated:
continue
compared += 1
cat, treatment = category_of_template.get(m.get("product_tmpl_id")), migrated[m["partner_id"]][1]
rule = rules_mod.pick(all_rules, cat or "", treatment, "any", cutover) if cat else None
expected = p.code_of_tax[tax_ids[0]]
if rule is None or rule.tax_code != expected:
hand_edited.append(f"move {m.get('move_id')} line {m.get('id')} (Odoo {expected}, rules {rule.tax_code if rule else 'none'})")
out.append(_check("determination parity", compared, compared - len(hand_edited), not hand_edited,
"every posted line is taxed here as Odoo taxed it" if not hand_edited
else "explained — hand-edited in Odoo: " + "; ".join(hand_edited)))
# VAT by code and by return box over the migrated frozen tax: ar's slice stores the invoices (order: tax, then ar)
odoo_by_tax: dict = {}
for m in rows.move_lines:
if m.get("tax_line_id") in p.code_of_tax:
c = p.code_of_tax[m["tax_line_id"]]
odoo_by_tax[c] = odoo_by_tax.get(c, Decimal(0)) + abs(Decimal(str(m.get("balance") or 0)))
ours_by_tax = _frozen_vat_by_code(repo, db)
equal = {k: str(v) for k, v in odoo_by_tax.items()} == {k: str(v) for k, v in ours_by_tax.items()} and bool(ours_by_tax or not odoo_by_tax)
out.append(_check("VAT by code, last 12 months", {k: str(v) for k, v in odoo_by_tax.items()}, {k: str(v) for k, v in ours_by_tax.items()},
equal, "equal" if equal else "ar has not stored its migrated invoices yet (the order is tax, then ar): run "
"reconcile_odoo again after ar's slice, or explain a filed adjustment"))
names = _tag_names(rows)
boxes: dict = {}
for m in rows.move_lines:
for tag in _ids(m.get("tax_tag_ids")):
nm = names.get(tag, "")
mm = re.search(r"(\d+)", nm)
if mm and mm.group(1) in ("1", "6"):
amt = Decimal(str(m.get("balance") or 0)) if mm.group(1) == "1" else Decimal(str(m.get("tax_base_amount") or m.get("balance") or 0))
boxes[mm.group(1)] = boxes.get(mm.group(1), Decimal(0)) + abs(amt)
ours_boxes = _frozen_by_box(repo, db, codes)
eq = {k: str(v) for k, v in boxes.items()} == {k: str(v) for k, v in ours_boxes.items()} and bool(ours_boxes or not boxes)
out.append(_check("VAT return boxes, last filed period", {k: str(v) for k, v in boxes.items()}, {k: str(v) for k, v in ours_boxes.items()}, eq,
"equal" if eq else "ar has not stored its migrated invoices yet: box totals compare once it has; a filed adjustment is explained here"))
return out
def _frozen_lines(repo, db: str) -> list:
from sqlalchemy import select
prefix = f"odoo:{db}:"
docs = {r[0] for r in repo.conn.execute(select(tax_document.c.document_ref, tax_document.c.status))
if str(r[0]).startswith(prefix) and r[1] == "frozen"}
return [dict(r._mapping) for r in repo.conn.execute(select(tax_document_line)) if r._mapping["document_ref"] in docs]
def _frozen_vat_by_code(repo, db: str) -> dict:
out: dict = {}
for ln in _frozen_lines(repo, db):
vat = Decimal(str(ln["amount"] or 0)) * Decimal(str(ln["charged_pct"] or 0)) / 100
out[ln["tax_code"]] = out.get(ln["tax_code"], Decimal(0)) + vat
return out
def _frozen_by_box(repo, db: str, codes: dict) -> dict:
out: dict = {}
for ln in _frozen_lines(repo, db):
for b in (codes[ln["tax_code"]].reporting_box.split(",") if ln["tax_code"] in codes else []):
if b == "1":
out["1"] = out.get("1", Decimal(0)) + Decimal(str(ln["amount"] or 0)) * Decimal(str(ln["charged_pct"] or 0)) / 100
elif b == "6":
out["6"] = out.get("6", Decimal(0)) + Decimal(str(ln["amount"] or 0))
return out
def window(door) -> tuple:
end = _d(door.window_to) if door.window_to else _d(door.cutover_date)
start = _d(door.window_from) if door.window_from else end - timedelta(days=365)
return start, end