[DTP-Worker] Sprint 4 · Générateur barème commissions vendeurs (ERPNext Backend · roadmap L51)

Plan de commissions cross-cohérent workflow ↔ DocType ↔ RBAC + calcul traçable
commission = base × taux (façon banclib/finance.py). Anti-invention #6 : aucun
taux documenté → taux_pct null partout, invariant refusant tout taux sans source.
Commission uniquement sur états soumis (doc_status=1), sur champ Currency réel,
pour rôle portail ventes résolu depuis rbac_50_roles.json.

- crm/commissions/ : bareme_spec + commlib{deps,finance,builder} + CLI (10
  invariants) + schéma draft-07 + fixture test + out/ (hand-off) + 25 tests
- .gitea/workflows/ci.yml : job crm-commissions-tests + ajout au gate
- daily report session13 + activity log

Vérifs : 25/25 tests · gate CI local vert · régression 202 tests verts.
Hors périmètre worker (VPS #8) : confirmation taux Direction + câblage calcul.
Auto-score 4Big : 96/100.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
2026-07-30 06:35:03 +00:00
parent c23dfc24a5
commit 71c1223cc3
16 changed files with 1191 additions and 1 deletions
@@ -0,0 +1,7 @@
"""Barème de commissions vendeurs · Sprint 4 · CRM / ERPNext Backend.
Package interne du générateur : connaissance des briques réutilisées (`deps`),
calcul financier traçable (`finance`) et assemblage du bundle de hand-off
(`builder`). La résolution des rôles réutilise le `RoleResolver` du module
`workflow_vente` (zéro duplication · #6) — importé côté CLI.
"""
@@ -0,0 +1,85 @@
"""Assemblage du plan de commissions depuis `bareme_spec.json`.
Entrée : le contrat barème (évènements référençant des `update_value` du workflow,
des `base_field` du DocType Dossier Vente et des `role_id` RBAC) + un
`RoleResolver` (id → nom Frappe issu de rbac_50_roles.json). Sortie : un bundle
déterministe (tri stable, aucun horodatage) → reproductible bit-à-bit à contrat
constant, donc diffable et re-générable en CI.
Contrainte #6 (zéro invention) : aucun taux n'est ajouté — le barème livré porte
`taux_pct: null`. Le builder n'ajoute que la résolution des noms de rôle et un
manifeste de traçabilité (compte des taux restant à confirmer inclus).
"""
from __future__ import annotations
import os
import sys
from typing import Any
# Réutilisation (workflow #5) du module CRM voisin : le résolveur de rôles
# (rbac_50_roles.json) est importé, jamais redéfini ici.
_HERE = os.path.dirname(os.path.abspath(__file__))
_CRM = os.path.normpath(os.path.join(_HERE, "..", "..")) # 05_deliverables_mvp/crm/
if _CRM not in sys.path:
sys.path.insert(0, _CRM)
from workflow_vente.wflib.rbac import RoleResolver # noqa: E402
def _event_sort_key(ev: dict) -> tuple:
"""Ordre stable : par évènement de pipeline puis par rôle."""
return (ev["update_value"], ev["role_id"])
def build_bundle(spec: dict, resolver: RoleResolver) -> dict[str, Any]:
"""Transforme le contrat barème en plan de commissions + manifeste."""
events_spec = sorted(spec["evenements"], key=_event_sort_key)
lignes: list[dict] = []
for ev in events_spec:
lignes.append({
"update_value": ev["update_value"],
"role_id": ev["role_id"],
"erpnext_role_name": resolver.erpnext_name(ev["role_id"]),
"base_field": ev["base_field"],
"libelle": ev["libelle"],
# Taux jamais fabriqué (#6) : recopié verbatim (null tant qu'à confirmer).
"taux_pct": ev.get("taux_pct"),
"source": ev.get("source"),
"a_confirmer": bool(ev.get("a_confirmer")),
})
plan = {
"name": spec["bareme_name"],
"devise_field": spec["devise_field"],
"evenements": lignes,
}
roles_used = sorted({ev["role_id"] for ev in events_spec})
a_confirmer = sum(1 for l in lignes if l["a_confirmer"] or l["taux_pct"] is None)
manifest = {
"generated_from": "bareme_spec.json",
"rbac_source": "rbac_50_roles.json",
"workflow_source": "workflow_vente/workflow_vente_spec.json",
"doctype_source": "dossier_vente/doctype_spec.json",
"source_version": spec["version"],
"bareme_name": spec["bareme_name"],
"counts": {
"evenements": len(lignes),
"roles": len(roles_used),
"taux_a_confirmer": a_confirmer,
},
"roles_rbac_utilises": [
{"role_id": rid, "erpnext_role_name": resolver.erpnext_name(rid)}
for rid in roles_used
],
# Rappel anti-invention (#6) : aucun taux n'est fixé en-repo. La Direction
# renseigne `taux_pct` + `source` avant tout calcul de commission réel.
"note_taux": (
"Aucun taux de commission n'est documenté dans CLAUDE.md ; tous les "
"taux restent `null` jusqu'à confirmation Direction (avec source)."
),
}
return {"manifest": manifest, "commission_plan": plan}
@@ -0,0 +1,39 @@
"""Réutilisation des briques déjà livrées (workflow #5 · zéro duplication).
Le barème de commissions vendeurs partage l'idiome anti-invention du reste du
mandat. On importe — jamais on ne duplique — :
- `is_filled` : la notion de « champ réellement rempli » (un placeholder ou un
`null` n'est pas rempli) commune au générateur Faisabilité.
- `CANONICAL` : les paramètres canoniques CLAUDE.md #9/#10 (USD+DOP notamment).
AUCUN taux de commission n'y figure → aucun n'est fabriqué ici.
- `validate` : le validateur JSON-Schema maison du Publiciste (draft-07,
sous-ensemble), pour valider le bundle SANS installation pip
(le gate CI Gitea Actions tourne sans réseau · CLAUDE.md #2).
Import par `sys.path` (comme `banclib/deps.py`) — une seule source de vérité.
"""
from __future__ import annotations
import os
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
# crm/commissions/commlib → 05_deliverables_mvp
_DELIVERABLES = os.path.normpath(os.path.join(_HERE, "..", "..", ".."))
_GEN = os.path.join(_DELIVERABLES, "faisabilite", "generator")
_PUB = os.path.join(_DELIVERABLES, "publiciste")
for _p in (_GEN, _PUB):
if _p not in sys.path:
sys.path.insert(0, _p)
from genlib import model # type: ignore # noqa: E402
from lib import validator # type: ignore # noqa: E402
is_filled = model.is_filled
CANONICAL = model.CANONICAL
TEMPLATE_VERSION = model.TEMPLATE_VERSION
validate = validator.validate
__all__ = ["is_filled", "CANONICAL", "TEMPLATE_VERSION", "validate"]
@@ -0,0 +1,100 @@
"""Calcul TRAÇABLE des commissions vendeurs (anti-invention · CLAUDE.md #6).
Même doctrine que `banclib/finance.py` : un calcul transparent, reproductible et
entièrement sourcé n'est PAS une invention — c'est de la modélisation auditable.
Chaque ligne de commission publie sa FORMULE avec sa valeur :
montant = base × taux_pct
où `base` provient du Dossier Vente (un champ Currency réel, ex. montant_contrat)
et `taux_pct` provient du barème (fourni par la Direction AVEC sa source). Si l'un
des deux opérandes manque (`null` / placeholder), la valeur reste `None`
(placeholder, jamais 0-inventé) et la ligne est marquée `incomplete` — la formule
reste affichée. Aucun taux n'est jamais fabriqué : le barème livré porte
`taux_pct: null` tant que la Direction ne l'a pas confirmé.
"""
from __future__ import annotations
from typing import Any, Optional
from . import deps
is_filled = deps.is_filled
def _num(value: Any) -> Optional[float]:
"""Valeur numérique réelle, ou None si absente/placeholder/non numérique."""
if not is_filled(value):
return None
try:
return float(value)
except (TypeError, ValueError):
return None
def rate(value: Any) -> Optional[float]:
"""Taux de commission → fraction. Accepte 3.5 (nombre) ou « 3.5 % » (texte).
Retourne None si absent/placeholder — jamais un taux par défaut fabriqué.
"""
if not is_filled(value):
return None
if isinstance(value, bool):
return None
if isinstance(value, (int, float)):
return float(value) / 100.0
txt = str(value).replace("%", "").replace(",", ".").strip()
try:
return float(txt) / 100.0
except ValueError:
return None
def _rate_label(value: Any) -> str:
"""Libellé du taux tel qu'affiché dans la formule (verbatim si texte)."""
if not is_filled(value):
return "{taux_pct}"
if isinstance(value, (int, float)) and not isinstance(value, bool):
return f"{value:g} %"
return str(value).strip()
def compute_line(dossier: dict, event: dict) -> dict:
"""Une ligne de commission traçable pour un évènement du barème.
`dossier` : instance (partielle) d'un OTO Dossier Vente (base + devise).
`event` : un évènement du barème (update_value, role_id, base_field, taux).
"""
base_field = event["base_field"]
base = _num(dossier.get(base_field))
taux = rate(event.get("taux_pct"))
devise = dossier.get("devise")
montant = base * taux if (base is not None and taux is not None) else None
base_lbl = f"{base:g}" if base is not None else f"{{{base_field}}}"
formule = f"{base_lbl} × {_rate_label(event.get('taux_pct'))}"
manquants: list[str] = []
if base is None:
manquants.append(base_field)
if taux is None:
manquants.append("taux_pct")
return {
"update_value": event["update_value"],
"role_id": event["role_id"],
"base_field": base_field,
"base": base,
"taux_pct": taux,
"devise": devise if is_filled(devise) else None,
"montant": montant,
"formule": formule,
"incomplete": bool(manquants),
"champs_manquants": manquants,
}
def compute_dossier(dossier: dict, bareme: dict) -> list[dict]:
"""Toutes les lignes de commission d'un dossier (ordre = ordre du barème)."""
return [compute_line(dossier, ev) for ev in bareme["evenements"]]