"""Learning-debt rules: progression and cap for open guides — the ONLY source. Rules (new creations only; topics + blocks unlimited): - Format "Guide": at most 3 created, not-yet-completed guides - No more progression/prerequisite (only a single guide format). - Completed: ALL blocks (section titles) of the latest finished guide have a passed exam. The rest is read-only (no progress, no exam). All functions work on data loaded once (load_learnstate) — no more query loops per guide. """ import json from database import list_block_scores_all, subs_per_level_all, list_guides from guide import guide_slot_files from learning import cap_final, LEVELS, _threshold from paths import blocks_path, guide_content_path from textkit import _norm_title MAX_OFFENE_GUIDES = 3 # Only ONE format "Guide" left (all relevant blocks, exam 0–cap). No progression, # no prerequisite → the guide is always unlockable. "Rest"/FullGuide are separate. PRESTAGE: dict[str, str] = {} FREISCHALT_LEVEL: dict[str, str] = {} FORMATE = ("Guide",) # 4 learning levels (floor in % of the cap) — keys from learning.LEVELS. _LEVEL_WORT = { "beginner": "to beginner (20%)", "advanced": "to advanced (40%)", "expert": "to expert (60%)", "master": "to mastery (100%)", } async def load_learnstate() -> tuple[list[dict], dict[str, dict[str, set[str]]]]: """Guides + blocks per level. levels: {"beginner"/"advanced"/"expert"/"master": {topic → normalized title}}. The level per block is derived from score + cap (4×relevant subs). """ scores = await list_block_scores_all() subs_by_level = await subs_per_level_all() levels: dict[str, dict[str, set[str]]] = {key: {} for key, _ in LEVELS} for topic, block, score in scores: cf = cap_final(subs_by_level.get((topic, _norm_title(block)), {})) for key, p in LEVELS: if cf and score >= _threshold(p, cf): levels[key].setdefault(topic, set()).add(_norm_title(block)) return await list_guides(), levels _content_cache: dict[str, tuple[float, dict | None]] = {} def _content_json(topic: str, fmt: str) -> dict | None: """Guide-Content-JSON (kann MB groß sein), mtime-gecacht — /stats und /topics/progress lasen die Datei bei JEDEM Frontend-Poll neu und synchron im Event-Loop.""" path = guide_content_path(topic, fmt) try: mtime = path.stat().st_mtime except OSError: _content_cache.pop(str(path), None) return None cached = _content_cache.get(str(path)) if cached is None or cached[0] != mtime: try: data = json.loads(path.read_text(encoding="utf-8")) except ValueError: data = None _content_cache[str(path)] = (mtime, data) return _content_cache[str(path)][1] def _section_title(topic: str, fmt: str) -> set[str] | None: """Normalized block titles (sections) from the guide content.""" content = _content_json(topic, fmt) if content is None: return None return { _norm_title(s.get("title", "")) for ch in content.get("chapters", []) for s in ch.get("sections", []) } def _latest_done(guides: list[dict], fmt: str) -> dict[str, dict]: """Per topic, the latest finished guide of this format.""" latest: dict[str, dict] = {} for g in guides: if g["format"] == fmt and g["status"] == "done": if g["topic"] not in latest or g["created_at"] > latest[g["topic"]]["created_at"]: latest[g["topic"]] = g return latest def _guide_all(g: dict, levelset: dict[str, set[str]]) -> bool: """Are ALL blocks of the guide at the required level?""" sections = _section_title(g["topic"], g["format"]) return bool(sections) and sections <= levelset.get(g["topic"], set()) def is_level(topic: str, fmt: str, guides: list[dict], levelset: dict[str, set[str]]) -> bool: """Latest finished guide (topic+format): all blocks at the level of levelset?""" g = _latest_done(guides, fmt).get(topic) return g is not None and _guide_all(g, levelset) def ist_completed(topic: str, fmt: str, guides: list[dict], levels: dict[str, dict[str, set[str]]]) -> bool: """All blocks of the latest finished guide at least beginner (≥20%)?""" return is_level(topic, fmt, guides, levels["beginner"]) def topic_completed(topic: str, guides: list[dict], levels: dict[str, dict[str, set[str]]]) -> bool: """Topic done: latest finished guide, all blocks at master (100%)?""" return is_level(topic, "Guide", guides, levels["master"]) def formats_stats(guides: list[dict], levels: dict[str, dict[str, set[str]]]) -> dict: """Per format created/completed — per topic only the latest finished guide counts.""" formats = {} for fmt in FORMATE: latest = _latest_done(guides, fmt) completed = sum(1 for g in latest.values() if _guide_all(g, levels["beginner"])) formats[fmt] = {"created": len(latest), "completed": completed} return formats def guide_lock(topic: str, fmt: str, guides: list[dict], levels: dict[str, dict[str, set[str]]]) -> str | None: """Reason why a fresh start for topic+format is locked — None = allowed. Exactly the rules from POST /guides: blocks required, no duplicate start, learning debt only for genuine new creations (resume/regenerate are free). """ if not blocks_path(topic).exists(): return "Create blocks first" for g in guides: if g["topic"] == topic and g["format"] == fmt and g["status"] in ("queued", "generating"): return "Generation already running" content = guide_content_path(topic, fmt) if not content.exists() and not guide_slot_files(content): prereq = PRESTAGE.get(fmt) if prereq: level = FREISCHALT_LEVEL[fmt] # completed=10 · understood=20 · mastered=30 if not is_level(topic, prereq, guides, levels[level]): return f"First take the {prereq} of this topic {_LEVEL_WORT[level]}" stat = formats_stats(guides, levels).get(fmt, {"created": 0, "completed": 0}) open_count = stat["created"] - stat["completed"] if open_count >= MAX_OFFENE_GUIDES: return f"Complete {fmt}s first — at most {MAX_OFFENE_GUIDES} open allowed ({open_count} open)" return None