oto-enterprise-os-dtp/05_deliverables_mvp/crm/commissions/commissions_gen.py

#!/usr/bin/env python3
"""Générateur du barème de commissions vendeurs · Sprint 4 · ERPNext Backend.

Roadmap ligne 51 : « commissions vendeurs auto ». Produit un plan de commissions
cross-cohérent avec les trois contrats CRM déjà livrés :
  - le pipeline vente  (`workflow_vente_spec.json`)   → quels évènements paient ;
  - le DocType porteur (`dossier_vente/doctype_spec.json`) → sur quel champ ;
  - le contrat RBAC    (`rbac_50_roles.json`)          → quel rôle touche.

Ce worker n'écrit JAMAIS sur le VPS (#8) : il émet les fichiers de hand-off ;
l'application réelle (création du champ commission / calcul en production) reste
côté agent ERPNext Backend.

ANTI-INVENTION (#6) : aucun taux de commission n'est documenté dans CLAUDE.md, et
aucun n'est fabriqué ici. Chaque évènement porte `taux_pct: null` tant que la
Direction ne l'a pas confirmé (avec sa source). Le calcul `commission = base ×
taux` (commlib/finance.py) est traçable : il reste `None` tant qu'un opérande
manque — jamais 0-inventé.

Sous-commandes :
  build    [-o OUT]  → écrit commission_plan.json + MANIFEST.json
  validate           → (re)génère en mémoire, valide schéma + invariants de
                       cross-cohérence barème↔workflow↔DocType↔RBAC ; sort en
                       erreur sinon.

Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable.
"""

from __future__ import annotations

import argparse
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_CRM = os.path.normpath(os.path.join(_HERE, ".."))          # crm/
_DELIVERABLES = os.path.normpath(os.path.join(_CRM, ".."))  # 05_deliverables_mvp/

sys.path.insert(0, _HERE)
sys.path.insert(0, _CRM)
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))

from commlib import builder, finance  # noqa: E402
from workflow_vente.wflib.rbac import RoleResolver  # noqa: E402
from lib import validator as maison  # type: ignore  # noqa: E402

_SPEC_PATH = os.path.join(_HERE, "bareme_spec.json")
_WF_SPEC_PATH = os.path.join(_CRM, "workflow_vente", "workflow_vente_spec.json")
_DT_SPEC_PATH = os.path.join(_CRM, "dossier_vente", "doctype_spec.json")
_SCHEMA_PATH = os.path.join(_HERE, "bareme.schema.json")
_DEFAULT_OUT = os.path.join(_HERE, "out")


def _eprint(*args) -> None:
    print(*args, file=sys.stderr)


def _load(path: str) -> dict:
    with open(path, encoding="utf-8") as fh:
        return json.load(fh)


def _write_json(path: str, data) -> None:
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2)
        fh.write("\n")


def _currency_fields(dt_spec: dict) -> set[str]:
    """Champs Currency du DocType Dossier Vente (bases de commission légitimes)."""
    out: set[str] = set()
    for grp in dt_spec.get("field_groups", []):
        for f in grp.get("fields", []):
            if f.get("fieldtype") == "Currency":
                out.add(f["fieldname"])
    return out


def _devise_field(dt_spec: dict) -> dict | None:
    for grp in dt_spec.get("field_groups", []):
        for f in grp.get("fields", []):
            if f["fieldname"] == "devise":
                return f
    return None


def _build() -> tuple[dict, dict, dict, dict, RoleResolver]:
    spec = _load(_SPEC_PATH)
    wf_spec = _load(_WF_SPEC_PATH)
    dt_spec = _load(_DT_SPEC_PATH)
    resolver = RoleResolver.from_path()
    bundle = builder.build_bundle(spec, resolver)
    return bundle, spec, wf_spec, dt_spec, resolver


def _validate(bundle: dict, spec: dict, wf_spec: dict, dt_spec: dict,
              resolver: RoleResolver) -> list[str]:
    """Schéma de sortie + invariants de cross-cohérence (les 4 contrats)."""
    schema = _load(_SCHEMA_PATH)
    errors = list(maison.validate(bundle, schema))

    plan = bundle["commission_plan"]
    m = bundle["manifest"]
    events = plan["evenements"]

    # Contexte dérivé des contrats voisins.
    wf_update_values = {s["update_value"] for s in wf_spec["states"]}
    submitted_values = {s["update_value"] for s in wf_spec["states"]
                        if s["doc_status"] == "1"}
    currency_fields = _currency_fields(dt_spec)

    seen: set[tuple] = set()
    for ev in events:
        tag = f"{ev['update_value']}/{ev['role_id']}"

        # 1 · update_value existe dans le workflow vente (anti-dérive).
        if ev["update_value"] not in wf_update_values:
            errors.append(f"[{tag}] update_value absent du workflow vente")
        # 2 · commission uniquement sur un état SOUMIS (doc_status=1) — jamais
        #     sur un brouillon (lead/visite/devis/abandonné).
        elif ev["update_value"] not in submitted_values:
            errors.append(f"[{tag}] update_value n'est pas un état soumis "
                          f"(doc_status≠1) — pas de commission sur brouillon")
        # 3 · base_field est un champ Currency réel du DocType Dossier Vente.
        if ev["base_field"] not in currency_fields:
            errors.append(f"[{tag}] base_field {ev['base_field']!r} n'est pas un "
                          f"champ Currency du DocType Dossier Vente")
        # 4 · rôle résolu + portail ventes (commission = concern ventes).
        if resolver.portail(ev["role_id"]) != "ventes":
            errors.append(f"[{tag}] role_id hors portail ventes "
                          f"({resolver.portail(ev['role_id'])!r})")
        # 5 · nom de rôle Frappe cohérent avec la résolution RBAC.
        if ev["erpnext_role_name"] != resolver.erpnext_name(ev["role_id"]):
            errors.append(f"[{tag}] erpnext_role_name incohérent avec RBAC")
        # 6 · ANTI-INVENTION (#6) : pas de taux sans source. Soit à confirmer
        #     (taux null), soit taux fourni AVEC sa source.
        if ev["taux_pct"] is None:
            if not ev["a_confirmer"]:
                errors.append(f"[{tag}] taux null mais a_confirmer=false")
        else:
            if not finance.is_filled(ev["source"]):
                errors.append(f"[{tag}] taux_pct fixé sans `source` — chiffre "
                              f"non sourcé (interdit #6)")
        # 7 · unicité (update_value, role_id).
        key = (ev["update_value"], ev["role_id"])
        if key in seen:
            errors.append(f"[{tag}] évènement dupliqué (update_value, role_id)")
        seen.add(key)

    # 8 · devise_field == champ `devise` (Select USD/DOP · #10) du DocType.
    devf = _devise_field(dt_spec)
    if plan["devise_field"] != "devise":
        errors.append("devise_field doit être 'devise' (champ du Dossier Vente)")
    if devf is None:
        errors.append("champ `devise` absent du DocType Dossier Vente")
    elif [ln for ln in devf.get("options", "").split("\n") if ln] != ["USD", "DOP"]:
        errors.append("options du champ `devise` ≠ USD/DOP (#10)")

    # 9 · rien perdu : autant d'évènements en sortie qu'en entrée.
    if len(events) != len(spec["evenements"]):
        errors.append("nombre d'évènements en sortie ≠ contrat barème")

    # 10 · comptes du manifeste cohérents.
    if m["counts"]["evenements"] != len(events):
        errors.append("counts.evenements incohérent")
    if m["counts"]["roles"] != len({e["role_id"] for e in events}):
        errors.append("counts.roles incohérent")
    a_conf = sum(1 for e in events if e["a_confirmer"] or e["taux_pct"] is None)
    if m["counts"]["taux_a_confirmer"] != a_conf:
        errors.append("counts.taux_a_confirmer incohérent")

    return errors


def cmd_build(args: argparse.Namespace) -> int:
    bundle, spec, wf_spec, dt_spec, resolver = _build()
    errors = _validate(bundle, spec, wf_spec, dt_spec, resolver)
    if errors:
        _eprint("❌ Bundle invalide — génération refusée (anti-régression) :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1

    out = os.path.abspath(args.out)
    os.makedirs(out, exist_ok=True)
    _write_json(os.path.join(out, "commission_plan.json"), bundle["commission_plan"])
    _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])

    m = bundle["manifest"]
    print(f"✅ Plan de commissions généré dans {out}")
    print(f"   commission_plan.json : {m['counts']['evenements']} évènements · "
          f"{m['counts']['roles']} rôles · {m['counts']['taux_a_confirmer']} taux à confirmer")
    print("   ⚠ Taux réels + champ commission côté ERPNext Backend (Direction "
          "renseigne taux_pct + source · VPS · #8).")
    return 0


def cmd_validate(args: argparse.Namespace) -> int:
    bundle, spec, wf_spec, dt_spec, resolver = _build()
    errors = _validate(bundle, spec, wf_spec, dt_spec, resolver)
    if errors:
        _eprint("❌ Validation KO :")
        for e in errors:
            _eprint(f"   - {e}")
        return 1
    m = bundle["manifest"]
    print(f"✅ Validation OK — barème {m['bareme_name']!r} : "
          f"{m['counts']['evenements']} évènements, schéma + 10 invariants verts.")
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="Générateur du barème de commissions vendeurs.")
    sub = p.add_subparsers(dest="cmd", required=True)

    pb = sub.add_parser("build", help="génère commission_plan.json / MANIFEST.json")
    pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (défaut: ./out)")
    pb.set_defaults(func=cmd_build)

    pv = sub.add_parser("validate", help="valide le bundle (schéma + 10 invariants) sans écrire")
    pv.set_defaults(func=cmd_validate)

    args = p.parse_args(argv)
    return args.func(args)


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