[DTP-Worker] Sprint 2 · Publiciste scaffold (parser faisabilité → JSON + generator + gate)

Livrable Publiciste · Sprint 2 · Semaine 2 (seul module net-neuf · chemin
critique · GAP_ANALYSIS §3.13). Cible portage VPS : otoia/capabilities/publiciste.py.

- lib/parser.py : data_room/PXX/ (template v1.0) → projet dict conforme au
  contrat projets_master.schema.json (livré S1). Mapping colonnes par en-tête,
  parsing montants USD/DOP robuste. Anti-invention #6 : rétrogradation défensive
  « en_developpement » si prix USD manquant ; absent → null (jamais 0/inventé).
- lib/validator.py : validateur JSON-Schema draft-07 (sous-ensemble) ZÉRO
  dépendance pip (runner Gitea sans pip). Oracle jsonschema en test si présent.
- lib/generator.py + template + branding : rendu HTML luxury #4 (dark+doré,
  Fraunces + Cormorant Garamond) ; sans prix → « Prochainement · Détails à venir ».
- publiciste.py : CLI parse/validate/generate/run.
- fixtures/ : données SYNTHÉTIQUES de test (jamais publiées) P01 complète + P02
  incomplète.
- tests/ : 23 tests unittest (stdlib) verts.
- CI : job publiciste-tests ajouté au gate (.gitea/workflows/ci.yml · Gitea #2).

Vérifs (en-repo, sans VPS) : 23/23 tests verts · gate CI local vert (guard/JSON/
docs) · pipeline CLI produit un master conforme au schéma. Auto-score 4Big 95/100.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Code DTP Worker
2026-07-30 01:37:47 +00:00
parent 1a87b22fd7
commit f0a7d71357
23 changed files with 1535 additions and 1 deletions
@@ -0,0 +1,11 @@
"""Publiciste Agent · bibliothèque interne.
Modules :
- parser : data_room/PXX/ → dict projet (contrat projets_master.schema.json)
- validator : validation JSON-Schema draft-07 (sous-ensemble, zéro dépendance pip)
- generator : projets_master.json → HTML public (vente.otov7.com)
- branding : tokens de marque luxury (CLAUDE.md #4)
Cible de portage sur le VPS : otoia/capabilities/publiciste.py
(voir 03_agents/publiciste/AGENT.md · GAP_ANALYSIS_SPRINT1.md §3.13).
"""
@@ -0,0 +1,40 @@
"""Tokens de marque luxury · CLAUDE.md contrainte #4 (source unique de vérité).
Dark + doré : `#0a0a12` fond · `#f0b429` accent doré.
Typographies : Fraunces (titres) + Cormorant Garamond (corps éditorial).
Ces constantes sont importées par le générateur ET vérifiées par la baseline QA
Playwright (tests/e2e/_shared/contract.ts) — garder les valeurs synchronisées.
"""
from __future__ import annotations
# Couleurs canoniques (contrainte #4).
COLOR_BG = "#0a0a12" # fond dark
COLOR_ACCENT = "#f0b429" # doré
COLOR_INK = "#f5f3ee" # texte clair sur fond dark
COLOR_MUTED = "#8b8778" # texte secondaire
# Typographies canoniques (contrainte #4).
FONT_DISPLAY = "Fraunces"
FONT_BODY = "Cormorant Garamond"
# Devises canoniques (contrainte #10 · USD + DOP).
DEVISE_PRIMAIRE = "USD"
DEVISE_SECONDAIRE = "DOP"
# Libellés de statut affichés sur le site public (FR par défaut).
STATUT_LABELS = {
"disponible": "Disponible",
"en_developpement": "En développement",
"bientot": "Bientôt",
"en_processus": "En processus",
}
# Statuts pour lesquels on affiche « Prochainement · Détails à venir »
# au lieu d'une grille de prix (jamais de prix inventé — contrainte #6).
STATUTS_SANS_PRIX = {"en_developpement", "bientot", "en_processus"}
def statut_label(statut: str) -> str:
return STATUT_LABELS.get(statut, statut)
@@ -0,0 +1,162 @@
"""Générateur HTML · projets_master.json → site public (vente.otov7.com).
Livrable Publiciste · Sprint 2 · Semaine 3 (aperçu ; le parser Semaine 2 est le
cœur du sprint courant). Rendu server-side pur stdlib, marque luxury (#4).
Interdits appliqués (AGENT.md §Règles absolues) :
- ❌ Jamais éditer directement l'index publié : on passe TOUJOURS par ce
générateur, qui écrit une sortie complète et horodatée.
- ❌ Zéro prix inventé : un projet sans prix numérique n'affiche PAS de grille,
mais « Prochainement · Détails à venir » (contrainte #6).
- ❌ Rendus IA génériques interdits : on n'affiche que les rendus référencés
dans la faisabilité (liés au projet réel).
"""
from __future__ import annotations
import html
import os
from typing import Any
from . import branding
_TEMPLATE = os.path.join(os.path.dirname(__file__), "..", "templates", "site_public.html.tmpl")
_TAGLINE = "Résidences d'exception en République dominicaine · faisabilités 4 volets."
_A_VENIR = "Prochainement · Détails à venir"
def _esc(text: Any) -> str:
return html.escape(str(text), quote=True)
def _fmt_usd(value: float) -> str:
return f"USD {value:,.0f}".replace(",", " ")
def _fmt_dop(value: float) -> str:
return f"DOP {value:,.0f}".replace(",", " ")
def _fmt_m2(value: float) -> str:
txt = f"{value:,.0f}".replace(",", " ") if float(value).is_integer() else f"{value:,.1f}".replace(",", " ")
return f"{txt}"
def _prix_depuis(typologies: list[dict[str, Any]]) -> float | None:
"""Prix « à partir de » du projet = min des prix USD numériques présents."""
prices = [t["prix_depuis_usd"] for t in typologies if isinstance(t.get("prix_depuis_usd"), (int, float))]
return min(prices) if prices else None
def _render_typologies(typologies: list[dict[str, Any]]) -> str:
rows = []
for t in typologies:
surface = t.get("surface_totale_m2") or t.get("surface_interieure_m2")
surface_txt = _fmt_m2(surface) if isinstance(surface, (int, float)) else ""
prix = t.get("prix_depuis_usd")
prix_txt = (
f'<span class="prix-depuis">{_esc(_fmt_usd(prix))}</span>'
if isinstance(prix, (int, float))
else ""
)
rows.append(
" <tr>"
f"<td>{_esc(t.get('nom', ''))}</td>"
f"<td>{surface_txt}</td>"
f"<td>{prix_txt}</td>"
"</tr>"
)
return (
' <table class="prix">\n'
" <tr><th>Typologie</th><th>Surface</th><th>À partir de</th></tr>\n"
+ "\n".join(rows)
+ "\n </table>"
)
def _render_inclus(inclus: list[str]) -> str:
if not inclus:
return ""
items = "\n".join(f" <li>{_esc(x)}</li>" for x in inclus)
return f' <ul class="inclus">\n{items}\n </ul>'
def _hero_media(projet: dict[str, Any]) -> str:
rendus = projet.get("rendus", [])
hero = next((r for r in rendus if r.get("hero")), rendus[0] if rendus else None)
if not hero:
return ' <div class="media"></div>'
# Convention de chemin public (miroir de /opt/oto/sites/static/projets/pXX/).
code = projet["code"].lower()
src = f"/static/projets/{code}/{hero['fichier']}"
style = f"background-image:url('{_esc(src)}')"
return f' <div class="media" style="{style}" role="img" aria-label="{_esc(projet.get("nom",""))}"></div>'
def render_projet(projet: dict[str, Any]) -> str:
statut = projet.get("statut", "en_developpement")
statut_label = branding.statut_label(statut)
nom = projet.get("nom", projet.get("code", ""))
loc = projet.get("localisation", "")
parts = [f' <article class="projet" data-code="{_esc(projet.get("code",""))}">']
parts.append(_hero_media(projet))
parts.append(' <div class="corps">')
parts.append(f' <span class="statut {_esc(statut)}">{_esc(statut_label)}</span>')
parts.append(f" <h2>{_esc(nom)}</h2>")
if loc:
parts.append(f' <p class="localisation">{_esc(loc)}</p>')
positionnement = (projet.get("positionnement") or {}).get("fr")
if positionnement:
parts.append(f' <p class="positionnement">{_esc(positionnement)}</p>')
typologies = projet.get("typologies", [])
show_prices = statut not in branding.STATUTS_SANS_PRIX and _prix_depuis(typologies) is not None
if show_prices:
parts.append(_render_typologies(typologies))
else:
# Aucun prix inventé — message d'attente (contrainte #6).
parts.append(f' <p class="a-venir">{_esc(_A_VENIR)}</p>')
inclus_html = _render_inclus(projet.get("inclus", []))
if inclus_html:
parts.append(inclus_html)
parts.append(" </div>")
parts.append(" </article>")
return "\n".join(parts)
def render_site(master: dict[str, Any]) -> str:
"""projets_master.json (dict) → page HTML complète (str)."""
with open(_TEMPLATE, encoding="utf-8") as fh:
tmpl = fh.read()
projets_html = "\n".join(render_projet(p) for p in master.get("projets", []))
generated_at = master.get("generated_at", "")
template_version = master.get("template_version", "")
footer = (
f"Contenu généré automatiquement depuis les faisabilités canoniques · "
f"template {_esc(template_version)} · {_esc(generated_at)}. "
"Prix « à partir de » en USD + DOP (Cardnet). Aucune donnée inventée — "
"chaque valeur provient de la faisabilité du projet."
)
replacements = {
"{{TITRE}}": "Helios RD · Résidences d'exception",
"{{META_DESCRIPTION}}": _TAGLINE,
"{{TAGLINE}}": _TAGLINE,
"{{COLOR_BG}}": branding.COLOR_BG,
"{{COLOR_ACCENT}}": branding.COLOR_ACCENT,
"{{COLOR_INK}}": branding.COLOR_INK,
"{{COLOR_MUTED}}": branding.COLOR_MUTED,
"{{FONT_DISPLAY}}": branding.FONT_DISPLAY,
"{{FONT_BODY}}": branding.FONT_BODY,
"{{PROJETS}}": projets_html,
"{{FOOTER}}": footer,
}
for key, val in replacements.items():
tmpl = tmpl.replace(key, val)
return tmpl
@@ -0,0 +1,442 @@
"""Parser Faisabilité → contrat Publiciste (`projets_master.json`).
Livrable Publiciste · Sprint 2 · Semaine 2 (AGENT.md §Livrable Sprint).
Lit une faisabilité canonique `data_room/PXX/` (template v1.0) et produit le dict
`projet` conforme à `projets_master.schema.json`, consommé ensuite par le
générateur HTML.
Règles ANTI-INVENTION (CLAUDE.md #6) appliquées ici — le parser est volontairement
défensif : il n'invente JAMAIS un chiffre absent, et il **rétrograde** un projet
en « en_developpement » plutôt que de publier une donnée douteuse.
1. Le statut publié dérive de `_META/version.json.statut_faisabilite` :
complete → disponible
en_developpement→ en_developpement
incomplete → en_developpement
2. Un projet n'est « disponible » QUE si version.json le dit complete ET que
chaque typologie a un prix USD numérique (sinon rétrogradation défensive).
3. Une cellule vide / « non défini » / placeholder `{{...}}` → None (jamais 0,
jamais une valeur inventée).
Sources lues (ordre du template canonique v1.0) :
_META/version.json → statut, score, traçabilité
00_brief/brief.md → localisation (§1.1)
20_architecture/architecture.md → typologies + prix (§3.2, bloc anti-gap)
30_paysage_experience/paysage_experience.md→ services inclus (§3.3)
40_llm_outputs/commercial.md → positionnement FR (fallback prix §5.3)
60_photos_site/ → rendus (liés au projet réel)
"""
from __future__ import annotations
import json
import os
import re
from datetime import datetime, timezone
from typing import Any, Optional
# Marqueurs signalant une donnée ABSENTE (jamais inventer — contrainte #6).
_ABSENT = {
"", "", "-", "", "n/d", "nd", "na", "n.a.", "s/o", "so",
"non défini", "non defini", "non renseigné", "non renseigne",
"à définir", "a definir", "tbd", "todo", "", "...",
}
# --------------------------------------------------------------------------- #
# Helpers de bas niveau : nettoyage de cellules Markdown → valeurs typées.
# --------------------------------------------------------------------------- #
def _clean(cell: str) -> str:
"""Retire gras/italique/backticks et espaces d'une cellule Markdown."""
return cell.replace("**", "").replace("`", "").replace("*", "").strip()
def _is_absent(text: str) -> bool:
t = _clean(text).lower()
if not t:
return True
if "{{" in t and "}}" in t: # placeholder de template non rempli
return True
return t in _ABSENT
def parse_number(text: str) -> Optional[float]:
"""« USD 250,000 » / « 1 250,50 » / « 3.5 » → float ; absent → None.
Détecte le séparateur décimal (dernier « . » ou « , » suivi de 1-2 chiffres)
et traite les autres séparateurs comme des milliers. Ignore tout habillage
(devises, unités m², %, texte). Retourne None dès qu'aucun chiffre exploitable.
"""
if _is_absent(text):
return None
raw = _clean(text)
# Isole le premier bloc numérique (chiffres, points, virgules, espaces).
m = re.search(r"[0-9][0-9\s., ]*", raw)
if not m:
return None
token = m.group(0).strip().replace(" ", " ")
token = token.rstrip(" .,")
# Décide du séparateur décimal : le dernier '.' ou ',' suivi de 1-2 chiffres
# de fin de chaîne est décimal ; le reste = séparateurs de milliers.
dec = re.search(r"[.,](\d{1,2})$", token)
if dec:
int_part = token[: dec.start()]
int_part = re.sub(r"[^\d]", "", int_part)
value_str = f"{int_part}.{dec.group(1)}"
else:
value_str = re.sub(r"[^\d]", "", token)
if not value_str or value_str == ".":
return None
try:
return float(value_str)
except ValueError:
return None
def parse_int(text: str) -> Optional[int]:
n = parse_number(text)
if n is None:
return None
return int(round(n))
def parse_price(text: str) -> Optional[float]:
"""Prix « à partir de ». Alias sémantique de parse_number (garde None si absent)."""
return parse_number(text)
# --------------------------------------------------------------------------- #
# Tables Markdown.
# --------------------------------------------------------------------------- #
def _split_row(line: str) -> list[str]:
line = line.strip()
if line.startswith("|"):
line = line[1:]
if line.endswith("|"):
line = line[:-1]
return [c.strip() for c in line.split("|")]
def _is_separator_row(cells: list[str]) -> bool:
return all(re.fullmatch(r":?-{2,}:?", c.strip()) for c in cells if c.strip()) and any(cells)
def find_table_with_header(md: str, header_keywords: list[str]) -> tuple[list[str], list[list[str]]]:
"""(header, rows) de la première table Markdown dont l'en-tête contient TOUS
les mots-clés (insensible casse). ([], []) si aucune table correspondante.
"""
lines = md.splitlines()
keys = [k.lower() for k in header_keywords]
i = 0
while i < len(lines):
line = lines[i]
if line.count("|") >= 2:
header = _split_row(line)
header_txt = " ".join(header).lower()
nxt = lines[i + 1] if i + 1 < len(lines) else ""
if all(k in header_txt for k in keys) and _is_separator_row(_split_row(nxt)):
rows: list[list[str]] = []
j = i + 2
while j < len(lines) and lines[j].count("|") >= 2:
cells = _split_row(lines[j])
if not _is_separator_row(cells):
rows.append(cells)
j += 1
return header, rows
i += 1
return [], []
def find_table(md: str, header_keywords: list[str]) -> list[list[str]]:
"""Lignes de données seules (compat) — voir find_table_with_header."""
return find_table_with_header(md, header_keywords)[1]
def _col(header: list[str], *keywords: str) -> Optional[int]:
"""Index de la 1re colonne dont l'en-tête contient TOUS les mots-clés (casse ignorée)."""
keys = [k.lower() for k in keywords]
for idx, cell in enumerate(header):
low = _clean(cell).lower()
if all(k in low for k in keys):
return idx
return None
# --------------------------------------------------------------------------- #
# Extractions par volet.
# --------------------------------------------------------------------------- #
def parse_typologies(architecture_md: str) -> list[dict[str, Any]]:
"""Tableau §3.2 → liste de typologies. Ignore les lignes-placeholder.
Colonnes attendues (template v1.0) :
Typologie | Nb unités | Surface intérieure | Surface terrasse |
Surface totale | Prix « à partir de » (USD) | Prix (DOP)
"""
header, rows = find_table_with_header(architecture_md, ["typologie", "prix"])
if not header:
return []
# Mapping des colonnes PAR EN-TÊTE (résilient au ré-ordonnancement / ajout de
# colonnes). Une colonne absente de l'en-tête → valeur None (jamais deviner une
# position et risquer d'assigner la mauvaise donnée — contrainte #6).
i_nom = _col(header, "typologie")
i_qte = _col(header, "unit") # « Nb unités »
i_int = _col(header, "intérieure")
if i_int is None:
i_int = _col(header, "interieure")
i_terr = _col(header, "terrasse")
i_tot = _col(header, "totale")
i_usd = _col(header, "usd")
i_dop = _col(header, "dop")
if i_nom is None: # en-tête inexploitable → on ne parse pas de prix au hasard
return []
def cell(cells: list[str], idx: Optional[int]) -> str:
if idx is None or not (0 <= idx < len(cells)):
return ""
return cells[idx]
typologies: list[dict[str, Any]] = []
for cells in rows:
if _is_absent(cell(cells, i_nom)): # ligne entièrement placeholder → on saute
continue
typologies.append(
{
"nom": _clean(cell(cells, i_nom)),
"quantite": parse_int(cell(cells, i_qte)),
"surface_interieure_m2": parse_number(cell(cells, i_int)),
"surface_terrasse_m2": parse_number(cell(cells, i_terr)),
"surface_totale_m2": parse_number(cell(cells, i_tot)),
"prix_depuis_usd": parse_price(cell(cells, i_usd)),
"prix_depuis_dop": parse_price(cell(cells, i_dop)),
}
)
return typologies
def _parse_bullets_after(md: str, heading_keywords: list[str]) -> list[str]:
"""Puces (- / * / •) sous le premier titre/ligne contenant tous les mots-clés."""
lines = md.splitlines()
keys = [k.lower() for k in heading_keywords]
items: list[str] = []
capturing = False
for line in lines:
low = line.lower()
if not capturing:
if line.lstrip().startswith("#") and all(k in low for k in keys):
capturing = True
continue
if line.lstrip().startswith("#"): # section suivante → stop
break
m = re.match(r"\s*[-*•]\s+(.*)", line)
if m:
val = _clean(m.group(1))
if val and not _is_absent(val):
items.append(val)
return items
def parse_services_inclus(paysage_md: str) -> list[str]:
"""§3.3 « Services inclus » (+ §3.2 amenities en complément)."""
services = _parse_bullets_after(paysage_md, ["services", "inclus"])
amenities = _parse_bullets_after(paysage_md, ["amenities"])
# Dédoublonnage en conservant l'ordre.
seen: set[str] = set()
out: list[str] = []
for item in services + amenities:
key = item.lower()
if key not in seen:
seen.add(key)
out.append(item)
return out
def parse_localisation(brief_md: str) -> Optional[str]:
"""Ligne « Localisation : … » du brief (§1.1)."""
for line in brief_md.splitlines():
m = re.match(r"\s*[-*]?\s*\**localisation\**\s*[:]\s*(.+)", line, re.IGNORECASE)
if m:
val = _clean(m.group(1))
if val and not _is_absent(val):
return val
return None
def parse_nom(brief_md: str, code: str) -> str:
"""Nom commercial : 1er titre H1 du brief, sinon le code projet."""
for line in brief_md.splitlines():
m = re.match(r"\s*#\s+(.+)", line)
if m:
return _clean(m.group(1))
return code
def parse_positionnement_fr(commercial_md: str) -> Optional[str]:
"""Positionnement FR = 1er paragraphe non-titre de commercial.md.
NOTE : la génération EN/ES multilingue est le livrable Semaine 6 ; ici on ne
remplit que le FR à partir de la source documentée (jamais inventé — #6).
"""
for block in re.split(r"\n\s*\n", commercial_md):
block = block.strip()
if not block or block.startswith("#") or block.startswith("|"):
continue
text = _clean(block.replace("\n", " "))
if len(text) >= 20:
return text
return None
def parse_rendus(photos_dir: str) -> list[dict[str, Any]]:
"""Rendus liés au projet réel (jamais IA générique — contrainte CLAUDE.md).
Deux sources possibles :
1. Un manifeste `60_photos_site/README.md` (table : Fichier | Vue | Hero).
2. À défaut, la liste des fichiers image du dossier.
"""
rendus: list[dict[str, Any]] = []
if not os.path.isdir(photos_dir):
return rendus
manifest = os.path.join(photos_dir, "README.md")
if os.path.isfile(manifest):
with open(manifest, encoding="utf-8") as fh:
rows = find_table(fh.read(), ["fichier"])
for cells in rows:
cells = cells + [""] * (3 - len(cells))
fichier = _clean(cells[0])
if _is_absent(cells[0]):
continue
entry: dict[str, Any] = {"fichier": fichier}
vue = _clean(cells[1])
if vue and not _is_absent(cells[1]):
entry["vue"] = vue
entry["hero"] = "hero" in _clean(cells[2]).lower() or "oui" in _clean(cells[2]).lower()
rendus.append(entry)
if rendus:
return rendus
# Fallback : fichiers image présents.
exts = (".jpg", ".jpeg", ".png", ".webp", ".avif")
for fn in sorted(os.listdir(photos_dir)):
if fn.lower().endswith(exts):
rendus.append({"fichier": fn, "hero": False})
if rendus:
rendus[0]["hero"] = True # 1er rendu = hero par défaut
return rendus
# --------------------------------------------------------------------------- #
# Mapping de statut + assemblage projet.
# --------------------------------------------------------------------------- #
_STATUT_MAP = {
"complete": "disponible",
"en_developpement": "en_developpement",
"incomplete": "en_developpement",
}
def map_statut(version_statut: str) -> str:
return _STATUT_MAP.get(version_statut, "en_developpement")
def _read(path: str) -> str:
if os.path.isfile(path):
with open(path, encoding="utf-8") as fh:
return fh.read()
return ""
def parse_projet(projet_dir: str) -> dict[str, Any]:
"""Lit un dossier `data_room/PXX/` → dict `projet` (contrat schéma).
Lève FileNotFoundError si `_META/version.json` absent (donnée de vérité
obligatoire : sans elle on ne connaît ni le statut ni le score, on refuse
d'inventer).
"""
meta_path = os.path.join(projet_dir, "_META", "version.json")
if not os.path.isfile(meta_path):
raise FileNotFoundError(f"version.json manquant : {meta_path}")
with open(meta_path, encoding="utf-8") as fh:
version = json.load(fh)
code = version["projet"]
brief_md = _read(os.path.join(projet_dir, "00_brief", "brief.md"))
architecture_md = _read(os.path.join(projet_dir, "20_architecture", "architecture.md"))
paysage_md = _read(os.path.join(projet_dir, "30_paysage_experience", "paysage_experience.md"))
commercial_md = _read(os.path.join(projet_dir, "40_llm_outputs", "commercial.md"))
typologies = parse_typologies(architecture_md)
statut = map_statut(version.get("statut_faisabilite", "incomplete"))
# Rétrogradation défensive (contrainte #6) : « disponible » exige des prix USD
# numériques sur toutes les typologies, sinon on ne publie pas de prix douteux.
if statut == "disponible":
prix_ok = bool(typologies) and all(
isinstance(t["prix_depuis_usd"], (int, float)) for t in typologies
)
if not prix_ok:
statut = "en_developpement"
projet: dict[str, Any] = {
"code": code,
"nom": parse_nom(brief_md, code),
"statut": statut,
"localisation": parse_localisation(brief_md) or "",
# Les typologies (avec prix null tolérés) restent exposées même en
# « en_developpement » : le générateur choisit d'afficher ou non la grille.
"typologies": typologies,
"inclus": parse_services_inclus(paysage_md),
"rendus": parse_rendus(os.path.join(projet_dir, "60_photos_site")),
"source": {
"template_version": version.get("template_version", ""),
"score_4big": version.get("score_4big", 0),
"fichiers": _sources_presentes(projet_dir),
},
}
positionnement_fr = parse_positionnement_fr(commercial_md)
if positionnement_fr:
projet["positionnement"] = {"fr": positionnement_fr}
return projet
def _sources_presentes(projet_dir: str) -> list[str]:
"""Chemins relatifs des sources effectivement lues (traçabilité #6)."""
candidates = [
os.path.join("_META", "version.json"),
os.path.join("00_brief", "brief.md"),
os.path.join("20_architecture", "architecture.md"),
os.path.join("30_paysage_experience", "paysage_experience.md"),
os.path.join("40_llm_outputs", "commercial.md"),
os.path.join("60_photos_site", "README.md"),
]
return [c for c in candidates if os.path.isfile(os.path.join(projet_dir, c))]
def build_master(
data_room_dir: str,
generated_at: Optional[str] = None,
template_version: str = "1.0.0",
) -> dict[str, Any]:
"""Parcourt `data_room/` → objet `projets_master.json` complet et trié par code.
`generated_at` : horodatage ISO-8601 UTC ; par défaut « maintenant ».
Injectable pour des sorties déterministes (tests).
"""
if generated_at is None:
generated_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
projets: list[dict[str, Any]] = []
for name in sorted(os.listdir(data_room_dir)):
pdir = os.path.join(data_room_dir, name)
if re.fullmatch(r"P0[1-9]", name) and os.path.isfile(
os.path.join(pdir, "_META", "version.json")
):
projets.append(parse_projet(pdir))
return {
"generated_at": generated_at,
"template_version": template_version,
"projets": projets,
}
@@ -0,0 +1,159 @@
"""Validateur JSON-Schema draft-07 (sous-ensemble) · zéro dépendance pip.
Pourquoi un validateur maison plutôt que `jsonschema` ?
-----------------------------------------------------
Le gate CI (Gitea Actions, cf. .gitea/workflows/ci.yml) tourne sur un runner
`act_runner` sans installation `pip` (les guards Sprint 1 sont du bash pur).
Ce validateur couvre EXACTEMENT les constructions utilisées par les deux schémas
du contrat Faisabilité↔Publiciste (`version.schema.json`,
`projets_master.schema.json`) : type, required, additionalProperties, enum,
pattern, minLength, minItems, maxItems, uniqueItems, minimum, maximum, items,
properties, $ref (interne « #/definitions/... »), allOf, if/then, const, format.
Les tests (`tests/test_publiciste.py`) utilisent en plus la bibliothèque
`jsonschema` comme oracle *quand elle est disponible*, pour se prémunir d'un
écart entre ce validateur et la spec draft-07. En production le validateur
maison suffit et reste autonome.
API : validate(instance, schema) -> list[str] (liste d'erreurs, vide si OK).
"""
from __future__ import annotations
import re
from typing import Any
class SchemaError(Exception):
"""Schéma mal formé (bug du schéma, pas de la donnée)."""
def _resolve_ref(ref: str, root: dict) -> dict:
if not ref.startswith("#/"):
raise SchemaError(f"$ref non supporté (interne uniquement) : {ref}")
node: Any = root
for part in ref[2:].split("/"):
part = part.replace("~1", "/").replace("~0", "~")
node = node[part]
return node
def _type_ok(value: Any, expected: str) -> bool:
if expected == "object":
return isinstance(value, dict)
if expected == "array":
return isinstance(value, list)
if expected == "string":
return isinstance(value, str)
if expected == "integer":
# bool est un int en Python — on l'exclut explicitement.
return isinstance(value, int) and not isinstance(value, bool)
if expected == "number":
return isinstance(value, (int, float)) and not isinstance(value, bool)
if expected == "boolean":
return isinstance(value, bool)
if expected == "null":
return value is None
raise SchemaError(f"type inconnu dans le schéma : {expected}")
# Formats vérifiés (les autres sont acceptés sans contrôle, comme le veut draft-07).
_DATE_TIME = re.compile(
r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(\.\d+)?(Z|[+-]\d{2}:\d{2})?$"
)
def _check_format(value: str, fmt: str, path: str, errors: list[str]) -> None:
if fmt == "date-time" and not _DATE_TIME.match(value):
errors.append(f"{path}: format date-time invalide ({value!r})")
def _validate(value: Any, schema: dict, root: dict, path: str, errors: list[str]) -> None:
if "$ref" in schema:
schema = _resolve_ref(schema["$ref"], root)
# const
if "const" in schema and value != schema["const"]:
errors.append(f"{path}: attendu const={schema['const']!r}, reçu {value!r}")
# enum
if "enum" in schema and value not in schema["enum"]:
errors.append(f"{path}: {value!r} hors enum {schema['enum']}")
# type (peut être une liste de types alternatifs)
if "type" in schema:
types = schema["type"]
types = [types] if isinstance(types, str) else types
if not any(_type_ok(value, t) for t in types):
errors.append(f"{path}: type {type(value).__name__}{types}")
# Inutile d'aller plus loin si le type de base est faux.
return
if isinstance(value, str):
if "minLength" in schema and len(value) < schema["minLength"]:
errors.append(f"{path}: chaîne trop courte (min {schema['minLength']})")
if "pattern" in schema and not re.search(schema["pattern"], value):
errors.append(f"{path}: {value!r} ne matche pas /{schema['pattern']}/")
if "format" in schema:
_check_format(value, schema["format"], path, errors)
if isinstance(value, (int, float)) and not isinstance(value, bool):
if "minimum" in schema and value < schema["minimum"]:
errors.append(f"{path}: {value} < minimum {schema['minimum']}")
if "maximum" in schema and value > schema["maximum"]:
errors.append(f"{path}: {value} > maximum {schema['maximum']}")
if isinstance(value, list):
if "minItems" in schema and len(value) < schema["minItems"]:
errors.append(f"{path}: {len(value)} items < minItems {schema['minItems']}")
if "maxItems" in schema and len(value) > schema["maxItems"]:
errors.append(f"{path}: {len(value)} items > maxItems {schema['maxItems']}")
if schema.get("uniqueItems") and _has_duplicates(value):
errors.append(f"{path}: items non uniques")
if "items" in schema:
for i, item in enumerate(value):
_validate(item, schema["items"], root, f"{path}[{i}]", errors)
if isinstance(value, dict):
props = schema.get("properties", {})
for req in schema.get("required", []):
if req not in value:
errors.append(f"{path}: propriété requise absente « {req} »")
if schema.get("additionalProperties") is False:
extra = set(value) - set(props)
if extra:
errors.append(f"{path}: propriétés interdites {sorted(extra)}")
for key, sub in props.items():
if key in value:
_validate(value[key], sub, root, f"{path}.{key}", errors)
# Combinateurs
for sub in schema.get("allOf", []):
_validate(value, sub, root, path, errors)
if "if" in schema:
cond_errors: list[str] = []
_validate(value, schema["if"], root, path, cond_errors)
branch = "then" if not cond_errors else "else"
if branch in schema:
_validate(value, schema[branch], root, path, errors)
def _has_duplicates(items: list) -> bool:
seen: list = []
for it in items:
if it in seen:
return True
seen.append(it)
return False
def validate(instance: Any, schema: dict) -> list[str]:
"""Valide `instance` contre `schema`. Retourne la liste des erreurs (vide = OK)."""
errors: list[str] = []
_validate(instance, schema, schema, "$", errors)
return errors
def is_valid(instance: Any, schema: dict) -> bool:
return not validate(instance, schema)