Guide d'intégration ERP

"""Synchronisation incrémentale des factures vers un logiciel comptable (journal des ventes CSV).

Incremental invoice sync to accounting software (sales journal as CSV).

Principe : on relit les factures modifiées depuis le dernier passage (filtre
``updated_after``), avec un recouvrement de 5 minutes, et on dédoublonne par
identifiant de facture : une facture déjà écrite n'est pas réécrite. La table de
correspondance des comptes est celle de VOTRE plan comptable : Tauraco ne reclasse
aucun compte. Les montants restent des chaînes décimales (``Decimal``), jamais des
flottants.

Usage : TAURACO_CLIENT_ID=… TAURACO_CLIENT_SECRET=… python3 synchro_comptable.py
"""
from __future__ import annotations

import csv
import json
import os
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from pathlib import Path
from typing import Iterable, Iterator, List

from client_credentials import get_token, iterate

STATE_FILE = Path(os.environ.get("SYNC_STATE", "etat_synchro.json"))
OUTPUT_FILE = Path(os.environ.get("SYNC_OUTPUT", "journal_ventes.csv"))
OVERLAP = timedelta(minutes=5)

#: Correspondance déclarée par l'entreprise (exemple SYSCOHADA : 411 clients, 701 ventes).
ACCOUNT_CUSTOMERS = os.environ.get("ACCOUNT_CUSTOMERS", "411100")
ACCOUNT_SALES = os.environ.get("ACCOUNT_SALES", "701100")
ACCOUNT_VAT = os.environ.get("ACCOUNT_VAT", "443100")
SYNCED_STATUSES = {"SENT", "PARTIALLY_PAID", "PAID", "OVERDUE"}


def journal_lines(invoice: dict) -> List[List[str]]:
    """Trois lignes équilibrées d'une facture : débit client, crédit ventes HT, crédit TVA."""
    total = Decimal(invoice["total"])
    subtotal = Decimal(invoice.get("subtotal") or invoice["total"])
    tax = Decimal(invoice.get("tax_total") or "0")
    if subtotal + tax != total:
        raise ValueError(f"facture {invoice['number']} : HT + taxes ≠ TTC, à vérifier")
    label = f"{invoice['number']} {invoice.get('customer_name') or ''}".strip()
    rows = [[invoice["issue_date"], "VT", invoice["number"], ACCOUNT_CUSTOMERS, label, str(total), "0"],
            [invoice["issue_date"], "VT", invoice["number"], ACCOUNT_SALES, label, "0", str(subtotal)]]
    if tax:
        rows.append([invoice["issue_date"], "VT", invoice["number"], ACCOUNT_VAT, label, "0", str(tax)])
    return rows


def new_invoices(invoices: Iterable[dict], already_synced: set) -> Iterator[dict]:
    for invoice in invoices:
        if invoice["status"] in SYNCED_STATUSES and invoice["id"] not in already_synced:
            already_synced.add(invoice["id"])
            yield invoice


def main() -> None:
    state = json.loads(STATE_FILE.read_text()) if STATE_FILE.exists() else {"since": None, "synced": []}
    started = datetime.now(timezone.utc)
    params = {"limit": "200"}
    if state["since"]:
        since = datetime.fromisoformat(state["since"].replace("Z", "+00:00")) - OVERLAP
        params["updated_after"] = since.isoformat().replace("+00:00", "Z")
    token = get_token(os.environ["TAURACO_CLIENT_ID"], os.environ["TAURACO_CLIENT_SECRET"],
                      "invoices:read", tenant=os.environ.get("TAURACO_TENANT"))["access_token"]
    synced = set(state["synced"])
    with OUTPUT_FILE.open("a", newline="", encoding="utf-8") as handle:
        writer = csv.writer(handle, delimiter=";")
        for invoice in new_invoices(iterate("/invoices", token, params), synced):
            writer.writerows(journal_lines(invoice))
    state = {"since": started.isoformat().replace("+00:00", "Z"), "synced": sorted(synced)}
    STATE_FILE.write_text(json.dumps(state))


if __name__ == "__main__":
    main()