#!/usr/bin/env python3 """Generateur du DocType porteur `CONFOTUR Application` · Sprint 5 · ONAPI/Legal. Roadmap ligne 55 : « Refactor oto_module_confotur_application.py -> depots automatiques ». Produit le fixture DocType custom qui porte les dossiers d'incitation touristique CONFOTUR (Ley 158-01, RD), cross-coherent avec les contrats deja livres : - le contrat RBAC (`rbac_50_roles.json`) -> nom, permissions, submit ; - le pipeline vente (`workflow_vente_spec.json`) -> etats `estado`, DocType lie `dossier_vente`, et evenements de depot (confotur_depose/_approuve). Ce worker n'ecrit JAMAIS sur le VPS (#8) : il emet les fichiers de hand-off ; la creation du module Frappe `OTOV7 CONFOTUR`, l'import du DocType et le cablage des evenements restent cote agent ONAPI/Legal / ERPNext Backend. ANTI-INVENTION (#6) : aucun taux d'incitation, article de loi, montant ni reference d'autorite n'est fabrique. Le DocType est une STRUCTURE ; ses champs substantiels restent vides, renseignes cote VPS depuis data_room P05/P07. Un invariant refuse tout `default` de champ (hors serie de nommage) et tout type de champ porteur de montant. Sous-commandes : build [-o OUT] -> ecrit doctype_confotur_application.json + MANIFEST.json validate -> (re)genere en memoire, valide schema + 14 invariants de cross-coherence DocType<->RBAC<->workflow ; sort en erreur sinon. Sortie deterministe (tri stable, aucun horodatage) -> diffable + re-generable. """ from __future__ import annotations import argparse import json import os import sys _HERE = os.path.dirname(os.path.abspath(__file__)) _DELIVERABLES = os.path.normpath(os.path.join(_HERE, "..", "..")) # 05_deliverables_mvp/ _CRM = os.path.join(_DELIVERABLES, "crm") sys.path.insert(0, _HERE) sys.path.insert(0, _CRM) sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste")) from cflib import builder, frappe, rbac_scan # 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, "confotur_spec.json") _WF_SPEC_PATH = os.path.join(_CRM, "workflow_vente", "workflow_vente_spec.json") _SCHEMA_PATH = os.path.join(_HERE, "confotur.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 _build() -> tuple[dict, dict, dict, dict, RoleResolver]: spec = _load(_SPEC_PATH) wf_spec = _load(_WF_SPEC_PATH) contract = rbac_scan.load_contract() resolver = RoleResolver.from_path() bundle = builder.build_bundle(spec, wf_spec, contract, resolver) return bundle, spec, wf_spec, contract, resolver def _granted_perms(perm_row: dict) -> set[str]: """Reconstruit l'ensemble d'actions accordees depuis une ligne DocPerm.""" return {p for p in frappe.VALID_PERMS if perm_row.get(p) == 1} def _validate(bundle: dict, spec: dict, wf_spec: dict, contract: dict, resolver: RoleResolver) -> list[str]: """Schema de sortie + 14 invariants de cross-coherence (3 contrats).""" schema = _load(_SCHEMA_PATH) errors = list(maison.validate(bundle, schema)) dt = bundle["doctype"] m = bundle["manifest"] fields = dt["fields"] by_name = {f["fieldname"]: f for f in fields} targeting = rbac_scan.roles_targeting(builder.TARGET_DOCTYPE, contract) tgt_by_name = {resolver.erpnext_name(r["role_id"]): r for r in targeting} confotur_vals = builder.confotur_state_values(wf_spec) docstatus_by_val = {s["update_value"]: s["doc_status"] for s in wf_spec["states"]} # 1 · nom du DocType coherent partout (identite du generateur). if dt["name"] != builder.TARGET_DOCTYPE or m["doctype_name"] != builder.TARGET_DOCTYPE: errors.append(f"nom du DocType != {builder.TARGET_DOCTYPE!r}") # 2 · permissions == permissions_cibles RBAC, mot pour mot + exhaustif. perm_roles = [p["role"] for p in dt["permissions"]] if len(perm_roles) != len(set(perm_roles)): errors.append("role duplique dans doctype.permissions") if set(perm_roles) != set(tgt_by_name): errors.append("ensemble des roles du DocType != roles RBAC visant le DocType") for row in dt["permissions"]: r = tgt_by_name.get(row["role"]) if r is None: errors.append(f"permission pour role {row['role']!r} absent du contrat RBAC") continue granted = _granted_perms(row) if granted != set(r["actions"]): errors.append( f"permissions de {row['role']!r} = {sorted(granted)} != RBAC " f"{sorted(set(r['actions']))}" ) if row["permlevel"] != 0: errors.append(f"permlevel != 0 pour {row['role']!r}") # 3 · custom : DocType custom ET tous les roles le declarent custom (#1). if dt["custom"] != 1 or m["custom"] is not True: errors.append("doctype.custom doit valoir 1") if not all(r["custom"] for r in targeting): errors.append("un role RBAC visant le DocType ne le marque pas `custom`") # 4 · is_submittable deduit de l'action `submit` cote RBAC. expect_sub = 1 if any("submit" in r["actions"] for r in targeting) else 0 if dt["is_submittable"] != expect_sub or m["is_submittable"] is not bool(expect_sub): errors.append(f"is_submittable={dt['is_submittable']} != attendu {expect_sub}") # 5 · coherence avec le workflow : les etats CONFOTUR sont soumis (doc_status 1) # -> un DocType soumissible porte un pipeline soumissible. if not confotur_vals: errors.append("aucun etat `confotur*` dans le workflow vente") for uv in confotur_vals: if docstatus_by_val.get(uv) != "1": errors.append(f"etat confotur {uv!r} n'est pas soumis (doc_status != 1)") # 6 · liens : options des champs Link = DocTypes attendus (dossier_vente derive # du workflow, jamais en dur) + manifeste `doctypes_lies_a_confirmer` aligne. if by_name.get("projet", {}).get("options") != "Project": errors.append("champ `projet` : options != 'Project'") if by_name.get("cliente", {}).get("options") != "Customer": errors.append("champ `cliente` : options != 'Customer'") if by_name.get("dossier_vente", {}).get("options") != wf_spec["document_type"]: errors.append("champ `dossier_vente` : options != workflow document_type") linked = sorted({f.get("options") for f in fields if f["fieldtype"] == "Link" and f.get("options")}) if m["doctypes_lies_a_confirmer"] != linked: errors.append("manifest.doctypes_lies_a_confirmer incoherent") # 7 · `estado` : options Select derivees des etats CONFOTUR (exhaustif, sans # extra) avec ligne vide en tete (facultatif a la creation). est = by_name.get("estado", {}) est_opts = est.get("options", "").split("\n") if est_opts[:1] != [""]: errors.append("`estado` : ligne vide en tete attendue (Select facultatif)") if [o for o in est_opts if o] != confotur_vals: errors.append("`estado` : options != etats confotur du workflow") # 8 · `entite_porteuse` : options = entites CLAUDE.md du spec (Select requis). ent = by_name.get("entite_porteuse", {}) if ent.get("options", "").split("\n") != spec["entites"]: errors.append("`entite_porteuse` : options != entites du spec") if ent.get("reqd") != 1: errors.append("`entite_porteuse` doit etre reqd") # 9 · depot_events derives des transitions confotur du workflow. de = m["depot_events"] wf_confotur_tr = [t for t in wf_spec["transitions"] if {s["state"]: s["update_value"] for s in wf_spec["states"]} .get(t["next_state"], "").startswith("confotur")] if len(de) != len(wf_confotur_tr): errors.append("depot_events : compte != transitions confotur du workflow") for ev in de: if not ev["update_value"].startswith("confotur"): errors.append(f"depot_event {ev['action']!r} : update_value non confotur") if ev["update_value"] not in confotur_vals: errors.append(f"depot_event {ev['action']!r} : etat absent du workflow") if ev["role_id"] not in {r["role_id"] for r in targeting}: errors.append( f"depot_event {ev['action']!r} : role {ev['role_id']!r} sans acces " f"au DocType CONFOTUR Application (RBAC)" ) if ev["erpnext_role_name"] != resolver.erpnext_name(ev["role_id"]): errors.append(f"depot_event {ev['action']!r} : erpnext_role_name incoherent") if ev["portail"] != resolver.portail(ev["role_id"]): errors.append(f"depot_event {ev['action']!r} : portail incoherent") # 10 · ANTI-INVENTION (#6) : aucun champ ne porte de montant ni de `default` # (hors serie de nommage) -> aucun chiffre OTO smuggle dans la structure. money_types = {"Currency", "Float", "Int", "Percent"} for f in fields: if f["fieldtype"] in money_types: errors.append(f"champ {f['fieldname']!r} de type montant interdit (#6)") if "default" in f and f["fieldname"] != "naming_series": errors.append(f"champ {f['fieldname']!r} porte un `default` (interdit #6)") # 11 · serie de nommage native (autoname naming_series + champ read_only). ns = by_name.get("naming_series", {}) if dt["autoname"] != "naming_series:": errors.append("autoname != 'naming_series:'") if ns.get("options") != spec["doctype"]["naming_series"] or ns.get("read_only") != 1: errors.append("champ naming_series absent / non read_only / options != serie") # 12 · field_order == fieldnames, unicite, title/search fields reels. names = [f["fieldname"] for f in fields] if dt["field_order"] != names: errors.append("field_order != ordre des fields") if len(names) != len(set(names)): errors.append("fieldname duplique") if dt["title_field"] not in by_name: errors.append(f"title_field {dt['title_field']!r} absent des champs") for sf in dt["search_fields"].split(","): if sf and sf not in by_name: errors.append(f"search_field {sf!r} absent des champs") # 13 · tous les champs metier du spec sont presents (aucune perte). spec_fields = {f["fieldname"] for g in spec["field_groups"] for f in g["fields"]} missing = spec_fields - set(by_name) if missing: errors.append(f"champs du spec absents du DocType : {sorted(missing)}") # 14 · comptes du manifeste coherents. 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"] checks = { "fields": len(fields), "data_fields": len(data_fields), "sections": len(sections), "permissions": len(dt["permissions"]), "roles": len(targeting), "depot_events": len(de), } for k, v in checks.items(): if m["counts"][k] != v: errors.append(f"counts.{k} incoherent ({m['counts'][k]} != {v})") return errors def cmd_build(args: argparse.Namespace) -> int: bundle, spec, wf_spec, contract, resolver = _build() errors = _validate(bundle, spec, wf_spec, contract, resolver) if errors: _eprint("❌ Bundle invalide — generation refusee (anti-regression) :") 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, "doctype_confotur_application.json"), bundle["doctype"]) _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"]) m = bundle["manifest"] print(f"✅ DocType CONFOTUR Application genere dans {out}") print(f" doctype_confotur_application.json : {m['counts']['fields']} champs " f"({m['counts']['data_fields']} de donnee) · {m['counts']['sections']} sections · " f"{m['counts']['permissions']} roles · soumissible={m['is_submittable']} · " f"{m['counts']['depot_events']} evenements de depot") print(" ⚠ Module `OTOV7 CONFOTUR` + import DocType + cablage evenements cote " "agent ONAPI/Legal (VPS · #8). Taux/loi/montants : data_room P05/P07 (#6).") return 0 def cmd_validate(args: argparse.Namespace) -> int: bundle, spec, wf_spec, contract, resolver = _build() errors = _validate(bundle, spec, wf_spec, contract, resolver) if errors: _eprint("❌ Validation KO :") for e in errors: _eprint(f" - {e}") return 1 m = bundle["manifest"] print(f"✅ Validation OK — DocType {m['doctype_name']!r} : " f"{m['counts']['permissions']} roles RBAC, {m['counts']['depot_events']} " f"evenements de depot, schema + 14 invariants verts.") return 0 def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser(description="Generateur du DocType CONFOTUR Application (ONAPI/Legal).") sub = p.add_subparsers(dest="cmd", required=True) pb = sub.add_parser("build", help="genere doctype_confotur_application.json / MANIFEST.json") pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (defaut: ./out)") pb.set_defaults(func=cmd_build) pv = sub.add_parser("validate", help="valide le bundle (schema + 14 invariants) sans ecrire") pv.set_defaults(func=cmd_validate) args = p.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())