erp/pilotage/snapshot.py

#!/usr/bin/env python3
"""Rassemble en un seul JSON ce qu'il faut pour piloter la société.

POURQUOI CE SCRIPT EXISTE. Le tableau de bord ne doit RIEN coder en dur. Trois
sources existent déjà et font autorité, chacune sur son domaine :

  - l'ERP de production      les écritures : factures, règlements, banque
  - fleet/profile/fiscal.yaml les règles fiscales, avec leurs dates d'effet
  - fleet/profile/calendar.yaml les échéances de conformité

Un tableau qui recopierait ces chiffres deviendrait faux le lendemain. Celui-ci
les relit et produit un instantané daté, que la page consomme.

CE QU'IL NE FAIT PAS. Aucune projection, aucune hypothèse, aucun taux
d'imposition. Les scénarios se calculent DANS la page, où l'opérateur peut en
changer les paramètres — les figer ici les rendrait invérifiables.

Usage :
    python3 pilotage/snapshot.py [--out pilotage/snapshot.json]
"""
import argparse, json, os, pathlib, re, subprocess, sys, time
from datetime import date, datetime

RACINE = pathlib.Path(__file__).resolve().parent.parent
DOL = RACINE / ".claude/skills/dolibarr/scripts/dol-curl.sh"


def erp(chemin: str):
    return json.loads(subprocess.check_output([str(DOL), chemin], text=True))


def yaml_entrees(fichier: pathlib.Path, champs: tuple[str, ...]) -> list[dict]:
    """Lit les listes `- id:` du sous-ensemble YAML utilisé par le profil.

    On ne dépend pas de PyYAML : le profil est écrit dans un sous-ensemble
    strict (cf. son README), et le validateur du dépôt le lit déjà ainsi.
    """
    t = fichier.read_text(encoding="utf-8")
    out = []
    for m in re.finditer(r'^\s{2}- id: "?([\w-]+)"?', t, re.M):
        seg = t[m.start(): m.start() + 900]
        e = {"id": m.group(1)}
        for c in champs:
            v = re.search(rf'{c}: "?([^"\n]+?)"?\s*(?:#.*)?$', seg, re.M)
            e[c] = None if not v or v.group(1) == "null" else v.group(1).strip()
        out.append(e)
    return out


def main() -> int:
    a = argparse.ArgumentParser()
    a.add_argument("--out", default=str(RACINE / "pilotage/snapshot.json"))
    o = a.parse_args()

    if not DOL.exists():
        print(f"introuvable : {DOL}", file=sys.stderr)
        return 2

    # ---- produits -----------------------------------------------------------
    factures, ca_ht = [], 0.0
    for f in sorted(erp("/invoices?limit=200"), key=lambda x: x["ref"]):
        # La paire annulée FAC001-CL00001 / AVC001 se neutralise : la compter
        # gonflerait le chiffre d'affaires de 510 EUR dans les deux sens.
        annulee = f["ref"].startswith("AVC") or f["ref"] == "FAC001-CL00001"
        ht = float(f["total_ht"])
        if not annulee:
            ca_ht += ht
        factures.append({
            "ref": f["ref"], "ht": ht, "ttc": float(f["total_ttc"]),
            "devise": f["multicurrency_code"],
            "montant_devise": float(f["multicurrency_total_ht"] or 0),
            "emise": int(f["date"]), "echeance": int(f["date_lim_reglement"] or 0) or None,
            "payee": f["paye"] == "1", "statut": f["status"],
            "reste_du": float(f.get("remaintopay") or 0),
            "annulee": annulee,
        })

    # ---- charges ------------------------------------------------------------
    tiers = {t["id"]: t["name"] for t in erp("/thirdparties?limit=300")}
    fournisseurs, charges_ht = [], 0.0
    for f in sorted(erp("/supplierinvoices?limit=200"), key=lambda x: int(x["date"] or 0)):
        ht = float(f["total_ht"]); charges_ht += ht
        fournisseurs.append({"ref": f["ref"], "ht": ht, "date": int(f["date"] or 0),
                             "tiers": tiers.get(f.get("socid"), "?"), "payee": f["paye"] == "1"})

    # ---- banque et compte courant -------------------------------------------
    comptes = []
    for c in erp("/bankaccounts?limit=20"):
        lignes = erp(f"/bankaccounts/{c['id']}/lines?limit=300")
        solde = sum(float(x.get("amount") or 0) for x in lignes)
        comptes.append({"id": int(c["id"]), "label": c.get("label"),
                        "solde": round(solde, 2), "mouvements": len(lignes)})

    # Le compte courant d'associé porte un solde NÉGATIF quand la société doit :
    # on le rend explicite plutôt que de laisser la page deviner le signe.
    cca = next((c for c in comptes if "COURANT" in (c["label"] or "").upper()), None)
    dette_associe = round(-cca["solde"], 2) if cca and cca["solde"] < 0 else 0.0
    tresorerie = round(sum(c["solde"] for c in comptes if c is not cca), 2)

    snap = {
        "genere_le": datetime.now().astimezone().isoformat(timespec="seconds"),
        "exercice": {"debut": "2026-01-16", "fin": "2026-12-31"},
        "produits": {"ca_ht_facture": round(ca_ht, 2), "factures": factures},
        "charges": {"fournisseurs_ht": round(charges_ht, 2), "factures": fournisseurs},
        "tresorerie": {"total": tresorerie, "comptes": comptes,
                       "dette_compte_courant": dette_associe},
        "regles_fiscales": yaml_entrees(RACINE / "fleet/profile/fiscal.yaml",
                                        ("label", "effective_from", "effective_until", "decision")),
        "calendrier": yaml_entrees(RACINE / "fleet/profile/calendar.yaml",
                                   ("label", "due", "status", "authority")),
    }

    pathlib.Path(o.out).write_text(json.dumps(snap, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"ok — {o.out}")
    print(f"   CA facturé {snap['produits']['ca_ht_facture']:.2f} EUR · "
          f"charges {snap['charges']['fournisseurs_ht']:.2f} EUR · "
          f"trésorerie {tresorerie:.2f} EUR · dette CCA {dette_associe:.2f} EUR")
    print(f"   {len(snap['regles_fiscales'])} règles fiscales · {len(snap['calendrier'])} échéances")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())