"""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}