spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : backend.py6# Description : Generic CSV backend — the first, universal journal export target.7# =============================================================================8"""Generic CSV backend.910Lowers a CompiledJournal to flat CSV rows (one row per journal line) that any11accounting system or spreadsheet can import. Reversal emits contra rows12(sides swapped) — history is never deleted.13"""14from __future__ import annotations1516import csv17import io18from pathlib import Path1920from backends.base import (21    Backend,22    BackendCapabilities,23    PostingReceipt,24    ReversalReceipt,25    TargetPayload,26)27from core.journal import CompiledJournal, Side2829COLUMNS = [30    "entry_id", "date", "description", "account_code", "account_name",31    "account_type", "side", "amount", "currency", "memo",32    "source_event_id", "provenance_id", "policy_set", "policy_version",33    "reverses",34]353637class GenericCsvBackend(Backend):38    """Writes journals to CSV files under an output directory."""3940    def __init__(self, output_dir: str | Path = "out"):41        self.output_dir = Path(output_dir)4243    def capabilities(self) -> BackendCapabilities:44        return BackendCapabilities(45            name="generic_csv",46            posts_remotely=False,47            native_reversal=False,      # reversal = contra rows48            multi_currency=True,49            idempotency="client-side",50            notes="universal flat-file journal export",51        )5253    def _rows(self, journal: CompiledJournal, *, reverse: bool = False) -> list[list[str]]:54        rows: list[list[str]] = []55        for entry in journal.entries:56            for line in entry.lines:57                side = line.side58                if reverse:59                    side = Side.CREDIT if side is Side.DEBIT else Side.DEBIT60                rows.append([61                    ("rev_" if reverse else "") + entry.id,62                    entry.date.isoformat(),63                    ("REVERSAL: " if reverse else "") + entry.description,64                    line.account.code,65                    line.account.name,66                    line.account.type.value,67                    side.value,68                    str(line.amount.amount),69                    line.amount.currency,70                    line.memo,71                    entry.source_event_id,72                    line.provenance_id or "",73                    entry.policy_set,74                    entry.policy_version,75                    (entry.id if reverse else entry.reverses or ""),76                ])77        return rows7879    def compile(self, journal: CompiledJournal) -> TargetPayload:80        buf = io.StringIO()81        writer = csv.writer(buf)82        writer.writerow(COLUMNS)83        writer.writerows(self._rows(journal))84        return TargetPayload(backend="generic_csv", format="csv", body=buf.getvalue())8586    def post(self, payload: TargetPayload, idempotency_key: str) -> PostingReceipt:87        self.output_dir.mkdir(parents=True, exist_ok=True)88        path = self.output_dir / f"journal_{idempotency_key}.csv"89        if not path.exists():             # idempotent: same key never rewrites90            path.write_text(str(payload.body), encoding="utf-8")91        return PostingReceipt(92            backend="generic_csv",93            reference=str(path),94            idempotency_key=idempotency_key,95        )9697    def reverse(self, receipt: PostingReceipt) -> ReversalReceipt:98        original = Path(receipt.reference)99        reader = csv.reader(io.StringIO(original.read_text(encoding="utf-8")))100        rows = list(reader)101        header, body = rows[0], rows[1:]102        side_i, entry_i, desc_i, rev_i = (103            header.index("side"), header.index("entry_id"),104            header.index("description"), header.index("reverses"),105        )106        for row in body:107            row[rev_i] = row[entry_i]108            row[entry_i] = "rev_" + row[entry_i]109            row[desc_i] = "REVERSAL: " + row[desc_i]110            row[side_i] = "credit" if row[side_i] == "debit" else "debit"111        buf = io.StringIO()112        writer = csv.writer(buf)113        writer.writerow(header)114        writer.writerows(body)115        path = original.with_name(original.stem + "_reversal.csv")116        path.write_text(buf.getvalue(), encoding="utf-8")117        return ReversalReceipt(118            backend="generic_csv",119            reference=str(path),120            reversed_reference=receipt.reference,121        )122