Claude Code DTP Worker
35a20247f8
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>
304 lines
14 KiB
Python
304 lines
14 KiB
Python
#!/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())
|