Files
open-school/backend/app/program_content.py
2026-05-12 21:31:08 +02:00

619 lines
25 KiB
Python

import base64
import html
import json
import os
import re
from datetime import datetime
from functools import lru_cache
from pathlib import Path
from urllib.parse import quote
def default_content_root() -> Path:
current = Path(__file__).resolve()
candidates = [
current.parents[1] / "contenus_pedagogiques",
current.parents[1] / "Programme" / "contenus_pedagogiques",
]
candidates.extend(parent / "Programme" / "contenus_pedagogiques" for parent in current.parents)
candidates.extend(
[
Path("/programme/contenus_pedagogiques"),
Path("/app/contenus_pedagogiques"),
Path("/app/Programme/contenus_pedagogiques"),
]
)
return next((candidate for candidate in candidates if candidate.exists()), candidates[0])
CONTENT_ROOT = Path(os.getenv("PROGRAM_CONTENT_ROOT", default_content_root())).resolve()
LESSON_BACKGROUND = "#f8fafc"
REVIEW_STATE_PATH = Path(os.getenv("PROGRAM_REVIEW_STATE_PATH", CONTENT_ROOT / ".program_review_state.json"))
REVIEW_UPDATES_PATH = Path(os.getenv("PROGRAM_REVIEW_UPDATES_PATH", CONTENT_ROOT / "review_updates"))
GRADE_ALIASES = {
"CM1": "cm1",
"CM2": "cm2",
"6e": "6e",
"6eme": "6e",
"6ieme": "6e",
"5e": "5e",
"5eme": "5e",
"5ieme": "5e",
"4e": "4e",
"4eme": "4e",
"4ieme": "4e",
"3e": "3e",
"3eme": "3e",
"3ieme": "3e",
}
def normalize_grade(grade: str) -> str:
compact = (grade or "").strip().replace("è", "e").replace("é", "e")
return GRADE_ALIASES.get(compact, compact.lower())
def title_from_slug(slug: str) -> str:
parts = slug.split("_")
if parts and parts[0][:2].isdigit():
parts = parts[1:]
return " ".join(part.capitalize() for part in parts if part)
def encode_asset_path(relative_path: str) -> str:
payload = relative_path.encode("utf-8")
return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=")
def decode_asset_token(token: str) -> Path:
padding = "=" * (-len(token) % 4)
relative = base64.urlsafe_b64decode(f"{token}{padding}").decode("utf-8")
path = (CONTENT_ROOT / relative).resolve()
if CONTENT_ROOT not in path.parents:
raise ValueError("Chemin de ressource invalide")
return path
def build_asset_url(relative_path: str) -> str:
return f"/api/program/assets/{quote(encode_asset_path(relative_path))}"
def build_section_url(relative_path: str, section_index: int) -> str:
token = quote(encode_asset_path(relative_path))
return f"/api/program/assets/{token}/sections/{section_index}"
def find_section_context_path(svg_path: Path, asset_number: int, section_number: int) -> Path | None:
fiche_slug = svg_path.parent.name if svg_path.name == "fiche.svg" else svg_path.stem
lesson_dir = svg_path.parents[2] if svg_path.name == "fiche.svg" else svg_path.parent.parent
canonical_context_dir = lesson_dir / "cards" / fiche_slug
context_dir = svg_path.parent.parent / "card_contexts" / svg_path.stem
candidates = [
canonical_context_dir / f"card_{section_number:02d}.md",
context_dir / f"Fiche{asset_number}Card{section_number}.md",
context_dir / f"fiche_{asset_number:02d}_card_{section_number:02d}.md",
context_dir / f"card_{section_number:02d}.md",
]
return next((candidate for candidate in candidates if candidate.exists()), None)
def find_section_svg_path(svg_path: Path, section_number: int) -> Path | None:
fiche_slug = svg_path.parent.name if svg_path.name == "fiche.svg" else svg_path.stem
lesson_dir = svg_path.parents[2] if svg_path.name == "fiche.svg" else svg_path.parent.parent
candidate = lesson_dir / "cards" / fiche_slug / f"card_{section_number:02d}.svg"
return candidate if candidate.exists() else None
def svg_attr(attrs: str, name: str, default: float | None = None) -> float | None:
match = re.search(rf'\b{name}="(?P<value>[-\d.]+)"', attrs)
if match:
return float(match.group("value"))
return default
def extract_svg_sections(svg_path: Path, asset_number: int | None = None) -> list[dict]:
text = svg_path.read_text(encoding="utf-8", errors="ignore")
sections: list[dict] = []
pattern = re.compile(
r'<g\s+transform="translate\((?P<x>[-\d.]+)\s+(?P<y>[-\d.]+)\)">\s*'
r'<rect\b(?P<attrs>[^>]*)/?>'
r'(?P<body>.*?)</g>',
re.DOTALL,
)
for match in pattern.finditer(text):
attrs = match.group("attrs")
rect_x = svg_attr(attrs, "x", 0)
rect_y = svg_attr(attrs, "y", 0)
width = svg_attr(attrs, "width")
height = svg_attr(attrs, "height")
if rect_x != 0 or rect_y != 0 or width is None or height is None:
continue
if width < 250 or height < 100:
continue
title_match = re.search(r'<text[^>]*>(?P<title>.*?)</text>', match.group("body"), re.DOTALL)
raw_title = re.sub(r"<[^>]+>", "", title_match.group("title") if title_match else "")
title = html.unescape(raw_title).strip() or f"Étape {len(sections) + 1}"
margin = 28
section_number = len(sections) + 1
context_path = find_section_context_path(svg_path, asset_number or 1, section_number)
section_svg_path = find_section_svg_path(svg_path, section_number)
section_relative = section_svg_path.relative_to(CONTENT_ROOT).as_posix() if section_svg_path else ""
sections.append(
{
"index": len(sections),
"title": title,
"context_key": f"Fiche{asset_number or 1}Card{section_number}",
"path": section_relative,
"context": context_path.read_text(encoding="utf-8", errors="ignore").strip()
if context_path
else "",
"view_box": [
max(float(match.group("x")) - margin, 0),
max(float(match.group("y")) - margin, 0),
width + margin * 2,
height + margin * 2,
],
}
)
return sections
def render_svg_section(path: Path, section_index: int) -> str:
sections = extract_svg_sections(path)
if section_index < 0 or section_index >= len(sections):
raise IndexError("Section introuvable")
x, y, width, height = sections[section_index]["view_box"]
text = path.read_text(encoding="utf-8", errors="ignore")
first_group_index = text.find("<g ")
if first_group_index > -1:
before_groups = text[:first_group_index]
after_groups = text[first_group_index:]
before_groups = re.sub(
r"\.bg\s*\{[^}]*\}",
f".bg {{ fill: {LESSON_BACKGROUND}; }}",
before_groups,
count=1,
)
before_groups = re.sub(
r"\.panel\s*\{[^}]*\}",
(
".panel { fill: #ffffff; stroke: #d5e0dc; stroke-width: 3; "
"filter: drop-shadow(0 10px 24px rgba(22, 42, 45, 0.08)); }"
),
before_groups,
count=1,
)
before_groups = re.sub(
r'(<rect\b(?=[^>]*class="panel")(?=[^>]*/>)[^>]*)/>',
rf'\1 style="fill:{LESSON_BACKGROUND};stroke:none;filter:none"/>',
before_groups,
count=1,
)
text = before_groups + after_groups
text = re.sub(r'\swidth="[^"]+"', f' width="{int(width)}"', text, count=1)
text = re.sub(r'\sheight="[^"]+"', f' height="{int(height)}"', text, count=1)
text = re.sub(
r'\sviewBox="[^"]+"',
f' viewBox="{x:g} {y:g} {width:g} {height:g}"',
text,
count=1,
)
return text
def discover_lesson_dirs() -> list[Path]:
lesson_dirs: set[Path] = set()
for svg_dir in CONTENT_ROOT.glob("cycle_*/*/*/**/svg"):
if svg_dir.is_dir() and "kit_adaptatif" not in svg_dir.relative_to(CONTENT_ROOT).parts:
lesson_dirs.add(svg_dir.parent)
for fiches_dir in CONTENT_ROOT.glob("cycle_*/*/*/**/fiches"):
if fiches_dir.is_dir() and list(fiches_dir.glob("*/fiche.svg")):
lesson_dirs.add(fiches_dir.parent)
return sorted(lesson_dirs)
def lesson_fiche_svg_paths(lesson_dir: Path) -> list[Path]:
canonical = sorted((lesson_dir / "fiches").glob("*/fiche.svg"))
if canonical:
return canonical
return sorted((lesson_dir / "svg").glob("*.svg"))
def fiche_markdown_for_svg(svg_path: Path) -> Path | None:
if svg_path.name == "fiche.svg":
candidate = svg_path.with_name("fiche.md")
return candidate if candidate.exists() else None
candidate = svg_path.parent.parent / f"{svg_path.stem}.md"
return candidate if candidate.exists() else None
@lru_cache(maxsize=1)
def list_lessons() -> list[dict]:
lessons: list[dict] = []
if not CONTENT_ROOT.exists():
return lessons
for lesson_dir in discover_lesson_dirs():
svg_files = lesson_fiche_svg_paths(lesson_dir)
if not svg_files:
continue
relative_lesson = lesson_dir.relative_to(CONTENT_ROOT).as_posix()
parts = relative_lesson.split("/")
if len(parts) < 4:
continue
cycle, subject, grade = parts[0], parts[1], parts[2]
title = title_from_slug(lesson_dir.name)
readme = lesson_dir / "README.md"
if readme.exists():
first_heading = next(
(
line.lstrip("#").strip()
for line in readme.read_text(encoding="utf-8", errors="ignore").splitlines()
if line.startswith("#")
),
"",
)
title = first_heading or title
assets = []
for asset_index, svg_path in enumerate(svg_files, start=1):
relative_path = svg_path.relative_to(CONTENT_ROOT).as_posix()
sections = extract_svg_sections(svg_path, asset_index)
fiche_md = fiche_markdown_for_svg(svg_path)
fiche_slug = svg_path.parent.name if svg_path.name == "fiche.svg" else svg_path.stem
assets.append(
{
"name": fiche_slug,
"title": title_from_slug(fiche_slug),
"path": relative_path,
"url": build_asset_url(relative_path),
"content": read_review_text(fiche_md) if fiche_md else "",
"sections": [
{
**section,
"url": build_asset_url(section["path"])
if section.get("path")
else build_section_url(relative_path, section["index"]),
}
for section in sections
],
}
)
lessons.append(
{
"id": relative_lesson,
"title": title,
"subject": title_from_slug(subject),
"grade": grade,
"cycle": cycle,
"asset_count": len(assets),
"assets": assets,
"lesson_context": read_review_text(lesson_dir / "lecon.md") if (lesson_dir / "lecon.md").exists() else "",
"adaptive_kit": collect_adaptive_kit(lesson_dir),
}
)
return lessons
def list_lessons_for_grade(grade: str) -> list[dict]:
normalized = normalize_grade(grade)
return [lesson for lesson in list_lessons() if lesson["grade"] == normalized]
def get_content_status() -> dict:
lessons = list_lessons()
by_grade: dict[str, int] = {}
for lesson in lessons:
by_grade[lesson["grade"]] = by_grade.get(lesson["grade"], 0) + 1
return {
"content_root": str(CONTENT_ROOT),
"content_root_exists": CONTENT_ROOT.exists(),
"lesson_count": len(lessons),
"lessons_by_grade": by_grade,
}
def get_lesson(lesson_id: str) -> dict | None:
return next((lesson for lesson in list_lessons() if lesson["id"] == lesson_id), None)
def load_review_state() -> dict:
if not REVIEW_STATE_PATH.exists():
return {}
try:
return json.loads(REVIEW_STATE_PATH.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return {}
def save_review_state(state: dict) -> None:
REVIEW_STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
REVIEW_STATE_PATH.write_text(
json.dumps(state, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def iter_review_update_files() -> list[Path]:
candidates = []
single_file = CONTENT_ROOT / ".program_review_updates.json"
if single_file.exists():
candidates.append(single_file)
if REVIEW_UPDATES_PATH.exists():
candidates.extend(sorted(REVIEW_UPDATES_PATH.glob("*.json")))
return candidates
def apply_program_review_updates() -> dict:
state = load_review_state()
applied = 0
skipped = 0
errors = []
for path in iter_review_update_files():
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
errors.append({"file": str(path), "error": f"JSON invalide: {exc}"})
continue
if payload.get("schema") != "professeur-top-review-updates-v1":
skipped += 1
continue
payload_updated_at = payload.get("updated_at") or datetime.utcnow().isoformat(timespec="seconds") + "Z"
payload_reviewer = payload.get("updated_by") or "codex"
for index, update in enumerate(payload.get("updates") or []):
node_key = update.get("node_key") or update.get("path")
if not node_key:
skipped += 1
continue
update_id = update.get("update_id") or f"{path.name}:{index}:{node_key}"
existing = state.get(node_key, {})
if existing.get("codex_update_id") == update_id:
skipped += 1
continue
next_review = {
**existing,
"status": update.get("status") or existing.get("status") or "treated",
"comment": update.get("comment", existing.get("comment", "")),
"reviewer": update.get("reviewer") or payload_reviewer,
"updated_at": update.get("updated_at") or payload_updated_at,
"codex_update_id": update_id,
}
if update.get("treated_note"):
next_review["treated_note"] = update["treated_note"]
state[node_key] = next_review
applied += 1
if applied:
save_review_state(state)
return {
"applied": applied,
"skipped": skipped,
"errors": errors,
"files": [str(path) for path in iter_review_update_files()],
}
def collect_lesson_status(lesson: dict) -> dict:
lesson_dir = CONTENT_ROOT / lesson["id"]
legacy_kit_dir = lesson_dir / "kit_adaptatif"
fiche_md = list((lesson_dir / "fiches").glob("*/fiche.md"))
fiche_svg = list((lesson_dir / "fiches").glob("*/fiche.svg"))
card_md = list((lesson_dir / "cards").glob("*/*.md"))
card_svg = list((lesson_dir / "cards").glob("*/*.svg"))
tests_dir = lesson_dir / "tests"
parcours_dir = lesson_dir / "exercices" / "parcours_adaptes"
return {
"lesson_md": 1 if (lesson_dir / "lecon.md").exists() else 0,
"micro_fiches": len(fiche_md) or len([path for path in lesson_dir.glob("*.md") if path.name.startswith(tuple("0123456789"))]),
"lesson_svg": len(fiche_svg) or len(lesson.get("assets") or []),
"cards": len(card_svg) or sum(len(asset.get("sections") or []) for asset in lesson.get("assets") or []),
"contexts": len(card_md) or len(list((lesson_dir / "card_contexts").rglob("Fiche*Card*.md"))),
"text_exercises": len(list((lesson_dir / "exercices").rglob("exercices.md"))),
"exercise_svg": len(list((lesson_dir / "exercices").rglob("exercices.svg"))) or len(list((lesson_dir / "exercices_svg").glob("*.svg"))),
"tests": len([path for path in tests_dir.glob("*.md") if path.name != "README.md"])
if tests_dir.exists()
else len([path for path in (legacy_kit_dir / "test_diagnostic").glob("*.md") if path.name != "README.md"]) if legacy_kit_dir.exists() else 0,
"test_svg": len(list(tests_dir.glob("*.svg"))) if tests_dir.exists() else len(list((legacy_kit_dir / "svg").glob("*.svg"))) if legacy_kit_dir.exists() else 0,
"adaptive_profiles": len([path for path in parcours_dir.rglob("*.md") if path.name != "README.md"])
if parcours_dir.exists()
else len([path for path in (legacy_kit_dir / "parcours_profils").glob("*.md") if path.name != "README.md"]) if legacy_kit_dir.exists() else 0,
"validations": sorted(path.stem for path in lesson_dir.glob("VALIDATION_PHASE_*.md")),
}
def append_tree_child(parent: dict, key: str, label: str, node_type: str, path: str = "") -> dict:
children = parent.setdefault("children", [])
for child in children:
if child["key"] == key:
return child
child = {"key": key, "label": label, "type": node_type, "path": path, "children": []}
children.append(child)
return child
def read_review_text(path: Path, max_chars: int = 6000) -> str:
text = path.read_text(encoding="utf-8", errors="ignore").strip()
return text[:max_chars]
def collect_adaptive_kit(lesson_dir: Path) -> dict | None:
tests_dir = lesson_dir / "tests"
parcours_dir = lesson_dir / "exercices" / "parcours_adaptes"
legacy_kit_dir = lesson_dir / "kit_adaptatif"
if not tests_dir.exists() and not parcours_dir.exists() and not legacy_kit_dir.exists():
return None
def markdown_items(base: Path) -> list[dict]:
if not base.exists():
return []
return [
{
"id": path.relative_to(CONTENT_ROOT).as_posix(),
"title": title_from_slug(path.parent.name if path.stem == "exercices" else path.stem),
"content": read_review_text(path),
}
for path in sorted(base.rglob("*.md"))
if path.name != "README.md"
]
def svg_items(base: Path) -> list[dict]:
if not base.exists():
return []
items = []
for path in sorted(base.rglob("*.svg")):
relative = path.relative_to(CONTENT_ROOT).as_posix()
items.append(
{
"id": relative,
"title": title_from_slug(path.stem),
"url": build_asset_url(relative),
}
)
return items
diagnostics = markdown_items(tests_dir)
profiles = markdown_items(parcours_dir)
supports = svg_items(tests_dir) + svg_items(parcours_dir)
if legacy_kit_dir.exists():
diagnostics = diagnostics or markdown_items(legacy_kit_dir / "test_diagnostic")
profiles = profiles or markdown_items(legacy_kit_dir / "parcours_profils")
supports = supports or svg_items(legacy_kit_dir / "svg")
return {
"id": lesson_dir.relative_to(CONTENT_ROOT).as_posix(),
"title": "Tests et exercices adaptes",
"diagnostics": diagnostics,
"profiles": profiles,
"supports": supports,
}
def format_adaptive_kit_context(lesson: dict | None) -> str:
kit = (lesson or {}).get("adaptive_kit")
if not kit:
return ""
diagnostics = ", ".join(item["title"] for item in kit.get("diagnostics") or [])
profiles = ", ".join(item["title"] for item in kit.get("profiles") or [])
return (
"Tests et exercices adaptes disponibles pour cette leçon.\n"
f"Tests diagnostics: {diagnostics or 'aucun'}.\n"
f"Parcours de reprise: {profiles or 'aucun'}.\n"
"Règle adaptative: si l'élève se trompe nettement, semble bloqué, inverse les notions ou demande de l'aide, "
"ne change pas de card tout de suite. Propose une micro-question de diagnostic liée à la difficulté observée, "
"puis oriente vers le parcours le plus probable en une phrase courte.\n"
)
def build_program_tree() -> dict:
reviews = load_review_state()
root = {
"key": "program",
"label": "Programme",
"type": "root",
"path": "",
"children": [],
"review": reviews.get("program", {}),
}
for lesson in list_lessons():
parts = lesson["id"].split("/")
cycle, subject, grade = parts[0], parts[1], parts[2]
theme_parts = parts[3:-1]
lesson_slug = parts[-1]
cycle_node = append_tree_child(root, cycle, title_from_slug(cycle), "cycle", cycle)
grade_key = f"{cycle}/{grade}"
grade_node = append_tree_child(cycle_node, grade_key, grade.upper(), "grade", grade_key)
subject_key = f"{grade_key}/{subject}"
subject_node = append_tree_child(grade_node, subject_key, title_from_slug(subject), "subject", subject_key)
parent = subject_node
current_path = f"{cycle}/{subject}/{grade}"
for theme_slug in theme_parts:
current_path = f"{current_path}/{theme_slug}"
parent = append_tree_child(parent, current_path, title_from_slug(theme_slug), "theme", current_path)
lesson_node = append_tree_child(parent, lesson["id"], lesson["title"], "lesson", lesson["id"])
lesson_node["status"] = collect_lesson_status(lesson)
lesson_node["review"] = reviews.get(lesson["id"], {})
lesson_node["assets"] = lesson.get("assets") or []
lesson_node["content"] = lesson.get("lesson_context") or ""
for asset in lesson.get("assets") or []:
asset_key = asset["path"]
asset_node = append_tree_child(lesson_node, asset_key, asset["title"], "fiche", asset_key)
asset_node["url"] = asset.get("url")
asset_node["content"] = asset.get("content") or ""
asset_node["review"] = reviews.get(asset_key, {})
for section in asset.get("sections") or []:
section_key = section.get("path") or f"{asset_key}#{section.get('context_key') or section.get('index')}"
section_node = append_tree_child(
asset_node,
section_key,
section.get("title") or section_key,
"card",
section_key,
)
section_node["url"] = section.get("url")
section_node["content"] = section.get("context") or ""
section_node["review"] = reviews.get(section_key, {})
lesson_dir = CONTENT_ROOT / lesson["id"]
exercises_dir = lesson_dir / "exercices"
if exercises_dir.exists():
exercises_node = append_tree_child(lesson_node, f"{lesson['id']}/exercices", "Exercices", "group", f"{lesson['id']}/exercices")
for exercise_md in sorted(exercises_dir.rglob("*.md")):
if exercise_md.name == "README.md":
continue
relative = exercise_md.relative_to(CONTENT_ROOT).as_posix()
exercise_node = append_tree_child(exercises_node, relative, title_from_slug(exercise_md.parent.name), "exercice", relative)
exercise_node["content"] = read_review_text(exercise_md)
exercise_svg = exercise_md.with_suffix(".svg") if exercise_md.stem != "exercices" else exercise_md.with_name("exercices.svg")
if exercise_svg.exists():
exercise_node["url"] = build_asset_url(exercise_svg.relative_to(CONTENT_ROOT).as_posix())
exercise_node["review"] = reviews.get(relative, {})
for exercise_svg in sorted(exercises_dir.rglob("*.svg")):
if exercise_svg.with_suffix(".md").exists() or (exercise_svg.name == "exercices.svg" and (exercise_svg.parent / "exercices.md").exists()):
continue
relative = exercise_svg.relative_to(CONTENT_ROOT).as_posix()
support_node = append_tree_child(exercises_node, relative, title_from_slug(exercise_svg.stem), "support", relative)
support_node["url"] = build_asset_url(relative)
support_node["review"] = reviews.get(relative, {})
tests_dir = lesson_dir / "tests"
if tests_dir.exists():
tests_node = append_tree_child(lesson_node, f"{lesson['id']}/tests", "Tests", "group", f"{lesson['id']}/tests")
for test_md in sorted(tests_dir.glob("*.md")):
if test_md.name == "README.md":
continue
relative = test_md.relative_to(CONTENT_ROOT).as_posix()
test_node = append_tree_child(tests_node, relative, title_from_slug(test_md.stem), "test", relative)
test_node["content"] = read_review_text(test_md)
test_node["review"] = reviews.get(relative, {})
for test_svg in sorted(tests_dir.glob("*.svg")):
relative = test_svg.relative_to(CONTENT_ROOT).as_posix()
support_node = append_tree_child(tests_node, relative, title_from_slug(test_svg.stem), "support", relative)
support_node["url"] = build_asset_url(relative)
support_node["review"] = reviews.get(relative, {})
return root
def update_program_review(node_key: str, status: str, comment: str, reviewer: str) -> dict:
state = load_review_state()
state[node_key] = {
"status": status,
"comment": comment,
"reviewer": reviewer,
"updated_at": datetime.utcnow().isoformat(timespec="seconds") + "Z",
}
save_review_state(state)
return state[node_key]