""" Katalog-Prompt-Slots — Slot-Typ-Vokabular + Werte pro Stammdaten-Zeile (H2). Prompts in ai_prompts referenzieren Platzhalter wie {{focus_area_hints_on_progression}}. Inhalte liegen in catalog_prompt_slots (Admin-editierbar), nicht im Code pro Eintrag. """ from __future__ import annotations import json from dataclasses import dataclass from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple from planning_catalog_context import ( ProgressionPlanningCatalogContext, PlanningCatalogContextItem, catalog_context_has_items, ) from catalog_slot_fallbacks import merge_stored_slots_with_fallbacks # --------------------------------------------------------------------------- # Dimensionen (Prioritätsreihenfolge) # --------------------------------------------------------------------------- @dataclass(frozen=True) class CatalogKindConfig: kind: str table: str context_attr: str label_de: str CATALOG_KINDS: Tuple[CatalogKindConfig, ...] = ( CatalogKindConfig("focus_area", "focus_areas", "focus_areas", "Primärfokus"), CatalogKindConfig("training_type", "training_types", "training_types", "Trainingsstil"), CatalogKindConfig("target_group", "target_groups", "target_groups", "Zielgruppe"), CatalogKindConfig("style_direction", "style_directions", "style_directions", "Stilrichtung"), ) _KIND_BY_NAME = {c.kind: c for c in CATALOG_KINDS} # --------------------------------------------------------------------------- # Slot-Typen (Vokabular — erweiterbar via catalog_prompt_slot_types) # --------------------------------------------------------------------------- SLOT_KEYS: Tuple[str, ...] = ( "description", "hints_on_progression", "hints_on_exercise", "hints_on_path_qa", "anti_patterns", "rematch_guard", ) LLM_SLOT_KEYS: Tuple[str, ...] = tuple(k for k in SLOT_KEYS if k != "rematch_guard") GUIDANCE_BLOCK_SLOTS: Tuple[str, ...] = ( "description", "hints_on_progression", "hints_on_path_qa", "anti_patterns", ) GUIDANCE_PROFILE_BY_SLUG: Dict[str, Tuple[str, ...]] = { "planning_exercise_path_qa": ("description", "hints_on_path_qa", "anti_patterns"), "planning_progression_roadmap": ("description", "hints_on_progression", "anti_patterns"), "planning_progression_stage_spec": ("hints_on_progression", "anti_patterns", "description"), "planning_progression_goal_analysis": ("description", "hints_on_progression"), "planning_progression_start_target": ("description",), } def placeholder_key(catalog_kind: str, slot_key: str) -> str: return f"{catalog_kind}_{slot_key}" def all_placeholder_keys() -> List[str]: keys: List[str] = [] for cfg in CATALOG_KINDS: for slot in SLOT_KEYS: keys.append(placeholder_key(cfg.kind, slot)) keys.extend(["catalog_guidance_block", "catalog_context_json", "has_catalog_guidance"]) return keys def empty_catalog_variables() -> Dict[str, str]: out = {k: "" for k in all_placeholder_keys()} return out # --------------------------------------------------------------------------- # Katalog-Kontext → aktiver Eintrag # --------------------------------------------------------------------------- def pick_active_catalog_item( items: Sequence[PlanningCatalogContextItem], ) -> Optional[PlanningCatalogContextItem]: if not items: return None primaries = [i for i in items if i.is_primary] if primaries: return primaries[0] if len(items) == 1: return items[0] return max(items, key=lambda i: (float(i.weight), -int(i.id))) def _load_catalog_row(cur, table: str, item_id: int) -> Optional[Dict[str, Any]]: cur.execute( f""" SELECT id, name, description FROM {table} WHERE id = %s """, (int(item_id),), ) row = cur.fetchone() if not row: return None return { "id": int(row["id"]), "name": str(row.get("name") or "").strip(), "description": str(row.get("description") or "").strip(), } def _load_slots_for_entry(cur, catalog_kind: str, catalog_id: int) -> Dict[str, str]: cur.execute( """ SELECT slot_key, content FROM catalog_prompt_slots WHERE catalog_kind = %s AND catalog_id = %s """, (catalog_kind, int(catalog_id)), ) out: Dict[str, str] = {} for row in cur.fetchall(): key = str(row.get("slot_key") or "").strip() if key: out[key] = str(row.get("content") or "").strip() return out def _slot_types_table_ready(cur) -> bool: cur.execute("SELECT to_regclass(%s)::text AS t", ("public.catalog_prompt_slot_types",)) row = cur.fetchone() if not row: return False val = row.get("t") if isinstance(row, dict) else row[0] return val is not None and str(val).strip() != "" def list_slot_type_definitions(cur) -> List[Dict[str, Any]]: if not _slot_types_table_ready(cur): return _fallback_slot_type_rows() cur.execute( """ SELECT slot_key, display_name, description, applicable_kinds, sort_order, for_llm, for_code FROM catalog_prompt_slot_types ORDER BY sort_order ASC NULLS LAST, slot_key ASC """ ) rows = [] for row in cur.fetchall(): d = dict(row) kinds = d.get("applicable_kinds") if isinstance(kinds, str): kinds = [k.strip() for k in kinds.strip("{}").split(",") if k.strip()] d["applicable_kinds"] = list(kinds or []) rows.append(d) return rows def _fallback_slot_type_rows() -> List[Dict[str, Any]]: labels = { "description": "Allgemeine Beschreibung", "hints_on_progression": "Hinweise Progressionsgraph", "hints_on_exercise": "Hinweise Übungsanlage", "hints_on_path_qa": "Hinweise Pfad-QS", "anti_patterns": "Anti-Patterns (Fehlbewertung vermeiden)", "rematch_guard": "Rematch-Guard (Code)", } kinds = [c.kind for c in CATALOG_KINDS] rows = [] for i, key in enumerate(SLOT_KEYS): rows.append( { "slot_key": key, "display_name": labels.get(key, key), "description": "", "applicable_kinds": kinds, "sort_order": (i + 1) * 10, "for_llm": key != "rematch_guard", "for_code": key == "rematch_guard", } ) return rows def _resolve_entry_slot_values( stored: Mapping[str, str], row: Mapping[str, Any], catalog_kind: str, ) -> Dict[str, str]: """DB → Namens-Fallback → Stammdaten-Beschreibung (nur description).""" return merge_stored_slots_with_fallbacks( stored, catalog_kind=catalog_kind, name=str(row.get("name") or ""), stammdaten_description=str(row.get("description") or ""), ) def get_catalog_entry_slots(cur, catalog_kind: str, catalog_id: int) -> Dict[str, Any]: cfg = _KIND_BY_NAME.get((catalog_kind or "").strip()) if not cfg: raise ValueError(f"Unbekannter catalog_kind: {catalog_kind!r}") row = _load_catalog_row(cur, cfg.table, catalog_id) if not row: raise LookupError("Katalog-Eintrag nicht gefunden") stored = _load_slots_for_entry(cur, cfg.kind, catalog_id) merged = _resolve_entry_slot_values(stored, row, cfg.kind) return { "catalog_kind": cfg.kind, "catalog_id": int(catalog_id), "name": row["name"], "slots": merged, "stored_slots": {k: stored.get(k, "") for k in SLOT_KEYS}, } def upsert_catalog_entry_slots( cur, catalog_kind: str, catalog_id: int, slots: Mapping[str, Any], ) -> Dict[str, Any]: cfg = _KIND_BY_NAME.get((catalog_kind or "").strip()) if not cfg: raise ValueError(f"Unbekannter catalog_kind: {catalog_kind!r}") row = _load_catalog_row(cur, cfg.table, catalog_id) if not row: raise LookupError("Katalog-Eintrag nicht gefunden") for slot_key, raw in (slots or {}).items(): sk = str(slot_key or "").strip() if sk not in SLOT_KEYS: continue content = str(raw or "").strip() if not content: cur.execute( """ DELETE FROM catalog_prompt_slots WHERE catalog_kind = %s AND catalog_id = %s AND slot_key = %s """, (cfg.kind, int(catalog_id), sk), ) continue cur.execute( """ INSERT INTO catalog_prompt_slots (catalog_kind, catalog_id, slot_key, content, updated_at) VALUES (%s, %s, %s, %s, NOW()) ON CONFLICT (catalog_kind, catalog_id, slot_key) DO UPDATE SET content = EXCLUDED.content, updated_at = NOW() """, (cfg.kind, int(catalog_id), sk, content), ) return get_catalog_entry_slots(cur, cfg.kind, catalog_id) def _render_dimension_section( label_de: str, name: str, slot_values: Mapping[str, str], *, slot_keys: Sequence[str], ) -> Optional[str]: parts: List[str] = [f"### {label_de} — {name}"] labels = { "description": "Beschreibung", "hints_on_progression": "Progressions-Hinweise", "hints_on_path_qa": "QS-Hinweise", "hints_on_exercise": "Übungsanlage", "anti_patterns": "Vermeiden", } added = False for sk in slot_keys: text = str(slot_values.get(sk) or "").strip() if not text: continue added = True if sk == "description": parts.append(text) else: parts.append(f"{labels.get(sk, sk)}: {text}") if not added: return None return "\n".join(parts) def _compose_guidance_block( sections: List[str], ) -> str: if not sections: return "" return "## Katalog-Kontext (Didaktik & Bewertung)\n\n" + "\n\n".join(sections) def resolve_catalog_prompt_variables( cur, catalog: Optional[ProgressionPlanningCatalogContext], *, slug: Optional[str] = None, ) -> Dict[str, Any]: """ Liefert Mustache-Strings + Metadaten. Returns dict mit allen {{kind_slot}} Keys, catalog_guidance_block, catalog_context_json, has_catalog_guidance (bool), active_slots (list). """ variables = empty_catalog_variables() meta: Dict[str, Any] = { "active_slots": [], "audit": {}, } if cur is None or not catalog_context_has_items(catalog): variables["catalog_context_json"] = "" return {**variables, **meta} profile = GUIDANCE_PROFILE_BY_SLUG.get((slug or "").strip().lower(), GUIDANCE_BLOCK_SLOTS) sections: List[str] = [] audit: Dict[str, Any] = {} has_any = False active_slots: List[str] = [] for cfg in CATALOG_KINDS: items = getattr(catalog, cfg.context_attr, None) or [] active = pick_active_catalog_item(items) if not active: continue row = _load_catalog_row(cur, cfg.table, active.id) if not row: continue stored = _load_slots_for_entry(cur, cfg.kind, row["id"]) if _slot_types_table_ready(cur) else {} slot_values = _resolve_entry_slot_values(stored, row, cfg.kind) for sk in SLOT_KEYS: pk = placeholder_key(cfg.kind, sk) text = slot_values.get(sk, "") variables[pk] = text if text.strip() and sk in LLM_SLOT_KEYS: has_any = True active_slots.append(pk) audit[cfg.context_attr] = { "catalog_kind": cfg.kind, "id": row["id"], "name": row["name"], "is_primary": bool(active.is_primary), "weight": float(active.weight), "filled_slots": [k for k in LLM_SLOT_KEYS if slot_values.get(k, "").strip()], "stored_slots": [k for k in SLOT_KEYS if (stored.get(k) or "").strip()], } section = _render_dimension_section(cfg.label_de, row["name"], slot_values, slot_keys=profile) if section: sections.append(section) variables["catalog_guidance_block"] = _compose_guidance_block(sections) ctx_json = json.dumps(audit, ensure_ascii=False, separators=(",", ":")) variables["catalog_context_json"] = f"Katalog-Audit: {ctx_json}" if audit else "" variables["has_catalog_guidance"] = "true" if has_any else "" return { **variables, "active_slots": active_slots, "audit": audit, } def get_rematch_guard_for_catalog( cur, catalog: Optional[ProgressionPlanningCatalogContext], ) -> Optional[str]: """Erste passende rematch_guard entlang der Dimensions-Priorität.""" if cur is None or not catalog_context_has_items(catalog): return None for cfg in CATALOG_KINDS: items = getattr(catalog, cfg.context_attr, None) or [] active = pick_active_catalog_item(items) if not active: continue stored = _load_slots_for_entry(cur, cfg.kind, active.id) row = _load_catalog_row(cur, cfg.table, active.id) if not row: continue slot_values = _resolve_entry_slot_values(stored, row, cfg.kind) guard = (slot_values.get("rematch_guard") or "").strip() if guard: return guard return None # Abwärtskompatibilität H1-API def build_catalog_guidance_for_prompt( cur, catalog: Optional[ProgressionPlanningCatalogContext], *, slug: Optional[str] = None, ) -> Dict[str, Any]: resolved = resolve_catalog_prompt_variables(cur, catalog, slug=slug) return { "catalog_guidance_block": resolved.get("catalog_guidance_block", ""), "catalog_context_json": resolved.get("catalog_context_json", ""), "has_catalog_guidance": resolved.get("has_catalog_guidance") == "true", "snippet_keys": list(resolved.get("active_slots") or []), "variables": {k: str(resolved.get(k) or "") for k in all_placeholder_keys()}, } __all__ = [ "CATALOG_KINDS", "GUIDANCE_PROFILE_BY_SLUG", "SLOT_KEYS", "build_catalog_guidance_for_prompt", "empty_catalog_variables", "get_catalog_entry_slots", "get_rematch_guard_for_catalog", "list_slot_type_definitions", "pick_active_catalog_item", "placeholder_key", "all_placeholder_keys", "resolve_catalog_prompt_variables", "upsert_catalog_entry_slots", ]