Backend: regeln.py (Lernregeln zentral), Stats O(n), GET /guides/locks, _norm_titel gehärtet

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/database.py

@@ -216,6 +216,17 @@ async def delete_element(element_id: str) -> bool:
 
 # --- Kapitel-Fortschritt ---
 
+async def list_progress_all() -> dict[str, set[str]]:
+    """Kompletter Kapitel-Fortschritt in einem Query: guide_id → Kapitel-Titel."""
+    db = await get_db()
+    cursor = await db.execute("SELECT guide_id, chapter FROM guide_progress")
+    rows = await cursor.fetchall()
+    out: dict[str, set[str]] = {}
+    for guide_id, chapter in rows:
+        out.setdefault(guide_id, set()).add(chapter)
+    return out
+
+
 async def list_progress(guide_id: str) -> list[str]:
     db = await get_db()
     cursor = await db.execute(

backend/regeln.py

@@ -0,0 +1,88 @@
+"""Lernschulden-Regeln: Progression und Deckel für offene Guides — die EINZIGE Quelle.
+
+Regeln (nur Neu-Erstellungen; Themen, Bausteine, OnePager unbegrenzt):
+- JE Format (MiniGuide/Guide/FullGuide) höchstens 3 erstellte, nicht absolvierte Guides
+- Progression pro Thema: Guide erst nach absolviertem MiniGuide, FullGuide erst nach absolviertem Guide
+Alle Funktionen arbeiten auf einmal geladenen Daten (lade_lernstand) — keine
+Query-Schleifen mehr pro Guide.
+"""
+
+import json
+
+from database import list_guides, list_progress_all
+from guide import guide_slot_dateien
+from paths import bausteine_path, guide_content_path
+
+MAX_OFFENE_GUIDES = 3
+VORSTUFE = {"Guide": "MiniGuide", "FullGuide": "Guide"}
+FORMATE = ("MiniGuide", "Guide", "FullGuide")
+
+
+async def lade_lernstand() -> tuple[list[dict], dict[str, set[str]]]:
+    """Guides + kompletter Kapitel-Fortschritt in zwei Queries."""
+    return await list_guides(), await list_progress_all()
+
+
+def _kapitel_titel(topic: str, fmt: str) -> set[str] | None:
+    path = guide_content_path(topic, fmt)
+    if not path.exists():
+        return None
+    try:
+        chapters = json.loads(path.read_text(encoding="utf-8")).get("chapters", [])
+    except ValueError:
+        return None
+    return {c.get("title") for c in chapters}
+
+
+def _neueste_done(guides: list[dict], fmt: str) -> dict[str, dict]:
+    """Pro Thema der neueste fertige Guide dieses Formats."""
+    neueste: dict[str, dict] = {}
+    for g in guides:
+        if g["format"] == fmt and g["status"] == "done":
+            if g["topic"] not in neueste or g["created_at"] > neueste[g["topic"]]["created_at"]:
+                neueste[g["topic"]] = g
+    return neueste
+
+
+def _guide_absolviert(g: dict, progress: dict[str, set[str]]) -> bool:
+    titles = _kapitel_titel(g["topic"], g["format"])
+    return bool(titles) and titles <= progress.get(g["id"], set())
+
+
+def ist_absolviert(topic: str, fmt: str, guides: list[dict], progress: dict[str, set[str]]) -> bool:
+    """Alle Kapitel des neuesten fertigen Guides (Thema+Format) abgehakt?"""
+    g = _neueste_done(guides, fmt).get(topic)
+    return g is not None and _guide_absolviert(g, progress)
+
+
+def formate_stats(guides: list[dict], progress: dict[str, set[str]]) -> dict:
+    """Pro Format erstellt/absolviert — pro Thema zählt nur der neueste fertige Guide."""
+    formate = {}
+    for fmt in FORMATE:
+        neueste = _neueste_done(guides, fmt)
+        absolviert = sum(1 for g in neueste.values() if _guide_absolviert(g, progress))
+        formate[fmt] = {"erstellt": len(neueste), "absolviert": absolviert}
+    return formate
+
+
+def guide_lock(topic: str, fmt: str, guides: list[dict], progress: dict[str, set[str]]) -> str | None:
+    """Grund, warum ein Neu-Start für Thema+Format gesperrt ist — None = erlaubt.
+
+    Exakt die Regeln aus POST /guides: Bausteine nötig, kein Duplikat-Start,
+    Lernschulden nur für echte Neu-Erstellungen (Resume/Regenerieren frei).
+    """
+    if fmt != "OnePager" and not bausteine_path(topic).exists():
+        return "Erst Bausteine erstellen"
+    for g in guides:
+        if g["topic"] == topic and g["format"] == fmt and g["status"] in ("queued", "generating"):
+            return "Generierung läuft bereits"
+    content = guide_content_path(topic, fmt)
+    if fmt != "OnePager" and not content.exists() and not guide_slot_dateien(content):
+        vorstufe = VORSTUFE.get(fmt)
+        if vorstufe and not ist_absolviert(topic, vorstufe, guides, progress):
+            return f"Erst den {vorstufe} dieses Themas absolvieren"
+        stat = formate_stats(guides, progress).get(fmt, {"erstellt": 0, "absolviert": 0})
+        offen = stat["erstellt"] - stat["absolviert"]
+        if offen >= MAX_OFFENE_GUIDES:
+            return f"Erst {fmt}s absolvieren — maximal {MAX_OFFENE_GUIDES} offene erlaubt ({offen} offen)"
+    return None

backend/routes.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import shutil
 import uuid
 from datetime import datetime, timezone
@@ -19,6 +18,7 @@ from bausteine import generate_bausteine, cancel_bausteine, bausteine_status, ac
 from elements import generate_element, chat_with_guide, chat_with_element, check_element, style_element, refine_suggestion
 from guide import generate_guide, guide_slot_dateien
 from pipeline import cancel_guide
+from regeln import FORMATE, formate_stats, guide_lock, ist_absolviert, lade_lernstand
 from models import (
     GuideCreateRequest, GuideResponse,
     TopicCreateRequest,
@@ -29,7 +29,7 @@ from models import (
     ElementRefineRequest, ElementRefineResponse,
     ProgressUpdate, ProgressResponse, ProjectResponse, ProviderInfo,
 )
-from paths import bausteine_path, bausteine_topics, guide_content_path, project_dir, topic_dir
+from paths import bausteine_topics, guide_content_path, project_dir, topic_dir
 
 router = APIRouter(prefix="/api")
 
@@ -50,73 +50,21 @@ async def get_topics():
     return db_topics + sorted(derived - set(db_topics))
 
 
-# Lernschulden-Regeln (nur Neu-Erstellungen; Themen, Bausteine, OnePager unbegrenzt):
-# - JE Format (MiniGuide/Guide/FullGuide) höchstens 3 erstellte, nicht absolvierte Guides
-# - Progression pro Thema: Guide erst nach absolviertem MiniGuide, FullGuide erst nach absolviertem Guide
-MAX_OFFENE_GUIDES = 3
-VORSTUFE = {"Guide": "MiniGuide", "FullGuide": "Guide"}
-
-
-async def _ist_absolviert(topic: str, fmt: str) -> bool:
-    """Alle Kapitel des neuesten fertigen Guides (Thema+Format) abgehakt?"""
-    neueste = None
-    for g in await list_guides():
-        if g["topic"] == topic and g["format"] == fmt and g["status"] == "done":
-            if neueste is None or g["created_at"] > neueste["created_at"]:
-                neueste = g
-    if neueste is None:
-        return False
-    path = guide_content_path(topic, fmt)
-    if not path.exists():
-        return False
-    try:
-        chapters = json.loads(path.read_text(encoding="utf-8")).get("chapters", [])
-    except ValueError:
-        return False
-    titles = {c.get("title") for c in chapters}
-    return bool(titles) and titles <= set(await list_progress(neueste["id"]))
-
-
-async def _formate_stats() -> dict:
-    """Pro Format erstellt/absolviert (alle Kapitel abgehakt) — pro Thema zählt nur der neueste fertige Guide."""
-    guides = await list_guides()
-    formate = {}
-    for fmt in ("MiniGuide", "Guide", "FullGuide"):
-        neueste: dict[str, dict] = {}
-        for g in guides:
-            if g["format"] == fmt and g["status"] == "done":
-                if g["topic"] not in neueste or g["created_at"] > neueste[g["topic"]]["created_at"]:
-                    neueste[g["topic"]] = g
-        absolviert = 0
-        for g in neueste.values():
-            path = guide_content_path(g["topic"], fmt)
-            if not path.exists():
-                continue
-            try:
-                chapters = json.loads(path.read_text(encoding="utf-8")).get("chapters", [])
-            except ValueError:
-                continue
-            titles = {c.get("title") for c in chapters}
-            if titles and titles <= set(await list_progress(g["id"])):
-                absolviert += 1
-        formate[fmt] = {"erstellt": len(neueste), "absolviert": absolviert}
-    return formate
-
-
 @router.get("/stats")
 async def get_stats():
     """Tracker: Themen-Anzahl + pro Format erstellt/absolviert."""
-    guides = await list_guides()
+    guides, progress = await lade_lernstand()
     themen = set(await db_list_topics()) | {g["topic"] for g in guides} | set(bausteine_topics())
     if PROJECTS_DIR.is_dir():
         themen |= {e.name for e in PROJECTS_DIR.iterdir() if e.is_dir()}
-    return {"themen": len(themen), "formate": await _formate_stats()}
+    return {"themen": len(themen), "formate": formate_stats(guides, progress)}
 
 
 @router.get("/topics/fortschritt")
 async def topic_fortschritt(topic: str):
     """Absolviert-Status pro Format — fürs Freischalten der nächsten Ausbaustufe."""
-    return {fmt: await _ist_absolviert(topic, fmt) for fmt in ("MiniGuide", "Guide", "FullGuide")}
+    guides, progress = await lade_lernstand()
+    return {fmt: ist_absolviert(topic, fmt, guides, progress) for fmt in FORMATE}
 
 
 @router.post("/topics")
@@ -194,23 +142,10 @@ async def remove_bausteine(topic: str):
 
 @router.post("/guides", response_model=GuideResponse)
 async def create(req: GuideCreateRequest):
-    if req.format != "OnePager" and not bausteine_path(req.topic.strip()).exists():
+    guides, progress = await lade_lernstand()
-        raise HTTPException(400, "Erst Bausteine erstellen")
+    grund = guide_lock(req.topic.strip(), req.format, guides, progress)
-    # Kein Duplikat-Start: pro Thema+Format höchstens eine laufende Generierung
+    if grund:
-    for g in await list_guides():
+        raise HTTPException(400 if grund == "Erst Bausteine erstellen" else 409, grund)
-        if g["topic"] == req.topic.strip() and g["format"] == req.format and g["status"] in ("queued", "generating"):
-            raise HTTPException(409, "Generierung läuft bereits")
-    # Lernschulden-Regeln — nur für Neu-Erstellungen; Resume (Schritt-Dateien
-    # vorhanden) und Neu-Generieren bestehender Guides sind ausgenommen.
-    content = guide_content_path(req.topic.strip(), req.format)
-    if req.format != "OnePager" and not content.exists() and not guide_slot_dateien(content):
-        vorstufe = VORSTUFE.get(req.format)
-        if vorstufe and not await _ist_absolviert(req.topic.strip(), vorstufe):
-            raise HTTPException(409, f"Erst den {vorstufe} dieses Themas absolvieren")
-        stat = (await _formate_stats()).get(req.format, {"erstellt": 0, "absolviert": 0})
-        offen = stat["erstellt"] - stat["absolviert"]
-        if offen >= MAX_OFFENE_GUIDES:
-            raise HTTPException(409, f"Erst {req.format}s absolvieren — maximal {MAX_OFFENE_GUIDES} offene erlaubt ({offen} offen)")
     await create_topic(req.topic.strip())
     now = datetime.now(timezone.utc).isoformat()
     guide = {
@@ -233,6 +168,13 @@ async def list_all():
     return await list_guides()
 
 
+@router.get("/guides/locks")
+async def guide_locks(topic: str):
+    """Sperr-Gründe pro Format für den ▶-Button — None = erstellbar."""
+    guides, progress = await lade_lernstand()
+    return {fmt: guide_lock(topic, fmt, guides, progress) for fmt in ("OnePager", *FORMATE)}
+
+
 @router.get("/guides/{guide_id}", response_model=GuideResponse)
 async def get_one(guide_id: str):
     guide = await get_guide(guide_id)

backend/textkit.py

@@ -4,14 +4,22 @@ Kein Zustand, keine IO — überall gefahrlos importierbar.
 """
 
 import re
+import unicodedata
 
 _CATEGORIES = ("KERN", "WICHTIG", "REST")  # nur noch für den Altformat-Reader
 
 
 def _norm_titel(s: str) -> str:
-    """Normalisiert einen Titel für den Schlüssel-Vergleich."""
+    """Normalisiert einen Titel für den Schlüssel-Vergleich.
-    s = re.sub(r"[`'\"<>]", "", s)
+
-    return re.sub(r"\s+", " ", s).strip().lower()
+    NFKC + casefold fangen Unicode-Varianten; Anführungszeichen, Markdown-
+    Emphasis und Dash-Varianten kommen aus KI-Output in allen Spielarten.
+    """
+    s = unicodedata.normalize("NFKC", s)
+    s = re.sub(r"[`'\"<>„“”‚’«»*_]", "", s)
+    s = re.sub(r"[–—‐]", "-", s)
+    s = re.sub(r"\s+", " ", s).strip().strip(".:;").strip()
+    return s.casefold()
 
 
 def _titel(entry: str) -> str: