157 lines
6.4 KiB
Python
157 lines
6.4 KiB
Python
"""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 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]]],
|
||
has_blocks: bool) -> 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).
|
||
`has_blocks` = DB consensus blocks OR the legacy blocks.md exist — checked by the
|
||
caller, because generate_guide builds from the DB (a paused inventory writes no file).
|
||
"""
|
||
if not has_blocks:
|
||
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
|