shinkan-jinkendo/backend/planning_exercise_form_context.py

"""
Planungs-KI Phase D: strukturierter Planungskontext für POST /exercises/ai/suggest.

Wird als ``planning_context_json`` in Übungs-Prompts (summary, skills, instructions) injiziert.
"""
from __future__ import annotations

import json
from typing import Any, Dict, Mapping, Optional

_MAX_JSON_CHARS = 6000
_MAX_STRING = 800


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


def _trim_str(val: Any, *, limit: int = _MAX_STRING) -> Optional[str]:
    if val is None:
        return None
    s = str(val).strip()
    if not s:
        return None
    if len(s) > limit:
        return s[: limit - 1] + "…"
    return s


def sanitize_planning_context_for_ai(ctx: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
    """Reduziert Client-Payload auf prompt-taugliche, begrenzte Felder."""
    if not ctx:
        return {}
    out: Dict[str, Any] = {}
    for key, val in dict(ctx).items():
        if val is None:
            continue
        k = str(key).strip()
        if not k:
            continue
        if isinstance(val, str):
            t = _trim_str(val)
            if t:
                out[k] = t
        elif isinstance(val, (int, float, bool)):
            out[k] = val
        elif isinstance(val, list):
            items = []
            for item in val[:12]:
                if isinstance(item, str):
                    t = _trim_str(item, limit=200)
                    if t:
                        items.append(t)
                elif isinstance(item, (int, float, bool)):
                    items.append(item)
                elif isinstance(item, dict):
                    sub = sanitize_planning_context_for_ai(item)
                    if sub:
                        items.append(sub)
            if items:
                out[k] = items
        elif isinstance(val, dict):
            sub = sanitize_planning_context_for_ai(val)
            if sub:
                out[k] = sub
    raw = compact_planning_context_json(out)
    if len(raw) > _MAX_JSON_CHARS:
        out["truncated"] = True
        out.pop("path_steps_preview", None)
        raw = compact_planning_context_json(out)
    if len(raw) > _MAX_JSON_CHARS:
        return {"source": out.get("source"), "truncated": True, "goal_query": out.get("goal_query")}
    return out


def planning_context_prompt_variables(
    planning_context: Optional[Mapping[str, Any]],
) -> Dict[str, str]:
    cleaned = sanitize_planning_context_for_ai(planning_context)
    if not cleaned:
        return {"planning_context_json": "-", "has_planning_context": ""}
    return {
        "planning_context_json": compact_planning_context_json(cleaned),
        "has_planning_context": "true",
    }


def build_progression_path_gap_planning_context(
    *,
    goal_query: str,
    primary_topic: Optional[str] = None,
    progression_graph_id: Optional[int] = None,
    offer: Optional[Mapping[str, Any]] = None,
    neighbor_before: Optional[Mapping[str, Any]] = None,
    neighbor_after: Optional[Mapping[str, Any]] = None,
    path_step_count: int = 0,
    major_step_count: Optional[int] = None,
    roadmap_phase: Optional[str] = None,
    roadmap_learning_goal: Optional[str] = None,
) -> Dict[str, Any]:
    """Kontext für KI-Neuanlage aus Progressionsgraph-Pfad-Lücke."""
    offer = offer or {}
    gap = offer.get("gap") if isinstance(offer.get("gap"), dict) else {}
    major_idx = offer.get("roadmap_major_step_index")
    if major_idx is None and isinstance(gap, dict):
        major_idx = gap.get("roadmap_major_step_index")

    ctx: Dict[str, Any] = {
        "source": "progression_path_gap_fill",
        "goal_query": _trim_str(goal_query, limit=2000),
        "primary_topic": _trim_str(primary_topic),
        "progression_graph_id": progression_graph_id,
        "gap_source": _trim_str(offer.get("source")),
        "gap_phase": _trim_str(offer.get("phase") or gap.get("expected_phase")),
        "roadmap_major_step_index": major_idx,
        "roadmap_phase": _trim_str(roadmap_phase or offer.get("phase")),
        "roadmap_learning_goal": _trim_str(
            roadmap_learning_goal or offer.get("title_hint") or gap.get("learning_goal"),
            limit=1200,
        ),
        "neighbor_before_title": _trim_str(
            (neighbor_before or {}).get("title") or offer.get("from_title")
        ),
        "neighbor_after_title": _trim_str(
            (neighbor_after or {}).get("title") or offer.get("to_title")
        ),
        "path_step_count": path_step_count,
        "major_step_count": major_step_count,
    }
    return sanitize_planning_context_for_ai(ctx)


__all__ = [
    "build_progression_path_gap_planning_context",
    "compact_planning_context_json",
    "planning_context_prompt_variables",
    "sanitize_planning_context_for_ai",
]