#!/usr/bin/env python3 """Générateur du manifest de dépendances PIE · Annexe 12 · V10.1. Directive : `DIRECTIVE_PIE_PROJECT_IDENTITY_ENGINE_20260803.md` (Michel · 2026-08-03). « La faisabilité devient le Project DNA · source unique de vérité (SSOT). » Ce module livre le **P0** du PIE : le schéma Project Master Data + le manifest de dépendances (§88 de la directive) qui liste, pour chaque déclencheur de la faisabilité, les livrables downstream à régénérer — et lesquels sont DÉJÀ gated par un module MVP existant vs restent à construire (P1-P4). Ce worker n'écrit JAMAIS sur le VPS (#8) : le storage réel du PIE (`/opt/oto/data/pie/{code}/`) et la génération effective des Brand Books / logos / PDF restent côté agents (Flux, ERPNext, OTOIA). Ici : uniquement le contrat de dépendances versionnable, cross-vérifié contre CLAUDE.md, la directive et le FS. Sous-commandes : build [-o OUT] → écrit pie_manifest.json + MANIFEST.json validate → (re)génère en mémoire, valide schéma + 10 invariants ; sort en erreur sinon. Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable. """ 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/ sys.path.insert(0, _HERE) sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste")) from pielib import builder, deps # noqa: E402 from lib import validator as maison # type: ignore # noqa: E402 _SPEC_PATH = os.path.join(_HERE, "pie_spec.json") _SCHEMA_PATH = os.path.join(_HERE, "pie.schema.json") _DEFAULT_OUT = os.path.join(_HERE, "out") # Ensemble documenté des déclencheurs de synchronisation (directive §82-86). _DECLENCHEURS = {"prix", "typologie", "amenities", "livraison"} # Clés canoniques des 12 groupes Project Master Data (directive §16-29). _GROUPES_CANONIQUES = { "identite", "localisation", "architecture", "unites", "budget", "marketing", "bim", "finances", "juridique", "images", "services", "calendrier", } 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 _validate(bundle: dict) -> list[str]: """Schéma de sortie + 10 invariants (SSOT · ancrages CLAUDE.md/directive/FS). Ledger des invariants (le compte « 10 » est ré-affiché à l'exécution) : 1 · schéma draft-07 de pie_manifest.json 2 · 12 groupes Master Data, clés uniques == ensemble canonique (§16-29) 3 · 4 règles sync, triggers == {prix, typologie, amenities, livraison} 4 · chaque downstream référencé existe dans le registre ; trigger_group réel 5 · registre downstream : statut cohérent avec présence d'un module 6 · chaque module downstream non nul est un répertoire de livrable réel (FS) 7 · workflow = étapes 1..N contiguës et uniques 8 · codes marques ⊆ CLAUDE.md §Projets (ancrage constitution) 9 · annexe (12) + version_workflow (V10.1) verbatim dans la directive 10 · storage layout hors-repo (/opt/oto/data/...) — jamais le out/ du module (#8) """ schema = _load(_SCHEMA_PATH) errors: list[str] = [] man = bundle["manifest"] trace = bundle["traceability"] # 1 · schéma de sortie. errors += list(maison.validate(man, schema)) # 2 · Project Master Data : 12 groupes, clés uniques, == ensemble canonique. groups = man["master_data_groups"] cles = [g["cle"] for g in groups] if len(cles) != 12: errors.append(f"master_data_groups : {len(cles)} groupes (attendu 12 · §16-29)") if len(set(cles)) != len(cles): errors.append("master_data_groups : clés dupliquées") if set(cles) != _GROUPES_CANONIQUES: manquants = _GROUPES_CANONIQUES - set(cles) surplus = set(cles) - _GROUPES_CANONIQUES errors.append(f"master_data_groups : clés ≠ ensemble canonique " f"(manquants={sorted(manquants)} · surplus={sorted(surplus)})") # 3 · règles de synchronisation : 4 triggers == ensemble documenté (§82-86). triggers = [r["trigger"] for r in man["sync_rules"]] if set(triggers) != _DECLENCHEURS: errors.append(f"sync_rules : triggers {sorted(set(triggers))} ≠ " f"{sorted(_DECLENCHEURS)} (§82-86)") if len(set(triggers)) != len(triggers): errors.append("sync_rules : trigger dupliqué") # 4 · intégrité référentielle : downstream + trigger_group pointent du réel. registry_keys = {d["cle"] for d in man["downstream_registry"]} group_keys = set(cles) for r in man["sync_rules"]: if r["trigger_group"] not in group_keys: errors.append(f"sync_rule {r['trigger']!r} : trigger_group " f"{r['trigger_group']!r} absent des Master Data groups") for t in r["downstream"]: if t not in registry_keys: errors.append(f"sync_rule {r['trigger']!r} : downstream {t!r} " f"absent du downstream_registry") # 5 · registre downstream : statut cohérent (gated ⇔ module non nul). for d in man["downstream_registry"]: attendu = "gated" if d["module"] else "a_construire" if d["statut"] != attendu: errors.append(f"downstream {d['cle']!r} : statut {d['statut']!r} " f"incohérent avec module {d['module']!r}") # 6 · chaque module downstream non nul EST un répertoire de livrable réel (FS). for d in man["downstream_registry"]: if d["module"] and not deps.module_exists(d["module"]): errors.append(f"downstream {d['cle']!r} : module {d['module']!r} " f"introuvable sous 05_deliverables_mvp/") # 7 · workflow : étapes 1..N contiguës et uniques. etapes = [s["etape"] for s in man["workflow"]] if etapes != list(range(1, len(etapes) + 1)): errors.append(f"workflow : étapes non contiguës/desordonnées : {etapes}") # 8 · codes marques ⊆ CLAUDE.md §Projets (ancrage constitution · #6). codes = {b["code"] for b in man["brands"]} canoniques = deps.claude_md_project_codes() if not canoniques: errors.append("CLAUDE.md §Projets : aucun code Pxx lu (ancrage impossible)") elif not codes <= canoniques: errors.append(f"brands : codes {sorted(codes - canoniques)} absents de " f"CLAUDE.md §Projets") # 9 · annexe + version verbatim dans la directive (ancrage directive · #6). decl = deps.directive_declares(trace["directive_source"], f"Annexe {man['annexe']}", man["version_workflow"]) for needle, present in decl.items(): if not present: errors.append(f"directive {trace['directive_source']!r} : " f"« {needle} » introuvable (ancrage rompu)") # 10 · storage layout hors-repo : aucun artefact PIE n'atterrit dans le out/ # du module (le worker n'écrit JAMAIS le storage VPS · #8). racine = man["storage_layout"]["racine"] if not racine.startswith("/opt/oto/data/pie/"): errors.append(f"storage_layout.racine {racine!r} hors /opt/oto/data/pie/") if os.path.abspath(deps.module_out_dir()).startswith(os.path.abspath("/opt/oto/data")): errors.append("out/ du module confondu avec le storage VPS (#8)") # Traçabilité : comptes du manifeste cohérents avec les tableaux. c = trace["counts"] if c["master_data_groups"] != len(groups): errors.append("counts.master_data_groups incohérent") if c["downstream_total"] != len(man["downstream_registry"]): errors.append("counts.downstream_total incohérent") if c["sync_rules"] != len(man["sync_rules"]): errors.append("counts.sync_rules incohérent") if c["workflow_steps"] != len(man["workflow"]): errors.append("counts.workflow_steps incohérent") if c["brands"] != len(man["brands"]): errors.append("counts.brands incohérent") gated = sum(1 for d in man["downstream_registry"] if d["statut"] == "gated") if c["downstream_gated"] != gated: errors.append("counts.downstream_gated incohérent") return errors def _build() -> dict: return builder.build_bundle(_load(_SPEC_PATH)) def cmd_build(args: argparse.Namespace) -> int: bundle = _build() errors = _validate(bundle) if errors: _eprint("❌ Manifest invalide — génération refusée (anti-régression) :") 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, "pie_manifest.json"), bundle["manifest"]) _write_json(os.path.join(out, "MANIFEST.json"), bundle["traceability"]) c = bundle["traceability"]["counts"] print(f"✅ Manifest PIE généré dans {out}") print(f" pie_manifest.json : {c['master_data_groups']} groupes Master Data · " f"{c['sync_rules']} règles sync · {c['downstream_total']} livrables downstream " f"({c['downstream_gated']} gated · {c['downstream_a_construire']} à construire) · " f"{c['workflow_steps']} étapes workflow · {c['brands']} marques") print(" ⚠ Storage PIE réel (/opt/oto/data/pie/{code}/) + génération Brand Book / " "logos / PDF côté agents (Flux · ERPNext · OTOIA · VPS · #8).") return 0 def cmd_validate(args: argparse.Namespace) -> int: bundle = _build() errors = _validate(bundle) if errors: _eprint("❌ Validation KO :") for e in errors: _eprint(f" - {e}") return 1 c = bundle["traceability"]["counts"] print(f"✅ Validation OK — PIE Annexe {bundle['manifest']['annexe']} · " f"{bundle['manifest']['version_workflow']} : schéma + 10 invariants verts " f"({c['downstream_total']} downstream · {c['brands']} marques).") return 0 def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser(description="Générateur du manifest de dépendances PIE (Annexe 12 · V10.1).") sub = p.add_subparsers(dest="cmd", required=True) pb = sub.add_parser("build", help="génère pie_manifest.json / MANIFEST.json") pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (défaut: ./out)") pb.set_defaults(func=cmd_build) pv = sub.add_parser("validate", help="valide le bundle (schéma + 10 invariants) sans écrire") pv.set_defaults(func=cmd_validate) args = p.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())