[DTP-Worker 20260803_093713] Auto exec · session 20260803_093713

05_deliverables_mvp/pie/manifest/pielib/__init__.py

@@ -0,0 +1,5 @@
+"""PIE · manifest de dépendances · bibliothèque interne.
+
+  - deps    → ancrages hors-module (CLAUDE.md §Projets · directive PIE · FS des livrables)
+  - builder → assemblage déterministe du manifest + du MANIFEST de traçabilité
+"""

05_deliverables_mvp/pie/manifest/pielib/builder.py

@@ -0,0 +1,100 @@
+"""Assemblage déterministe du manifest PIE depuis `pie_spec.json`.
+
+Entrée : le contrat PIE (taxonomie Project Master Data, registre downstream,
+règles de synchronisation, workflow 10 étapes, marques, storage layout) —
+transcrit VERBATIM de la directive PIE (SSOT · #6, aucune valeur inventée).
+Sortie : un bundle déterministe (tri stable, aucun horodatage) → reproductible
+bit-à-bit à contrat constant, donc diffable et re-générable en CI
+([[artifact-reproducibility-gate]]).
+
+Le builder n'ajoute AUCUN chiffre propre à OTO : il résout le `statut` de chaque
+livrable downstream (« gated » si un module de livrado réel l'implémente déjà,
+sinon « a_construire ») et un manifeste de traçabilité (comptes dérivés).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from . import deps
+
+
+def _statut(module: str | None) -> str:
+    """« gated » si un module réel implémente déjà le downstream, sinon « a_construire »."""
+    return "gated" if module else "a_construire"
+
+
+def build_bundle(spec: dict) -> dict[str, Any]:
+    """Transforme le contrat PIE en manifest de dépendances + manifeste de traçabilité."""
+    groups = sorted(spec["master_data_groups"], key=lambda g: g["cle"])
+
+    registry = [
+        {
+            "cle": d["cle"],
+            "libelle": d["libelle"],
+            "module": d["module"],
+            "statut": _statut(d["module"]),
+        }
+        for d in sorted(spec["downstream_registry"], key=lambda d: d["cle"])
+    ]
+
+    sync_rules = [
+        {
+            "trigger": r["trigger"],
+            "trigger_group": r["trigger_group"],
+            "downstream": list(r["downstream"]),   # ordre directive préservé (§82-86)
+        }
+        for r in sorted(spec["sync_rules"], key=lambda r: r["trigger"])
+    ]
+
+    workflow = sorted(spec["workflow"], key=lambda s: s["etape"])
+    brands = sorted(spec["brands"], key=lambda b: b["code"])
+
+    manifest_out = {
+        "engine": spec["engine"],
+        "annexe": spec["annexe"],
+        "version_workflow": spec["version_workflow"],
+        "ssot": spec["ssot"],
+        "source_spec": "pie_spec.json",
+        "master_data_groups": groups,
+        "downstream_registry": registry,
+        "sync_rules": sync_rules,
+        "workflow": workflow,
+        "brands": brands,
+        "storage_layout": spec["storage_layout"],
+    }
+
+    gated = [d for d in registry if d["statut"] == "gated"]
+    a_construire = [d for d in registry if d["statut"] == "a_construire"]
+
+    traceability = {
+        "generated_from": "pie_spec.json",
+        "claude_md_source": "CLAUDE.md",
+        "directive_source": spec["directive"],
+        "engine": spec["engine"],
+        "annexe": spec["annexe"],
+        "version_workflow": spec["version_workflow"],
+        "ssot": spec["ssot"],
+        "counts": {
+            "master_data_groups": len(groups),
+            "downstream_total": len(registry),
+            "downstream_gated": len(gated),
+            "downstream_a_construire": len(a_construire),
+            "sync_rules": len(sync_rules),
+            "workflow_steps": len(workflow),
+            "brands": len(brands),
+        },
+        "modules_gated_references": [
+            {"cle": d["cle"], "module": d["module"]} for d in gated
+        ],
+        "note_anti_invention": (
+            "Toute donnée est transcrite de la directive PIE (SSOT · #6) : aucune "
+            "valeur projet (prix, surfaces, marges) n'est fabriquée ici — le PIE "
+            "les EXTRAIT de la faisabilité validée côté VPS (/opt/oto/data/pie/{code}/, "
+            "jamais écrit par ce worker · #8). Les codes projets sont ancrés à "
+            "CLAUDE.md §Projets ; annexe/version à la directive ; les modules "
+            "downstream « gated » à des répertoires de livrables réels."
+        ),
+    }
+
+    return {"manifest": manifest_out, "traceability": traceability}

05_deliverables_mvp/pie/manifest/pielib/deps.py

@@ -0,0 +1,60 @@
+"""Ancrages hors-module du PIE manifest (anti-dérive · anti-invention #6).
+
+Le manifest PIE ne se valide pas seulement contre lui-même : trois faits sont
+RE-DÉRIVÉS de sources de vérité EXTERNES au module, jamais recopiés à la main —
+
+  1. Les codes projets (`brands[].code`) sont confrontés à `CLAUDE.md §Projets`
+     (la constitution · [[claude-md-constant-anchor-gate]]). Une marque PIE dont
+     le code ne figure pas dans la constitution est une invention refusée.
+  2. `annexe` (12) et `version_workflow` (V10.1) doivent apparaître VERBATIM dans
+     la directive PIE. Renuméroter l'annexe dans la directive sans régénérer ici
+     casse la validation ([[roadmap-anchor-gate]] appliqué à la directive).
+  3. Chaque `downstream_registry[].module` non nul doit être un répertoire de
+     livrable RÉEL sous `05_deliverables_mvp/` (comme fiscal/ecf vérifie ses
+     contrats voisins). Un module fantôme (renommé/supprimé) casse la validation.
+
+Zéro dépendance pip · zéro réseau · aucune écriture VPS (#8) : lecture seule de
+fichiers déjà versionnés dans le dépôt.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_MODULE_DIR = os.path.normpath(os.path.join(_HERE, ".."))          # pie/manifest/
+DELIVERABLES_ROOT = os.path.normpath(os.path.join(_HERE, "..", "..", ".."))  # 05_deliverables_mvp/
+_REPO_ROOT = os.path.normpath(os.path.join(DELIVERABLES_ROOT, ".."))
+_CLAUDE_MD = os.path.join(_REPO_ROOT, "CLAUDE.md")
+
+
+def claude_md_project_codes() -> set[str]:
+    """Codes Pxx de la ligne `## Projets` de CLAUDE.md (source de vérité #Projets)."""
+    with open(_CLAUDE_MD, encoding="utf-8") as fh:
+        text = fh.read()
+    # Section entre « ## Projets » et le prochain titre « ## ».
+    m = re.search(r"^##\s+Projets\s*\n(.*?)(?:\n##\s|\Z)", text, re.S | re.M)
+    section = m.group(1) if m else ""
+    return set(re.findall(r"\bP0[1-9]\b", section))
+
+
+def directive_declares(directive_relpath: str, *needles: str) -> dict[str, bool]:
+    """Vrai/faux, pour chaque `needle`, s'il apparaît verbatim dans la directive PIE."""
+    path = os.path.join(_REPO_ROOT, directive_relpath)
+    try:
+        with open(path, encoding="utf-8") as fh:
+            text = fh.read()
+    except OSError:
+        return {n: False for n in needles}
+    return {n: (n in text) for n in needles}
+
+
+def module_exists(rel_module: str) -> bool:
+    """Le module de livrable `rel_module` (ex. 'faisabilite/bancable') existe-t-il ?"""
+    return os.path.isdir(os.path.join(DELIVERABLES_ROOT, rel_module))
+
+
+def module_out_dir() -> str:
+    """Répertoire de sortie du module (le SEUL endroit où ce worker écrit · #8)."""
+    return os.path.join(_MODULE_DIR, "out")