[DTP-Worker] Sprint 5 · Générateur DocType CONFOTUR Application (ONAPI/Legal · roadmap L55)

DocType custom porteur des dossiers d'incitation touristique CONFOTUR (Ley
158-01, RD), référencé par RBAC (3 rôles) et le workflow vente mais jamais
généré. Permissions = permissions_cibles RBAC mot pour mot ; is_submittable
déduit de l'action submit ; estado/dossier_vente dérivés du workflow ; aucun
taux/loi/montant fabriqué (#6). 14 invariants, 44 tests, hand-off dans out/.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
2026-07-30 08:09:41 +00:00
parent c0d2e21ef3
commit 35a20247f8
15 changed files with 1810 additions and 1 deletions
@@ -0,0 +1,14 @@
"""Librairie du generateur DocType `CONFOTUR Application` (Sprint 5 · ONAPI/Legal).
Trois briques :
- `frappe` : connaissance Frappe/ERPNext v15 (types de champ, DocPerm,
enveloppe DocType) — separee de l'assemblage metier.
- `rbac_scan` : lecture (jamais reecriture) des roles du contrat RBAC qui
referencent le DocType `CONFOTUR Application` -> source unique
du nom, des permissions et du caractere soumissible.
- `builder` : assemblage deterministe du bundle {manifest, doctype}.
"""
from __future__ import annotations
__all__ = ["frappe", "rbac_scan", "builder"]
@@ -0,0 +1,193 @@
"""Assemblage du fixture DocType `CONFOTUR Application` — cross-coherent RBAC.
Entrees :
- `confotur_spec.json` : structure metier (libelles + types de champ).
- `rbac_50_roles.json` : source UNIQUE du nom du DocType, des permissions
et du caractere soumissible (roles referencant `CONFOTUR Application`).
- `workflow_vente_spec.json` : le pipeline vente (source des options `estado`,
du DocType lie `dossier_vente`, et des evenements de depot CONFOTUR).
- `RoleResolver` (rbac) : `role_id -> erpnext_role_name` + `portail`.
Sortie : un bundle deterministe `{manifest, doctype}` reproductible bit-a-bit.
Anti-invention / anti-derive (#6, workflow #5) :
- Le NOM du DocType est l'identite de ce generateur ; un invariant PROUVE qu'au
moins un role RBAC le reference et que tous le marquent `custom` (=> custom=1).
- Les PERMISSIONS sont, mot pour mot, les `permissions_cibles` du contrat RBAC —
ni ajout ni retrait. Renommer/retirer un role cote RBAC se propage ici.
- `is_submittable` est DEDUIT de la presence d'une action `submit` cote RBAC.
- Les options de `dossier_vente`/`estado` sont DERIVEES du workflow (jamais en
dur) ; celles de `entite_porteuse` viennent des entites CLAUDE.md du spec.
- AUCUN chiffre d'incitation, article de loi, montant ou reference d'autorite :
ces champs restent vides, renseignes cote VPS par l'agent ONAPI.
"""
from __future__ import annotations
from typing import Any
from . import frappe, rbac_scan
# Identite du generateur : le DocType custom qu'il produit. Un invariant verifie
# que le contrat RBAC le reference reellement (sinon le livrable n'a pas lieu).
TARGET_DOCTYPE = "CONFOTUR Application"
# Drapeaux de docfield que le spec metier peut porter (le reste = mise en page).
_SPEC_FLAGS = ("reqd", "read_only", "in_list_view", "in_standard_filter", "hidden", "bold")
def confotur_state_values(wf_spec: dict) -> list[str]:
"""`update_value` des etats CONFOTUR du workflow, dans l'ordre du pipeline."""
return [s["update_value"] for s in wf_spec["states"]
if s["update_value"].startswith("confotur")]
def _resolve_options(f: dict, spec: dict, wf_spec: dict) -> str | None:
"""Resout `options` litterales OU derivees (`options_source`)."""
src = f.get("options_source")
if src is None:
return f.get("options")
if src == "workflow_document_type":
return wf_spec["document_type"]
if src == "entites":
return "\n".join(spec["entites"])
if src == "workflow_confotur_states":
# Select facultatif -> ligne vide en tete (valeur non imposee a la creation).
return "\n" + "\n".join(confotur_state_values(wf_spec))
raise ValueError(f"options_source inconnu pour {f['fieldname']!r} : {src!r}")
def _spec_docfield(f: dict, spec: dict, wf_spec: dict) -> dict[str, Any]:
flags = {k: f[k] for k in _SPEC_FLAGS if k in f}
return frappe.docfield(
f["fieldname"],
f["fieldtype"],
label=f.get("label"),
options=_resolve_options(f, spec, wf_spec),
default=f.get("default"),
flags=flags,
)
def _depot_events(wf_spec: dict, resolver) -> list[dict[str, Any]]:
"""Evenements de depot CONFOTUR derives des transitions du workflow vente.
Ce sont les points de cablage « depots automatiques » (roadmap L55) : quelle
transition depose / approuve, par quel role. Aucun evenement fabrique — ils
sortent tous du contrat pipeline.
"""
val_by_state = {s["state"]: s["update_value"] for s in wf_spec["states"]}
events: list[dict[str, Any]] = []
for t in wf_spec["transitions"]:
uv = val_by_state.get(t["next_state"], "")
if not uv.startswith("confotur"):
continue
events.append({
"action": t["action"],
"next_state": t["next_state"],
"update_value": uv,
"role_id": t["role_id"],
"erpnext_role_name": resolver.erpnext_name(t["role_id"]),
"portail": resolver.portail(t["role_id"]),
})
events.sort(key=lambda e: (e["update_value"], e["role_id"]))
return events
def build_bundle(spec: dict, wf_spec: dict, contract: dict, resolver) -> dict[str, Any]:
"""Transforme les contrats en bundle `{manifest, doctype}` deterministe."""
dt_cfg = spec["doctype"]
# --- 1. Roles RBAC visant le DocType (source du nom / permissions) ---------
targeting = rbac_scan.roles_targeting(TARGET_DOCTYPE, contract)
if not targeting:
raise ValueError(
f"Aucun role RBAC ne reference {TARGET_DOCTYPE!r} — livrable sans objet."
)
is_submittable = 1 if any("submit" in r["actions"] for r in targeting) else 0
# --- 2. Champs de pilotage (naming series injectee, non re-saisie) ---------
fields: list[dict] = [frappe.naming_series_field(dt_cfg["naming_series"])]
# --- 3. Champs metier (spec) : une Section Break par groupe ----------------
linked_doctypes: set[str] = set()
for gi, group in enumerate(spec["field_groups"]):
slug = "sb_" + "".join(
c if (c.isascii() and c.isalnum()) else "_" for c in group["section"].lower()
).strip("_")
fields.append(frappe.section_break(f"{slug}_{gi}", group["section"]))
for f in group["fields"]:
row = _spec_docfield(f, spec, wf_spec)
fields.append(row)
if row["fieldtype"] == "Link" and row.get("options"):
linked_doctypes.add(row["options"])
# --- 4. Permissions = permissions_cibles RBAC (mot pour mot) ---------------
permissions = [
frappe.permission_row(resolver.erpnext_name(r["role_id"]), set(r["actions"]))
for r in sorted(targeting, key=lambda r: resolver.erpnext_name(r["role_id"]))
]
# --- 5. Document DocType ---------------------------------------------------
doctype = frappe.doctype_doc(
name=TARGET_DOCTYPE,
module=dt_cfg["module"],
is_submittable=is_submittable,
title_field=dt_cfg["title_field"],
search_fields=dt_cfg["search_fields"],
track_changes=dt_cfg.get("track_changes", 1),
track_seen=dt_cfg.get("track_seen", 0),
fields=fields,
permissions=permissions,
)
# --- 6. Evenements de depot derives du workflow ----------------------------
depot_events = _depot_events(wf_spec, resolver)
# --- 7. Manifeste de tracabilite -------------------------------------------
data_fields = [f for f in fields if f["fieldtype"] not in frappe.LAYOUT_FIELDTYPES]
sections = [f for f in fields if f["fieldtype"] == "Section Break"]
manifest = {
"generated_from": "confotur_spec.json",
"rbac_source": "rbac_50_roles.json",
"workflow_source": "crm/workflow_vente/workflow_vente_spec.json",
"source_version": spec["version"],
"rbac_version": contract["version"],
"workflow_version": wf_spec["version"],
"doctype_name": TARGET_DOCTYPE,
"custom": bool(doctype["custom"]),
"is_submittable": bool(is_submittable),
"counts": {
"fields": len(fields),
"data_fields": len(data_fields),
"sections": len(sections),
"permissions": len(permissions),
"roles": len(targeting),
"depot_events": len(depot_events),
},
"module_a_confirmer": dt_cfg["module"],
"doctypes_lies_a_confirmer": sorted(linked_doctypes),
"roles_rbac_utilises": [
{
"role_id": r["role_id"],
"erpnext_role_name": resolver.erpnext_name(r["role_id"]),
"portail": resolver.portail(r["role_id"]),
"permissions": sorted(r["actions"]),
"custom": r["custom"],
}
for r in targeting
],
"depot_events": depot_events,
"note_anti_invention": (
"Aucun taux d'incitation, article de loi, montant ni reference "
"d'autorite CONFOTUR n'est encode (#6). Ces valeurs restent vides "
"jusqu'a confirmation de l'agent ONAPI/Legal depuis data_room P05/P07."
),
"hand_off_vps": (
"Creer le module Frappe 'OTOV7 CONFOTUR' puis importer ce DocType "
"(bench migrate / import-fixtures) AVANT de cabler les evenements de "
"depot du Workflow 'OTO Vente Pipeline' (crm/workflow_vente/out/)."
),
}
return {"manifest": manifest, "doctype": doctype}
@@ -0,0 +1,151 @@
"""Modele Frappe/ERPNext v15 : structure native du fixture `DocType`.
Separe la CONNAISSANCE FRAPPE (types de champ legitimes, drapeaux de docfield,
modele DocPerm, enveloppe du document DocType) de l'assemblage metier
(`builder.py`). Contrainte #1 « ERPNext natif = priorite absolue » : on produit
le DocType standard du moteur Frappe, porteur des dossiers d'incitation CONFOTUR.
Aucun acces VPS : chaque fonction renvoie un dict serialisable que l'agent
ONAPI/Legal importera via `bench` (hand-off · #8).
Nota vs `crm/dossier_vente/dvlib/frappe.py` : ce module ajoute la permission
native `report` (le role `legal-directeur` la porte cote RBAC). Les deux modules
restent volontairement independants — chaque livrable possede sa connaissance
Frappe, on n'ouvre que ce que le contrat emploie reellement.
"""
from __future__ import annotations
from typing import Any
# Types de champ Frappe employes par ce DocType. Hors de cet ensemble = invention
# -> refusee par `docfield()`.
VALID_FIELDTYPES: frozenset[str] = frozenset(
{
"Section Break",
"Column Break",
"Data",
"Select",
"Link",
"Date",
"Datetime",
"Small Text",
"Text",
"Check",
}
)
# Types « de mise en page » : ni requis, ni porteurs de donnee.
LAYOUT_FIELDTYPES: frozenset[str] = frozenset({"Section Break", "Column Break"})
# Permissions natives du child table `DocPerm` d'un DocType Frappe (v15).
# `report` est inclus (porte par le role legal-directeur cote RBAC).
VALID_PERMS: frozenset[str] = frozenset(
{"read", "write", "create", "submit", "cancel", "amend", "delete",
"print", "email", "report", "import", "export", "share"}
)
# Ordre de serialisation stable des drapeaux DocPerm (diff propre).
_PERM_ORDER = ("read", "write", "create", "submit", "cancel", "amend", "delete",
"print", "email", "report", "import", "export", "share")
# Drapeaux de docfield connus (defaut 0) — on ne serialise que ceux a 1.
_DOCFIELD_FLAGS = ("reqd", "read_only", "in_list_view", "in_standard_filter", "hidden", "bold")
def docfield(
fieldname: str,
fieldtype: str,
*,
label: str | None = None,
options: str | None = None,
default: str | None = None,
flags: dict[str, int] | None = None,
) -> dict[str, Any]:
"""Une ligne de la table enfant `fields` d'un DocType."""
if fieldtype not in VALID_FIELDTYPES:
raise ValueError(f"Fieldtype non natif Frappe pour {fieldname!r} : {fieldtype!r}")
row: dict[str, Any] = {"fieldname": fieldname, "fieldtype": fieldtype}
if label is not None:
row["label"] = label
if options is not None:
row["options"] = options
if default is not None:
row["default"] = default
flags = flags or {}
for f in _DOCFIELD_FLAGS:
v = int(flags.get(f, 0))
if v not in (0, 1):
raise ValueError(f"Drapeau {f} de {fieldname!r} non booleen : {v!r}")
if v:
row[f] = 1
return row
def section_break(fieldname: str, label: str) -> dict[str, Any]:
return {"fieldname": fieldname, "fieldtype": "Section Break", "label": label}
def permission_row(role: str, perms: set[str], *, permlevel: int = 0) -> dict[str, Any]:
"""Ligne `permissions` (DocPerm integre) d'un DocType.
Emet uniquement les drapeaux a 1 (defaut sur = refus), dans l'ordre stable
`_PERM_ORDER` — diff reproductible.
"""
unknown = perms - VALID_PERMS
if unknown:
raise ValueError(f"Permission(s) non native(s) pour {role!r} : {sorted(unknown)}")
row: dict[str, Any] = {"role": role, "permlevel": permlevel}
for p in _PERM_ORDER:
if p in perms:
row[p] = 1
return row
def naming_series_field(series: str) -> dict[str, Any]:
"""Le champ Select `naming_series` requis quand `autoname = "naming_series:"`.
Frappe nomme le document depuis ce champ ; ses `options` portent le motif de
serie. Rendu `read_only` (serie imposee, pas de saisie libre).
"""
return docfield(
"naming_series",
"Select",
label="Serie de nommage",
options=series,
default=series,
flags={"read_only": 1},
)
def doctype_doc(
*,
name: str,
module: str,
is_submittable: int,
title_field: str,
search_fields: str,
track_changes: int,
track_seen: int,
fields: list[dict],
permissions: list[dict],
) -> dict[str, Any]:
"""Le document `DocType` complet (fixture custom pret pour `bench`)."""
return {
"doctype": "DocType",
"name": name,
"module": module,
"custom": 1,
"is_submittable": 1 if is_submittable else 0,
"naming_rule": "By \"Naming Series\" field",
"autoname": "naming_series:",
"title_field": title_field,
"search_fields": search_fields,
"track_changes": 1 if track_changes else 0,
"track_seen": 1 if track_seen else 0,
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [f["fieldname"] for f in fields],
"fields": fields,
"permissions": permissions,
}
@@ -0,0 +1,57 @@
"""Lecture des roles RBAC qui referencent un DocType donne (source unique · #6).
Le contrat `rbac_50_roles.json` est l'UNIQUE source de verite des roles ERPNext
(valide par `rbac.schema.json` dans le job CI `rbac-tests`). Ici on ne fait que
le LIRE : pour le DocType `CONFOTUR Application`, quels roles y ont acces et avec
quelles actions. Ces `permissions_cibles` deviennent, mot pour mot, les DocPerm
du DocType genere -> les permissions du fixture SONT le contrat RBAC, pas une
reinvention.
Reutilisation (workflow #5, zero duplication) : la resolution `id -> nom Frappe`
et le `portail` restent delegues au `RoleResolver` du module CRM voisin — on ne
redefinit ni le format des roles ni leur nommage.
"""
from __future__ import annotations
import json
import os
from typing import Any
_HERE = os.path.dirname(os.path.abspath(__file__))
_DELIVERABLES = os.path.normpath(os.path.join(_HERE, "..", "..", ".."))
RBAC_CONTRACT_PATH = os.path.join(_DELIVERABLES, "rbac", "rbac_50_roles.json")
def load_contract(path: str = RBAC_CONTRACT_PATH) -> dict[str, Any]:
with open(path, encoding="utf-8") as fh:
return json.load(fh)
def roles_targeting(doctype: str, contract: dict[str, Any]) -> list[dict[str, Any]]:
"""Roles dont un `permissions_cibles` vise `doctype`.
Retour trie par `id` de role (determinisme). Chaque element :
{role_id, actions: [..], custom: bool}
`actions` reprend, sans ajout ni retrait, la liste declaree dans le contrat.
Leve si un meme role declare deux fois le meme doctype (contrat ambigu).
"""
out: list[dict[str, Any]] = []
for role in contract.get("roles", []):
matches = [pc for pc in role.get("permissions_cibles", [])
if pc.get("doctype") == doctype]
if not matches:
continue
if len(matches) > 1:
raise ValueError(
f"Role {role.get('id')!r} declare {doctype!r} plusieurs fois "
f"dans permissions_cibles (contrat ambigu)."
)
pc = matches[0]
out.append({
"role_id": role["id"],
"actions": list(pc.get("actions", [])),
"custom": bool(pc.get("custom", False)),
})
out.sort(key=lambda r: r["role_id"])
return out