Lars
53f2b027cc
Some checks failed
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Failing after 2s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 13s
Test Suite / k6 /health Baseline (push) Successful in 34s
Test Suite / playwright-tests (push) Successful in 1m20s
- Introduced new catalog context handling in planning prompt functions, allowing for improved integration of planning variables. - Added optional catalog context parameters in various functions to streamline the merging of planning prompt variables. - Updated frontend components to include CatalogPromptSlotsEditor for managing prompt slots across different catalog types. - Enhanced API utilities to support fetching and updating catalog prompt slots, improving backend functionality for catalog management. - Incremented version numbers and updated changelog to reflect the new features and improvements.
420 lines
14 KiB
Python
420 lines
14 KiB
Python
"""
|
|
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,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 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: Dict[str, str] = {}
|
|
for slot in SLOT_KEYS:
|
|
if slot == "description":
|
|
merged[slot] = stored.get("description") or row.get("description") or ""
|
|
else:
|
|
merged[slot] = stored.get(slot, "")
|
|
return {
|
|
"catalog_kind": cfg.kind,
|
|
"catalog_id": int(catalog_id),
|
|
"name": row["name"],
|
|
"slots": merged,
|
|
}
|
|
|
|
|
|
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: Dict[str, str] = {}
|
|
for sk in SLOT_KEYS:
|
|
if sk == "description":
|
|
slot_values[sk] = stored.get("description") or row.get("description") or ""
|
|
else:
|
|
slot_values[sk] = stored.get(sk, "")
|
|
pk = placeholder_key(cfg.kind, sk)
|
|
variables[pk] = slot_values[sk]
|
|
if slot_values[sk].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()],
|
|
}
|
|
|
|
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)
|
|
guard = (stored.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",
|
|
]
|