shinkan-jinkendo/backend/planning_exercise_intent.py

"""
P1: LLM-Intent aus Planungs-Suchfrage → strukturiertes Query-Overlay für PlanningTargetProfile.

Prompt: planning_exercise_search_intent (Migration 073)
"""
from __future__ import annotations

import json
import logging
import re
from typing import Any, Dict, List, Mapping, Optional, Sequence, Set, Tuple

from pydantic import BaseModel, Field, field_validator

from ai_prompt_runtime import AiPromptUnavailableError, load_and_render_ai_prompt
from openrouter_chat import (
    effective_openrouter_model_for_prompt_row,
    normalize_openrouter_env,
    openrouter_chat_completion,
)

_logger = logging.getLogger("shinkan.planning_exercise_intent")

VALID_PARSED_INTENTS = {
    "suggest_next",
    "progression_next",
    "deepen_exercise",
    "continue_plan_goal",
    "free_search",
}

VALID_SCENARIOS = {
    "preset_next",
    "progression",
    "deepen",
    "continue_plan",
    "additive_constraint",
    "free_search",
}

VALID_EMPHASIS = {"additive", "replace", "neutral"}


class SkillHint(BaseModel):
    name: str = Field(..., min_length=1, max_length=120)
    weight: float = Field(default=1.0, ge=0.1, le=1.0)


class PlanningQueryIntentParsed(BaseModel):
    intent: str = "free_search"
    scenario: str = "free_search"
    skill_hints: List[SkillHint] = Field(default_factory=list)
    focus_hints: List[str] = Field(default_factory=list)
    style_hints: List[str] = Field(default_factory=list)
    training_type_hints: List[str] = Field(default_factory=list)
    target_group_hints: List[str] = Field(default_factory=list)
    requires_partner: Optional[bool] = None
    emphasis: str = "additive"
    rationale: Optional[str] = Field(default=None, max_length=400)

    @field_validator("intent")
    @classmethod
    def _intent(cls, v: str) -> str:
        s = (v or "").strip().lower()
        return s if s in VALID_PARSED_INTENTS else "free_search"

    @field_validator("scenario")
    @classmethod
    def _scenario(cls, v: str) -> str:
        s = (v or "").strip().lower()
        return s if s in VALID_SCENARIOS else "free_search"

    @field_validator("emphasis")
    @classmethod
    def _emphasis(cls, v: str) -> str:
        s = (v or "").strip().lower()
        return s if s in VALID_EMPHASIS else "additive"

    @field_validator("focus_hints", "style_hints", "training_type_hints", "target_group_hints", mode="before")
    @classmethod
    def _str_list(cls, v: Any) -> List[str]:
        if not v:
            return []
        if isinstance(v, str):
            return [v.strip()] if v.strip() else []
        out: List[str] = []
        for item in v:
            s = str(item or "").strip()
            if s and s not in out:
                out.append(s[:120])
        return out[:8]


def _extract_json_object(text: str) -> Dict[str, Any]:
    s = (text or "").strip()
    if s.startswith("```"):
        s = re.sub(r"^```[a-zA-Z0-9]*\s*", "", s)
        if s.endswith("```"):
            s = s[:-3].strip()
    start = s.find("{")
    end = s.rfind("}")
    if start < 0 or end <= start:
        raise ValueError("Kein JSON-Objekt in LLM-Antwort")
    obj = json.loads(s[start : end + 1])
    if not isinstance(obj, dict):
        raise ValueError("LLM-Antwort ist kein JSON-Objekt")
    return obj


def parse_planning_query_intent_response(text: str) -> PlanningQueryIntentParsed:
    obj = _extract_json_object(text)
    return PlanningQueryIntentParsed.model_validate(obj)


def _compact_json(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, separators=(",", ":"))


def _load_compact_catalog(cur, table: str, id_col: str, name_col: str = "name", limit: int = 80) -> List[Dict[str, Any]]:
    cur.execute(
        f"""
        SELECT {id_col} AS id, {name_col} AS name
        FROM {table}
        ORDER BY {name_col} ASC NULLS LAST
        LIMIT %s
        """,
        (limit,),
    )
    return [{"id": int(r["id"]), "name": str(r["name"] or "")[:80]} for r in cur.fetchall()]


def _load_skills_catalog_compact(cur, limit: int = 120) -> List[Dict[str, Any]]:
    cur.execute(
        """
        SELECT id, name, category
        FROM skills
        WHERE status IS NULL OR status = 'active'
        ORDER BY name ASC
        LIMIT %s
        """,
        (limit,),
    )
    return [
        {
            "id": int(r["id"]),
            "name": str(r["name"] or "")[:80],
            "category": str(r.get("category") or "")[:40],
        }
        for r in cur.fetchall()
    ]


def _resolve_name_hint(cur, table: str, hint: str, *, extra_where: str = "") -> Optional[int]:
    h = (hint or "").strip()
    if len(h) < 2:
        return None
    q = h.lower()
    cur.execute(
        f"""
        SELECT id, name
        FROM {table}
        WHERE LOWER(name) LIKE %s {extra_where}
        ORDER BY CASE WHEN LOWER(name) = %s THEN 0 WHEN LOWER(name) LIKE %s THEN 1 ELSE 2 END,
                 LENGTH(name) ASC
        LIMIT 1
        """,
        (f"%{q}%", q, f"{q}%"),
    )
    row = cur.fetchone()
    return int(row["id"]) if row else None


def resolve_query_intent_catalog_ids(
    cur,
    parsed: PlanningQueryIntentParsed,
) -> Tuple[Dict[int, float], Dict[int, float], Dict[int, float], Dict[int, float], Dict[int, float], List[Dict[str, Any]]]:
    """
    Mappt Text-Hints auf Katalog-IDs. Returns (focus, style, tt, tg, skills, resolved_skills_meta).
    """
    focus: Dict[int, float] = {}
    style: Dict[int, float] = {}
    tt: Dict[int, float] = {}
    tg: Dict[int, float] = {}
    skills: Dict[int, float] = {}
    resolved_skills: List[Dict[str, Any]] = []

    for hint in parsed.focus_hints:
        fid = _resolve_name_hint(cur, "focus_areas", hint)
        if fid:
            focus[fid] = max(focus.get(fid, 0.0), 0.9)

    for hint in parsed.style_hints:
        sid = _resolve_name_hint(cur, "style_directions", hint)
        if sid:
            style[sid] = max(style.get(sid, 0.0), 0.85)

    for hint in parsed.training_type_hints:
        tid = _resolve_name_hint(cur, "training_types", hint)
        if tid:
            tt[tid] = max(tt.get(tid, 0.0), 0.85)

    for hint in parsed.target_group_hints:
        gid = _resolve_name_hint(cur, "target_groups", hint)
        if gid:
            tg[gid] = max(tg.get(gid, 0.0), 0.85)

    for sh in parsed.skill_hints[:8]:
        cur.execute(
            """
            SELECT id, name FROM skills
            WHERE (status IS NULL OR status = 'active')
              AND LOWER(name) LIKE %s
            ORDER BY CASE WHEN LOWER(name) = %s THEN 0 WHEN LOWER(name) LIKE %s THEN 1 ELSE 2 END,
                     LENGTH(name) ASC
            LIMIT 1
            """,
            (f"%{sh.name.lower()}%", sh.name.lower(), f"{sh.name.lower()}%"),
        )
        row = cur.fetchone()
        if row:
            sid = int(row["id"])
            skills[sid] = max(skills.get(sid, 0.0), float(sh.weight))
            resolved_skills.append({"skill_id": sid, "name": str(row["name"] or sh.name), "weight": skills[sid]})

    return focus, style, tt, tg, skills, resolved_skills


def try_parse_planning_query_intent(
    cur,
    *,
    query: str,
    heuristic_intent: str,
    scenario_hint: str,
    context_summary: Mapping[str, Any],
    target_profile_summary: Mapping[str, Any],
) -> Tuple[Optional[PlanningQueryIntentParsed], bool]:
    api_key, _ = normalize_openrouter_env()
    if not api_key or not (query or "").strip():
        return None, False

    variables = {
        "search_query": (query or "").strip(),
        "heuristic_intent": heuristic_intent or "",
        "scenario_hint": scenario_hint or "",
        "planning_context_json": _compact_json(dict(context_summary or {})),
        "target_profile_json": _compact_json(dict(target_profile_summary or {})),
        "skills_catalog_json": _compact_json(_load_skills_catalog_compact(cur)),
        "focus_areas_catalog_json": _compact_json(_load_compact_catalog(cur, "focus_areas", "id")),
        "training_types_catalog_json": _compact_json(_load_compact_catalog(cur, "training_types", "id")),
        "style_directions_catalog_json": _compact_json(_load_compact_catalog(cur, "style_directions", "id")),
        "target_groups_catalog_json": _compact_json(_load_compact_catalog(cur, "target_groups", "id")),
    }

    try:
        prow, rendered = load_and_render_ai_prompt(cur, "planning_exercise_search_intent", variables)
        model = effective_openrouter_model_for_prompt_row(prow)
        raw = openrouter_chat_completion(api_key=api_key, model=model, user_content=rendered.text)
        parsed = parse_planning_query_intent_response(raw)
        return parsed, True
    except AiPromptUnavailableError:
        return None, False
    except Exception as exc:
        _logger.warning("Planungs-Intent-LLM fehlgeschlagen: %s", exc)
        return None, False


__all__ = [
    "PlanningQueryIntentParsed",
    "parse_planning_query_intent_response",
    "resolve_query_intent_catalog_ids",
    "try_parse_planning_query_intent",
]