Claude Code DTP Worker
a2c2e0b097
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
443 lines
17 KiB
Python
443 lines
17 KiB
Python
"""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 (1er paragraphe non-titre)
|
||
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,
|
||
}
|