Claude Code DTP Worker
34f202af40
Contrat pipeline commercial CRM natif (lead → visite → devis → réservation → contrat → CONFOTUR) → fixtures Frappe/ERPNext v15 : Workflow (9 états / 11 transitions) + Workflow State + Workflow Action Master + MANIFEST. Rôles gardant états/transitions résolus depuis rbac_50_roles.json (réutilisation, zéro duplication · #6) : le contrat ne cite qu'un id de rôle, jamais un nom Frappe en dur. CLI build/validate refuse d'écrire si l'un des 9 invariants de graphe casse (monotonie doc_status, atteignabilité, séparation des pouvoirs sur les étapes engageant de l'argent / clôturant). 25 tests (stdlib pur + oracle jsonschema) · job CI crm-workflow-vente-tests ajouté au gate · 146 tests de régression verts au total. Application VPS (DocType porteur OTO Dossier Vente + bench migrate) = agent ERPNext Backend, hors périmètre worker. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""Assemblage du bundle de fixtures Workflow depuis `workflow_vente_spec.json`.
|
|
|
|
Entrée : le contrat pipeline (états + transitions référençant des `role_id`
|
|
RBAC) + un `RoleResolver` (table id → nom Frappe issue 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) : tous les noms de rôle proviennent du contrat
|
|
RBAC ; toutes les valeurs d'état/transition proviennent du contrat pipeline.
|
|
Le builder n'ajoute que la structure Frappe native (flags par défaut) + un
|
|
manifeste de traçabilité.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from . import erpnext
|
|
from .rbac import RoleResolver
|
|
|
|
_UPDATE_FIELD = "statut_pipeline"
|
|
|
|
|
|
def build_bundle(spec: dict, resolver: RoleResolver) -> dict[str, Any]:
|
|
"""Transforme le contrat pipeline en bundle de fixtures Workflow Frappe."""
|
|
states_spec = spec["states"]
|
|
transitions_spec = spec["transitions"]
|
|
|
|
# --- 1. Table enfant `states` (ordre = ordre du contrat = ordre pipeline) --
|
|
state_rows: list[dict] = []
|
|
state_styles: dict[str, str] = {}
|
|
for st in states_spec:
|
|
allow_edit = resolver.erpnext_name(st["role_id"])
|
|
state_rows.append(
|
|
erpnext.document_state_row(
|
|
st["state"],
|
|
st["doc_status"],
|
|
allow_edit,
|
|
update_field=_UPDATE_FIELD,
|
|
update_value=st["update_value"],
|
|
message=st["message"],
|
|
)
|
|
)
|
|
state_styles[st["state"]] = st["style"]
|
|
|
|
# --- 2. Table enfant `transitions` (tri stable état→action) ---------------
|
|
trans_rows: list[dict] = []
|
|
for tr in transitions_spec:
|
|
allowed = resolver.erpnext_name(tr["role_id"])
|
|
trans_rows.append(
|
|
erpnext.transition_row(
|
|
tr["state"],
|
|
tr["action"],
|
|
tr["next_state"],
|
|
allowed,
|
|
allow_self_approval=tr.get("allow_self_approval", 0),
|
|
)
|
|
)
|
|
trans_rows.sort(key=lambda t: (t["state"], t["action"]))
|
|
|
|
# --- 3. Document Workflow -------------------------------------------------
|
|
workflow = erpnext.workflow_doc(
|
|
name=spec["workflow_name"],
|
|
document_type=spec["document_type"],
|
|
workflow_state_field=spec["workflow_state_field"],
|
|
is_active=spec.get("is_active", 1),
|
|
send_email_alert=spec.get("send_email_alert", 0),
|
|
override_status=spec.get("override_status", 0),
|
|
states=state_rows,
|
|
transitions=trans_rows,
|
|
)
|
|
|
|
# --- 4. Maîtres `Workflow State` + `Workflow Action Master` ---------------
|
|
# Uniques + triés → import idempotent, diff stable.
|
|
unique_states = sorted({s["state"] for s in state_rows})
|
|
workflow_state_masters = [
|
|
erpnext.workflow_state_master(s, state_styles[s]) for s in unique_states
|
|
]
|
|
unique_actions = sorted({t["action"] for t in trans_rows})
|
|
workflow_action_masters = [
|
|
erpnext.workflow_action_master(a) for a in unique_actions
|
|
]
|
|
|
|
# --- 5. Manifeste de traçabilité ------------------------------------------
|
|
roles_used = sorted({st["role_id"] for st in states_spec} | {tr["role_id"] for tr in transitions_spec})
|
|
manifest = {
|
|
"generated_from": "workflow_vente_spec.json",
|
|
"rbac_source": "rbac_50_roles.json",
|
|
"source_version": spec["version"],
|
|
"workflow_name": spec["workflow_name"],
|
|
"document_type": spec["document_type"],
|
|
"document_type_custom": bool(spec.get("document_type_custom")),
|
|
"counts": {
|
|
"states": len(state_rows),
|
|
"transitions": len(trans_rows),
|
|
"workflow_state_masters": len(workflow_state_masters),
|
|
"workflow_action_masters": len(workflow_action_masters),
|
|
"terminal_states": sum(1 for st in states_spec if st.get("is_terminal")),
|
|
},
|
|
# DocType porteur non natif : à CRÉER + confirmer sur le VPS (SPEC §7)
|
|
# avant d'importer le Workflow qui le cible. Jamais supposé existant.
|
|
"custom_doctypes_a_confirmer": (
|
|
[spec["document_type"]] if spec.get("document_type_custom") else []
|
|
),
|
|
"roles_rbac_utilises": [
|
|
{"role_id": rid, "erpnext_role_name": resolver.erpnext_name(rid)}
|
|
for rid in roles_used
|
|
],
|
|
}
|
|
|
|
return {
|
|
"manifest": manifest,
|
|
"workflow": workflow,
|
|
"workflow_state": workflow_state_masters,
|
|
"workflow_action_master": workflow_action_masters,
|
|
}
|