"""Normalized monthly-handoff model. This is NOT a BMD import/export format. Run: python3 reconcile-export.py Uses invented document data and DEMO mappings. No Odoo/BMD connection. A real implementation must validate its actual file profile, import and records. """ from collections import Counter, defaultdict from copy import deepcopy from decimal import Decimal from hashlib import sha256 import json MONEY = ("net", "tax", "gross") FIELDS = ("company", "source_id", "revision", "document_no", "kind", "posting_period", "currency", "state", "net", "tax", "gross", "mapping_code", "document_ref") def baseline(): common = dict(company="AT-DEMO", revision=1, posting_period="2026-10", currency="EUR", state="posted", mapping_code="DEMO-MAP-1") rows = [ dict(common, source_id="SRC-001", document_no="INV-001", kind="invoice", net="100.00", tax="20.00", gross="120.00", document_ref="doc-001"), dict(common, source_id="SRC-002", document_no="INV-002", kind="invoice", net="200.00", tax="40.00", gross="240.00", document_ref="doc-002"), dict(common, source_id="SRC-003", document_no="CR-001", kind="credit", net="-50.00", tax="-10.00", gross="-60.00", document_ref="doc-003"), ] return dict(profile="DEMO-v1", company="AT-DEMO", period="2026-10", rows=rows) def identity(row): return (row["company"], row["source_id"], row["revision"]) def payload_hash(batch): # Binds this normalized model payload, NOT the bytes of a real BMD file. encoded = json.dumps(batch, sort_keys=True, separators=(",", ":")).encode() return sha256(encoded).hexdigest() def totals(rows): grouped = defaultdict(lambda: {f: Decimal("0") for f in MONEY}) for row in rows: for field in MONEY: grouped[(row["company"], row["currency"])][field] += Decimal(row[field]) return {"/".join(k): {f: format(v, ".2f") for f, v in values.items()} for k, values in sorted(grouped.items())} def full_receipt(batch): return dict(payload_hash=payload_hash(batch), documents=[dict(identity=list(identity(r)), accepted=True, target_ref="DEMO-TARGET-" + r["source_id"]) for r in batch["rows"]]) def reconcile(reference, candidate, receipt=None): findings = [] for f in ("profile", "company", "period"): if reference[f] != candidate[f]: findings.append(dict(kind="batch_scope_difference", field=f)) expected, actual = reference["rows"], candidate["rows"] for side, rows in (("reference", expected), ("candidate", actual)): for key, count in Counter(map(identity, rows)).items(): if count > 1: findings.append(dict(kind="duplicate_identity", side=side, identity=list(key), count=count)) left = {identity(r): r for r in expected} right = {identity(r): r for r in actual} for key in sorted(left.keys() - right.keys()): findings.append(dict(kind="missing_document", identity=list(key))) for key in sorted(right.keys() - left.keys()): findings.append(dict(kind="unexpected_document", identity=list(key))) for key in sorted(left.keys() & right.keys()): for f in FIELDS: a, b = left[key][f], right[key][f] if f in MONEY: a, b = Decimal(a), Decimal(b) if a != b: findings.append(dict(kind="document_difference", identity=list(key), field=f, expected=str(a), actual=str(b))) for row in actual: if not row["mapping_code"] or not row["document_ref"]: findings.append(dict(kind="missing_required_reference", identity=list(identity(row)))) if Decimal(row["net"]) + Decimal(row["tax"]) != Decimal(row["gross"]): findings.append(dict(kind="arithmetic_difference", identity=list(identity(row)))) result = dict(row_counts_match=len(expected) == len(actual), totals_match=totals(expected) == totals(actual), reference_totals=totals(expected), candidate_totals=totals(actual), findings=findings) if findings: result["decision"] = "BLOCKED" return result if receipt is None: result["decision"] = "READY_FOR_IMPORT_REHEARSAL" return result received = receipt.get("documents", []) wanted_keys = Counter(map(identity, actual)) receipt_keys = Counter(tuple(r["identity"]) for r in received) if (receipt.get("payload_hash") != payload_hash(candidate) or receipt_keys != wanted_keys or any(not r.get("accepted") or not r.get("target_ref") for r in received)): findings.append(dict(kind="destination_receipt_incomplete_or_mismatched")) result["decision"] = "RECONCILIATION_REQUIRED" else: result["decision"] = "ACKNOWLEDGED_IN_MODEL" return result def main(): source = baseline() cases = [] def run(name, mutate=lambda b: None, receipt_change=lambda r: None, expected="BLOCKED", no_receipt=False): candidate = deepcopy(source) mutate(candidate) receipt = None if no_receipt else full_receipt(candidate) if receipt is not None: receipt_change(receipt) result = reconcile(source, candidate, receipt) assert result["decision"] == expected, (name, result) cases.append(dict(case=name, expected_decision=expected, observed=result)) run("unchanged_with_complete_model_receipt", expected="ACKNOWLEDGED_IN_MODEL") run("no_destination_receipt", expected="READY_FOR_IMPORT_REHEARSAL", no_receipt=True) run("missing_credit", lambda b: b["rows"].pop()) run("duplicate_document", lambda b: b["rows"].append(deepcopy(b["rows"][0]))) def offset(batch): batch["rows"][0].update(net="101.00", gross="121.00") batch["rows"][1].update(net="199.00", gross="239.00") run("offsetting_document_amounts", offset) offset_result = cases[-1]["observed"] assert offset_result["row_counts_match"] and offset_result["totals_match"] assert len(offset_result["findings"]) == 4 run("wrong_company", lambda b: b["rows"][0].update(company="OTHER-DEMO")) run("wrong_posting_period", lambda b: b["rows"][0].update(posting_period="2026-11")) run("wrong_currency", lambda b: b["rows"][0].update(currency="USD")) run("missing_mapping", lambda b: b["rows"][0].update(mapping_code="")) run("missing_document_reference", lambda b: b["rows"][0].update(document_ref="")) run("changed_import_profile", lambda b: b.update(profile="DEMO-v2")) run("partial_destination_receipt", receipt_change=lambda r: r["documents"].pop(), expected="RECONCILIATION_REQUIRED") run("receipt_for_other_payload", receipt_change=lambda r: r.update(payload_hash="not-this-payload"), expected="RECONCILIATION_REQUIRED") output = dict(kind="executed normalized handoff model", run_on="2026-11-05", runtime="Python standard library; no Odoo or BMD connection", case_count=len(cases), all_expected_observations_matched=True, baseline=source, cases=cases, limits=[ "Not a BMD file format, connector, import or posting test.", "DEMO mapping and document references are invented labels.", "A document reference does not prove attachment contents or access.", "Reference values require independent business approval.", "Amounts demonstrate comparison, not a valid tax treatment.", "ACKNOWLEDGED_IN_MODEL is a fixture state, not production approval.", "Real imports need actual file hashes, destination records and owner sign-off.", ]) print(json.dumps(output, indent=2)) if __name__ == "__main__": main()