shinkan-jinkendo/backend/ai_prompt_runtime.py

"""
Gemeinsame KI-Prompt-Laufzeit (Shinkan): DB-Lesezugriff ai_prompts + Kontext-Arten.

Bleibt ohne Import von exercise_ai (kein Zirkel). Domänen wie exercise_ai nutzen
load_ai_prompt_row und die Enum; Platzhalter bauen sie selbst oder über geteilte Builder.
"""
from __future__ import annotations

from enum import Enum
from typing import Any, Dict, Mapping, Optional, Tuple

from prompt_resolver import MustacheRenderResult, render_mustache_template

_PLANNING_AI_SLUGS = frozenset(
    {
        "planning_exercise_search_rank",
        "planning_exercise_search_intent",
    }
)

_EXERCISE_AI_SLUGS = frozenset(
    {
        "exercise_summary",
        "exercise_skill_suggestions",
        "exercise_instruction_rewrite",
    }
)


class AiPromptContextKind(str, Enum):
    """
    Logischer Kontext fuer Platzhalter/Builder — erweiterbar fuer Planung/Rahmen
    ohne bestehende Slugs zu invalidieren.
    """

    PLANNING_EXERCISE_SEARCH = "planning_exercise_search"
    EXERCISE_FORM_AI = "exercise_form_ai"


def context_kind_for_slug(slug: str) -> Optional[AiPromptContextKind]:
    """Ordnet einen DB-Slug einer Kontext-Art zu, sofern registriert."""
    s = (slug or "").strip().lower()
    if s in _PLANNING_AI_SLUGS:
        return AiPromptContextKind.PLANNING_EXERCISE_SEARCH
    if s in _EXERCISE_AI_SLUGS:
        return AiPromptContextKind.EXERCISE_FORM_AI
    return None


def load_ai_prompt_row(cur, slug: str, *, active_only: bool = True) -> Optional[Dict[str, Any]]:
    """
    Laedt eine Zeile ai_prompts fuer Laufzeit-Orchestrierung.

    active_only=True: inaktive Prompts werden wie fehlend behandelt (503 im Aufrufer).
    """
    if active_only:
        cur.execute(
            """
            SELECT slug, display_name, template, output_format, active, openrouter_model
            FROM ai_prompts
            WHERE slug = %s AND active = true
            """,
            (slug,),
        )
    else:
        cur.execute(
            """
            SELECT slug, display_name, template, output_format, active, openrouter_model
            FROM ai_prompts
            WHERE slug = %s
            """,
            (slug,),
        )
    row = cur.fetchone()
    if not row:
        return None
    d = dict(row)
    if active_only and not d.get("active", True):
        return None
    return d


class AiPromptUnavailableError(LookupError):
    """Kein aktiver Prompt fuer slug (oder Zeile fehlt)."""

    def __init__(self, slug: str) -> None:
        self.slug = (slug or "").strip()
        super().__init__(self.slug)


def render_ai_prompt_template_for_row(
    row: Mapping[str, Any],
    variables: Mapping[str, str],
) -> MustacheRenderResult:
    """Ersetzt Platzhalter anhand einer bereits geladenen ai_prompts-Zeile (z. B. Admin-Vorschauch, inkl. inaktiv)."""
    return render_mustache_template(str(row.get("template") or ""), variables)


def load_and_render_ai_prompt(
    cur,
    slug: str,
    variables: Mapping[str, str],
    *,
    active_only: bool = True,
) -> Tuple[Dict[str, Any], MustacheRenderResult]:
    """
    Laedt einen aktiven Prompt und wendet Mustache-Variablen an.
    Wirft AiPromptUnavailableError, wenn die Zeile fehlt oder (bei active_only) inaktiv ist.
    """
    row = load_ai_prompt_row(cur, slug, active_only=active_only)
    if not row:
        raise AiPromptUnavailableError(slug)
    rr = render_ai_prompt_template_for_row(row, variables)
    return dict(row), rr


__all__ = [
    "AiPromptContextKind",
    "AiPromptUnavailableError",
    "context_kind_for_slug",
    "load_ai_prompt_row",
    "load_and_render_ai_prompt",
    "render_ai_prompt_template_for_row",
]