oto-enterprise-os-dtp/05_deliverables_mvp/pie/manifest/pie_manifest_gen.py

#!/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())