# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : mapper.py # Description : Lower a CompiledJournal to QuickBooks Online JournalEntry payloads (pure, offline). # ============================================================================= """CompiledJournal -> QBO JournalEntry payloads. Pure lowering, no network: this module runs identically with or without a QuickBooks account, so the whole backend is testable offline. Shape per docs/research/erp-apis.md (QBO API, minorversion=75 baseline): - Line[] with unsigned Amount + PostingType ("Debit"/"Credit") + AccountRef - DocNumber max 21 chars (longer ids are truncated with a stable hash suffix) - amounts stay Decimal end-to-end here; serialization to exact JSON numbers is the transport's job (never through float). """ from __future__ import annotations import hashlib from typing import Any from core.journal import CompiledJournal, JournalEntry, Side DOCNUMBER_MAX = 21 class AccountMappingError(KeyError): """An account code has no QuickBooks AccountRef mapping.""" def doc_number(entry_id: str) -> str: """QBO DocNumber (<= 21 chars), stable for a given entry id.""" if len(entry_id) <= DOCNUMBER_MAX: return entry_id digest = hashlib.sha256(entry_id.encode("utf-8")).hexdigest()[:6] return f"{entry_id[:DOCNUMBER_MAX - 7]}-{digest}" def map_entry( entry: JournalEntry, account_map: dict[str, dict[str, str]] | None = None, ) -> dict[str, Any]: """One JournalEntry -> one QBO JournalEntry create body. account_map: our account code -> {"value": , "name": ...}. When None, the code itself is used as the AccountRef value (fine for the mock transport and for offline export; a real realm needs the map). """ lines: list[dict[str, Any]] = [] for line in entry.lines: if account_map is not None: try: ref = account_map[line.account.code] except KeyError: raise AccountMappingError( f"account '{line.account.code} {line.account.name}' has no " "QuickBooks AccountRef in the account map; add it or post " "to the native backend instead" ) from None else: ref = {"value": line.account.code, "name": line.account.name} lines.append({ "Description": line.memo, "Amount": line.amount.amount, # Decimal, exact "DetailType": "JournalEntryLineDetail", "JournalEntryLineDetail": { "PostingType": "Debit" if line.side is Side.DEBIT else "Credit", "AccountRef": dict(ref), }, }) return { "DocNumber": doc_number(entry.id), "TxnDate": entry.date.isoformat(), "PrivateNote": ( f"{entry.description} | AIR event {entry.source_event_id} | " f"policies {entry.policy_set}@{entry.policy_version}" ), "Line": lines, "CurrencyRef": {"value": entry.lines[0].amount.currency}, } def map_journal( journal: CompiledJournal, account_map: dict[str, dict[str, str]] | None = None, ) -> list[dict[str, Any]]: return [map_entry(entry, account_map) for entry in journal.entries] def contra_body(body: dict[str, Any]) -> dict[str, Any]: """Reversal payload: QBO has no native journal-entry reversal, so we synthesize an exact contra entry (sides swapped, amounts identical).""" contra = { "DocNumber": doc_number("R" + body["DocNumber"]), "TxnDate": body["TxnDate"], "PrivateNote": "REVERSAL of " + body["DocNumber"] + " | " + body["PrivateNote"], "Line": [], "CurrencyRef": dict(body["CurrencyRef"]), } for line in body["Line"]: detail = line["JournalEntryLineDetail"] contra["Line"].append({ "Description": "reversal: " + line["Description"], "Amount": line["Amount"], "DetailType": "JournalEntryLineDetail", "JournalEntryLineDetail": { "PostingType": ("Credit" if detail["PostingType"] == "Debit" else "Debit"), "AccountRef": dict(detail["AccountRef"]), }, }) return contra