#!/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())