""" 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, Sequence _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 _major_index_from_step(step: Mapping[str, Any]) -> Optional[int]: for key in ("roadmap_major_step_index", "major_step_index"): raw = step.get(key) if raw is None: continue try: return int(raw) except (TypeError, ValueError): continue return None def prior_path_steps_before_major( steps: Sequence[Mapping[str, Any]], major_idx: int, ) -> List[Dict[str, Any]]: """Pfadschritte mit kleinerem roadmap_major_step_index, sortiert.""" prior: List[Dict[str, Any]] = [] for step in steps: mi = _major_index_from_step(step) if mi is not None and mi < major_idx: prior.append(dict(step)) prior.sort(key=lambda s: _major_index_from_step(s) or 0) return prior def _step_display_fields(step: Mapping[str, Any]) -> Dict[str, Any]: title = _trim_str( step.get("title") or step.get("exercise_title"), limit=200, ) learning_goal = _trim_str( step.get("roadmap_learning_goal") or step.get("learning_goal"), limit=500, ) summary = _trim_str(step.get("summary"), limit=400) start_state = _trim_str(step.get("roadmap_start_state") or step.get("start_state")) target_state = _trim_str(step.get("roadmap_target_state") or step.get("target_state")) phase = _trim_str(step.get("roadmap_phase") or step.get("phase")) criteria_raw = step.get("stage_success_criteria") or step.get("success_criteria") or [] criteria = [ t for x in criteria_raw if (t := _trim_str(x, limit=200)) ][:4] out: Dict[str, Any] = { "title": title, "learning_goal": learning_goal, "summary": summary, "start_state": start_state, "target_state": target_state, "phase": phase, "success_criteria": criteria or None, "major_step_index": _major_index_from_step(step), } return {k: v for k, v in out.items() if v is not None and v != "" and v != []} def build_progression_entry_state( *, major_step_index: Optional[int] = None, prior_steps: Sequence[Mapping[str, Any]] = (), start_situation: Optional[str] = None, current_stage_start: Optional[str] = None, ) -> Dict[str, Any]: """ Eingangszustand für eine Roadmap-Stufe: erreichte Voraussetzungen aus Vorstufen. """ prior_compact = [_step_display_fields(s) for s in prior_steps] prior_compact = [ p for p in prior_compact if any(p.get(k) for k in ("title", "learning_goal", "summary", "success_criteria")) ] achievements: List[str] = [] detail_lines: List[str] = [] for p in prior_compact: if p.get("success_criteria"): achievements.extend(p["success_criteria"]) elif p.get("learning_goal"): achievements.append(p["learning_goal"]) label_parts: List[str] = [] if p.get("major_step_index") is not None: label_parts.append(f"Stufe {int(p['major_step_index']) + 1}") if p.get("phase"): label_parts.append(f"({p['phase']})") if p.get("title"): label_parts.append(f"„{p['title']}\"") prefix = " ".join(label_parts) if label_parts else "Vorstufe" achieved = "" if p.get("target_state"): achieved = p["target_state"] elif p.get("success_criteria"): achieved = "; ".join(p["success_criteria"]) elif p.get("learning_goal"): achieved = p["learning_goal"] elif p.get("summary"): achieved = p["summary"] if achieved: detail_lines.append(f"{prefix}: erreicht — {achieved}") immediate_entry: Optional[str] = _trim_str(current_stage_start) if not immediate_entry and prior_compact: immediate = prior_compact[-1] if immediate.get("target_state"): immediate_entry = immediate["target_state"] elif immediate.get("success_criteria"): immediate_entry = "; ".join(immediate["success_criteria"]) elif immediate.get("learning_goal"): immediate_entry = immediate["learning_goal"] elif immediate.get("summary"): immediate_entry = immediate["summary"] elif not immediate_entry and start_situation: immediate_entry = start_situation entry_state = immediate_entry or start_situation if prior_compact and start_situation and not immediate_entry: detail_lines.insert(0, f"Ausgangsbasis Pfad: {start_situation}") out: Dict[str, Any] = {} if entry_state: out["entry_state"] = _trim_str(entry_state, limit=1200) if detail_lines: out["entry_state_detail"] = _trim_str("\n".join(detail_lines), limit=2000) if prior_compact: out["prior_steps"] = prior_compact[:6] if achievements: out["prior_achievements"] = list(dict.fromkeys(achievements))[:8] return out def enrich_gap_snapshot_with_entry_state( snapshot: Mapping[str, Any], *, steps: Sequence[Mapping[str, Any]], major_step_index: Optional[int], ) -> Dict[str, Any]: snap = dict(snapshot) if major_step_index is None: return snap try: mi = int(major_step_index) except (TypeError, ValueError): return snap prior = prior_path_steps_before_major(steps, mi) entry = build_progression_entry_state( major_step_index=mi, prior_steps=prior, start_situation=snap.get("start_situation"), current_stage_start=snap.get("stage_start_state"), ) snap.update(entry) return snap 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_start_state": _trim_str(spec.get("start_state")), "stage_target_state": _trim_str(spec.get("target_state")), "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, prior_path_steps: Optional[Sequence[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 major_idx is not None and prior_path_steps: ctx.update( build_progression_entry_state( major_step_index=major_idx, prior_steps=list(prior_path_steps), start_situation=ctx.get("start_situation"), ) ) 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_entry_state", "build_progression_gap_snapshot", "build_progression_path_gap_planning_context", "enrich_gap_snapshot_with_entry_state", "prior_path_steps_before_major", "compact_planning_context_json", "planning_context_prompt_variables", "sanitize_planning_context_for_ai", ]