""" 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, List, 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_gap_snapshot( *, goal_analysis: Optional[Mapping[str, Any]] = None, resolved_structured: Optional[Mapping[str, Any]] = None, stage_spec: Optional[Mapping[str, Any]] = None, semantic_brief: Optional[Mapping[str, Any]] = None, ) -> Dict[str, Any]: """Kompakter Roadmap-Kontext für Lücken-Übungen (Start, Ziel, Stufe, Fähigkeiten-Hinweise).""" ga = dict(goal_analysis or {}) rs = dict(resolved_structured or {}) spec = dict(stage_spec or {}) brief = dict(semantic_brief or {}) start = _trim_str(rs.get("start_situation") or ga.get("start_assumption")) target = _trim_str(rs.get("target_state") or ga.get("target_state")) notes = _trim_str(rs.get("roadmap_notes")) topic = _trim_str(ga.get("primary_topic") or brief.get("primary_topic")) skill_hints: List[str] = [] for item in (brief.get("must_phrases") or [])[:4]: t = _trim_str(item, limit=120) if t: skill_hints.append(t) arc = brief.get("development_arc") if isinstance(arc, list) and arc: skill_hints.append(f"Entwicklungsbogen: {' → '.join(str(x) for x in arc[:5])}") success_path = [ _trim_str(x, limit=200) for x in (ga.get("success_criteria") or []) if _trim_str(x, limit=200) ][:4] stage_success = [ _trim_str(x, limit=200) for x in (spec.get("success_criteria") or []) if _trim_str(x, limit=200) ][:4] load_profile = [ _trim_str(x, limit=80) for x in (spec.get("load_profile") or []) if _trim_str(x, limit=80) ][:6] anti_patterns = [ _trim_str(x, limit=200) for x in (spec.get("anti_patterns") or []) if _trim_str(x, limit=200) ][:3] snap: Dict[str, Any] = { "primary_topic": topic, "start_situation": start, "target_state": target, "roadmap_notes": notes, "stage_learning_goal": _trim_str( spec.get("learning_goal"), limit=1200 ), "stage_phase": _trim_str(spec.get("phase")), "stage_exercise_type": _trim_str(spec.get("exercise_type")), "stage_load_profile": load_profile or None, "stage_success_criteria": stage_success or None, "stage_anti_patterns": anti_patterns or None, "path_success_criteria": success_path or None, "skill_hints": skill_hints or None, } return {k: v for k, v in snap.items() if v is not None and v != "" and v != []} 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, goal_analysis: Optional[Mapping[str, Any]] = None, resolved_structured: Optional[Mapping[str, Any]] = None, stage_spec: Optional[Mapping[str, Any]] = None, semantic_brief: Optional[Mapping[str, Any]] = None, stage_learning_goal_override: Optional[str] = None, gap_trainer_supplements: 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, } snap = build_progression_gap_snapshot( goal_analysis=goal_analysis, resolved_structured=resolved_structured, stage_spec=stage_spec, semantic_brief=semantic_brief, ) ctx.update(snap) if stage_learning_goal_override and stage_learning_goal_override.strip(): ctx["stage_learning_goal"] = _trim_str(stage_learning_goal_override, limit=1200) ctx["roadmap_learning_goal"] = ctx["stage_learning_goal"] if gap_trainer_supplements and gap_trainer_supplements.strip(): ctx["gap_trainer_supplements"] = _trim_str(gap_trainer_supplements, limit=2000) return sanitize_planning_context_for_ai(ctx) __all__ = [ "build_progression_gap_snapshot", "build_progression_path_gap_planning_context", "compact_planning_context_json", "planning_context_prompt_variables", "sanitize_planning_context_for_ai", ]