automatisierte vorgenerirung von Initalien Sokrates Nachrichten

math-tutor/backend/app/scripts/__init__.py

+"""Backend maintenance scripts."""

math-tutor/backend/app/scripts/generate_socratic_chats.py

+from __future__ import annotations
+
+import argparse
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from app.LLM_services import socratic_LLM
+from app.deterministic_services import task_catalog
+from app.deterministic_services.vector_store import parse_markdown_with_frontmatter
+
+
+BACKEND_ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_SOURCE_ROOT = BACKEND_ROOT / "sources" / "with_chapters"
+DEFAULT_OUTPUT = BACKEND_ROOT / "sources" / "socratic_chats" / "initial_prompts.yml"
+DEFAULT_TEMPLATE = BACKEND_ROOT / "sources" / "socratic_chats" / "template.yml"
+DEFAULT_PROMPT = (
+    "Was sind die Themen dieses Abschnitts? Numeriere sie durch. Frage mich, mit welchem ich mich "
+    "zuerst beschäftigen möchte."
+)
+DEFAULT_TEMPLATE_FIELDS = ["subsection", "index", "inital_message", "sources"]
+
+
+SubsectionRef = tuple[int, int, int]
+
+
+@dataclass(frozen=True)
+class MarkdownDoc:
+    path: Path
+    relative_path: str
+    meta: dict[str, Any]
+    body: str
+    ref: SubsectionRef | None
+    title: str
+
+
+def _meta_int(meta: dict[str, Any], key: str) -> int | None:
+    value = meta.get(key)
+    if value is None or value == "":
+        return None
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _read_markdown_doc(
+    path: Path,
+    source_root: Path,
+    subsection_map: dict[str, SubsectionRef],
+) -> MarkdownDoc:
+    text = path.read_text(encoding="utf-8")
+    meta, body = parse_markdown_with_frontmatter(text)
+    title = str(
+        meta.get("title")
+        or meta.get("subsection_title")
+        or meta.get("section_title")
+        or path.stem
+    ).strip()
+    ref = _resolve_subsection_ref(meta, title, subsection_map)
+    return MarkdownDoc(
+        path=path,
+        relative_path=path.relative_to(source_root).as_posix(),
+        meta=meta,
+        body=body.strip(),
+        ref=ref,
+        title=title,
+    )
+
+
+def _resolve_subsection_ref(
+    meta: dict[str, Any],
+    title: str,
+    subsection_map: dict[str, SubsectionRef],
+) -> SubsectionRef | None:
+    chapter_index = _meta_int(meta, "chapter_index")
+    section_index = _meta_int(meta, "section_index")
+    subsection_index = _meta_int(meta, "subsection_index")
+    if chapter_index is not None and section_index is not None and subsection_index is not None:
+        return (chapter_index, section_index, subsection_index)
+
+    key_candidates = [
+        title,
+        str(meta.get("subsection_title") or ""),
+        str(meta.get("title") or ""),
+    ]
+    for candidate in key_candidates:
+        key = task_catalog._normalize_subsection_key(candidate)
+        if key and key in subsection_map:
+            return subsection_map[key]
+
+    if section_index is not None and subsection_index is not None:
+        matches = [
+            ref
+            for ref in subsection_map.values()
+            if int(ref[1]) == section_index and int(ref[2]) == subsection_index
+        ]
+        unique_matches = sorted(set(matches))
+        if len(unique_matches) == 1:
+            return unique_matches[0]
+
+    return None
+
+
+def _load_markdown_docs(
+    source_root: Path,
+    folder_name: str,
+    subsection_map: dict[str, SubsectionRef],
+) -> list[MarkdownDoc]:
+    folder = source_root / folder_name
+    if not folder.exists():
+        return []
+    return [
+        _read_markdown_doc(path, source_root, subsection_map)
+        for path in sorted(folder.glob("*.md"))
+    ]
+
+
+def load_subsections_and_children(
+    source_root: Path,
+    subsection_map: dict[str, SubsectionRef] | None = None,
+) -> tuple[list[MarkdownDoc], dict[SubsectionRef, list[MarkdownDoc]]]:
+    mapping = subsection_map if subsection_map is not None else task_catalog.load_subsection_map()
+    subsections = _load_markdown_docs(source_root, "subsections", mapping)
+    children = _load_markdown_docs(source_root, "childs", mapping)
+
+    children_by_ref: dict[SubsectionRef, list[MarkdownDoc]] = {}
+    for child in children:
+        if child.ref is None:
+            continue
+        children_by_ref.setdefault(child.ref, []).append(child)
+
+    for ref_children in children_by_ref.values():
+        ref_children.sort(key=lambda item: (_meta_int(item.meta, "child_index") or 0, item.relative_path))
+
+    return subsections, children_by_ref
+
+
+def _subsection_key(subsection: MarkdownDoc) -> str:
+    return task_catalog._normalize_subsection_key(subsection.title or subsection.path.stem)
+
+
+def _format_ref(ref: SubsectionRef) -> str:
+    return f"{ref[0]}:{ref[1]}:{ref[2]}"
+
+
+def build_llm_sources_text(subsection: MarkdownDoc, children: list[MarkdownDoc]) -> str:
+    ref_text = _format_ref(subsection.ref) if subsection.ref else ""
+    parts = [
+        f"Abschnitt: {subsection.title}",
+        f"Index: {ref_text}",
+        f"Datei: {subsection.relative_path}",
+        "Abschnittsinhalt:",
+        subsection.body,
+    ]
+
+    for child in children:
+        parts.extend(
+            [
+                "",
+                f"Child-Datei: {child.relative_path}",
+                f"Child-Titel: {child.title}",
+                "Child-Inhalt:",
+                child.body,
+            ]
+        )
+
+    return "\n".join(parts).strip()
+
+
+def generate_initial_message(subsection: MarkdownDoc, children: list[MarkdownDoc]) -> str:
+    sources_text = build_llm_sources_text(subsection, children)
+    refs = [subsection.ref] if subsection.ref else None
+    return socratic_LLM.generate_dialog(
+        query=DEFAULT_PROMPT,
+        subsection_refs=refs,
+        history=None,
+        sources=sources_text,
+    ).strip()
+
+
+def _load_existing_items(output_path: Path) -> dict[str, dict[str, Any]]:
+    if not output_path.exists():
+        return {}
+    try:
+        payload = yaml.safe_load(output_path.read_text(encoding="utf-8")) or {}
+    except Exception:
+        return {}
+    items = payload.get("items") if isinstance(payload, dict) else None
+    if not isinstance(items, dict):
+        return {}
+    return {
+        str(key): value
+        for key, value in items.items()
+        if isinstance(value, dict)
+    }
+
+
+def _load_template_fields(template_path: Path) -> list[str]:
+    if not template_path.exists():
+        return DEFAULT_TEMPLATE_FIELDS[:]
+    payload = yaml.safe_load(template_path.read_text(encoding="utf-8")) or {}
+    if not isinstance(payload, dict):
+        return DEFAULT_TEMPLATE_FIELDS[:]
+    fields = [str(key) for key in payload.keys()]
+    return fields or DEFAULT_TEMPLATE_FIELDS[:]
+
+
+def _ordered_item(item: dict[str, Any], template_fields: list[str]) -> dict[str, Any]:
+    ordered: dict[str, Any] = {}
+    for field in template_fields:
+        if field in item:
+            ordered[field] = item[field]
+    for field, value in item.items():
+        if field not in ordered:
+            ordered[field] = value
+    return ordered
+
+
+def _display_path(path: Path) -> str:
+    try:
+        return path.resolve().relative_to(BACKEND_ROOT).as_posix()
+    except ValueError:
+        return path.as_posix()
+
+
+def build_initial_prompt_index(
+    source_root: Path,
+    output_path: Path,
+    template_path: Path = DEFAULT_TEMPLATE,
+    *,
+    force: bool = False,
+    limit: int | None = None,
+    strict: bool = False,
+) -> dict[str, Any]:
+    source_root = source_root.resolve()
+    output_path = output_path.resolve()
+    template_fields = _load_template_fields(template_path)
+    existing_items = _load_existing_items(output_path)
+    subsections, children_by_ref = load_subsections_and_children(source_root)
+
+    items: dict[str, dict[str, Any]] = {}
+    warnings: list[str] = []
+    processed = 0
+
+    for subsection in subsections:
+        if limit is not None and processed >= limit:
+            break
+
+        key = _subsection_key(subsection)
+        if not key:
+            warnings.append(f"Skipping subsection without key: {subsection.relative_path}")
+            continue
+        if subsection.ref is None:
+            warnings.append(f"Skipping subsection without resolved index: {subsection.relative_path}")
+            continue
+
+        children = children_by_ref.get(subsection.ref, [])
+        if not children:
+            message = f"No child chunks matched subsection {subsection.relative_path}"
+            if strict:
+                raise ValueError(message)
+            warnings.append(message)
+
+        sources = [subsection.relative_path, *[child.relative_path for child in children]]
+        existing_message = str(existing_items.get(key, {}).get("inital_message") or "").strip()
+        if existing_message and not force:
+            initial_message = existing_message
+        else:
+            initial_message = generate_initial_message(subsection, children)
+
+        item = {
+            "subsection": subsection.title,
+            "index": _format_ref(subsection.ref),
+            "inital_message": initial_message,
+            "sources": sources,
+        }
+        items[key] = _ordered_item(item, template_fields)
+        processed += 1
+
+    if strict and warnings:
+        raise ValueError("\n".join(warnings))
+
+    for warning in warnings:
+        print(f"warning: {warning}", file=sys.stderr)
+
+    return {
+        "prompt": DEFAULT_PROMPT,
+        "source_root": _display_path(source_root),
+        "items": items,
+    }
+
+
+def write_initial_prompt_index(payload: dict[str, Any], output_path: Path) -> None:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(
+        yaml.safe_dump(payload, sort_keys=False, allow_unicode=True),
+        encoding="utf-8",
+    )
+
+
+def generate_file(
+    source_root: Path = DEFAULT_SOURCE_ROOT,
+    output_path: Path = DEFAULT_OUTPUT,
+    template_path: Path = DEFAULT_TEMPLATE,
+    *,
+    force: bool = False,
+    limit: int | None = None,
+    strict: bool = False,
+) -> dict[str, Any]:
+    payload = build_initial_prompt_index(
+        source_root=source_root,
+        output_path=output_path,
+        template_path=template_path,
+        force=force,
+        limit=limit,
+        strict=strict,
+    )
+    write_initial_prompt_index(payload, output_path)
+    return payload
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Generate pregenerated Socratic initial prompts for subsections."
+    )
+    parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT)
+    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
+    parser.add_argument("--template", type=Path, default=DEFAULT_TEMPLATE)
+    parser.add_argument("--force", action="store_true")
+    parser.add_argument("--limit", type=int, default=None)
+    parser.add_argument("--strict", action="store_true")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> None:
+    args = parse_args(argv)
+    payload = generate_file(
+        source_root=args.source_root,
+        output_path=args.output,
+        template_path=args.template,
+        force=args.force,
+        limit=args.limit,
+        strict=args.strict,
+    )
+    print(f"Wrote {len(payload.get('items', {}))} Socratic prompts to {args.output}")
+
+
+if __name__ == "__main__":
+    main()