Lars
7e5ef4561a
Some checks failed
Deploy Development / deploy (push) Successful in 40s
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 1m22s
- Introduced a new function `_resolve_entry_slot_values` to streamline the merging of stored slot values with fallbacks, improving code clarity and maintainability. - Updated `get_catalog_entry_slots` and `resolve_catalog_prompt_variables` functions to utilize the new fallback logic, enhancing the handling of catalog entries. - Enhanced the `CatalogPromptSlotsEditor` component to display fallback information for slots, improving user experience in managing catalog prompts. - Incremented version numbers and updated changelog to reflect the new features and improvements.
433 lines
14 KiB
Python
433 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,
|
|
)
|
|
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",
|
|
]
|