""" Planungs-KI Phase E2/E3: KI-Neuanlage für Pfad-Lücken + strukturierte Angebote für die UI. """ from __future__ import annotations import logging import uuid from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple from ai_prompt_context import ExerciseFormAiPromptContext from ai_prompt_job import run_exercise_form_ai_suggestion from exercise_ai import strip_html_to_plain from planning_exercise_path_qa import find_step_pair_index from planning_exercise_semantics import PlanningSemanticBrief _logger = logging.getLogger("shinkan.planning_exercise_path_ai_fill") def _build_gap_ai_context( *, goal_query: str, brief: PlanningSemanticBrief, step_a: Mapping[str, Any], step_b: Mapping[str, Any], gap: Mapping[str, Any], title_hint: Optional[str] = None, sketch_hint: Optional[str] = None, ) -> ExerciseFormAiPromptContext: topic = (brief.primary_topic or "Technik").strip() phase = gap.get("expected_phase") or "vertiefung" from_title = (step_a.get("title") or f"Übung #{step_a.get('exercise_id')}").strip() to_title = (step_b.get("title") or f"Übung #{step_b.get('exercise_id')}").strip() title = (title_hint or f"Brücke {topic} ({phase})").strip()[:280] sketch = (sketch_hint or "").strip() goal_parts = [ f"Planungsziel: {goal_query}", "", f"Didaktische Brücken-Übung zwischen „{from_title}“ und „{to_title}“.", f"Phase: {phase}. Thema: {topic}.", "Die Übung schließt die Lücke im Progressionspfad und bereitet sinnvoll auf den nächsten Schritt vor.", ] if sketch: goal_parts.extend(["", f"Hinweis: {sketch}"]) goal = "\n".join(goal_parts) focus_hint = topic if brief.topic_type == "technique" else None if brief.must_phrases: focus_hint = ", ".join(brief.must_phrases[:2]) return ExerciseFormAiPromptContext( title=title[:280], goal=goal[:8000], execution=None, focus_hint=focus_hint, ) def ai_proposal_to_path_step( *, ai_payload: Mapping[str, Any], ctx_title: str, gap: Mapping[str, Any], step_a: Mapping[str, Any], step_b: Mapping[str, Any], ) -> Dict[str, Any]: summary_text = "" summary_obj = ai_payload.get("summary") if isinstance(summary_obj, dict): summary_text = str(summary_obj.get("text") or "").strip() elif isinstance(summary_obj, str): summary_text = summary_obj.strip() proposal_key = f"ai-{uuid.uuid4().hex[:10]}" title = (ctx_title or "").strip() or "KI-Vorschlag (Brücke)" reasons = ["KI-Neuanlage-Vorschlag — Lücke ohne passende Bibliotheks-Übung"] return { "exercise_id": None, "proposal_key": proposal_key, "variant_id": None, "title": title, "summary": summary_text or None, "score": None, "semantic_score": None, "reasons": reasons, "variants": [], "is_bridge": True, "is_ai_proposal": True, "ai_suggestion": dict(ai_payload), "bridge_for_gap": { "from_exercise_id": step_a.get("exercise_id"), "to_exercise_id": step_b.get("exercise_id"), "gap_score": gap.get("gap_score"), "expected_phase": gap.get("expected_phase"), }, } def try_suggest_ai_bridge_step( cur, *, goal_query: str, brief: PlanningSemanticBrief, step_a: Mapping[str, Any], step_b: Mapping[str, Any], gap: Mapping[str, Any], title_hint: Optional[str] = None, sketch_hint: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """Ruft exercise AI suggest auf — kein Speichern in DB.""" ctx = _build_gap_ai_context( goal_query=goal_query, brief=brief, step_a=step_a, step_b=step_b, gap=gap, title_hint=title_hint, sketch_hint=sketch_hint, ) g_plain = strip_html_to_plain(ctx.goal) if not g_plain.strip() and not (ctx.title or "").strip(): return None try: payload = run_exercise_form_ai_suggestion( cur, ctx, want_summary=True, want_skills=True, want_instructions=False, ) except Exception as exc: _logger.warning("KI-Lückenfüller fehlgeschlagen: %s", exc) return None if not payload: return None return ai_proposal_to_path_step( ai_payload=payload, ctx_title=ctx.title or "", gap=gap, step_a=step_a, step_b=step_b, ) def _default_sketch( *, goal_query: str, brief: PlanningSemanticBrief, step_a: Optional[Mapping[str, Any]], step_b: Optional[Mapping[str, Any]], phase: str, rationale: str = "", ) -> str: topic = (brief.primary_topic or "Technik").strip() from_t = (step_a or {}).get("title") or "vorherigem Schritt" to_t = (step_b or {}).get("title") or "nächstem Schritt" parts = [ f"Planungsziel: {goal_query}", f"Zwischenschritt für {topic} ({phase}) zwischen „{from_t}“ und „{to_t}“.", ] if rationale: parts.append(rationale) return " ".join(parts)[:1200] def _spec_dedupe_key(spec: Mapping[str, Any]) -> Tuple[Any, ...]: return ( spec.get("source"), int(spec.get("insert_after_index") or 0), str(spec.get("title_hint") or "")[:48], ) def collect_gap_fill_specs( *, steps: Sequence[Mapping[str, Any]], unfilled_gaps: Sequence[Mapping[str, Any]], off_topic_steps: Sequence[Mapping[str, Any]], llm_specs: Sequence[Mapping[str, Any]], brief: PlanningSemanticBrief, goal_query: str, ) -> List[Dict[str, Any]]: """Sammelt alle Lücken, für die ein KI-Anlege-Angebot sinnvoll ist.""" topic = (brief.primary_topic or "Technik").strip() specs: List[Dict[str, Any]] = [] seen: set = set() def add(spec: Dict[str, Any]) -> None: key = _spec_dedupe_key(spec) if key in seen: return seen.add(key) specs.append(spec) for gap in unfilled_gaps: idx = find_step_pair_index( steps, int(gap["from_exercise_id"]), int(gap["to_exercise_id"]), ) if idx is None: continue phase = gap.get("expected_phase") or "vertiefung" add( { "source": "unfilled_gap", "insert_after_index": idx, "gap": dict(gap), "phase": phase, "title_hint": f"{topic} — {phase}", "sketch": _default_sketch( goal_query=goal_query, brief=brief, step_a=steps[idx], step_b=steps[idx + 1], phase=str(phase), rationale="Bibliothek enthält keine passende Brücke.", ), "rationale": "Lücke zwischen benachbarten Schritten — keine passende Bibliotheks-Übung.", } ) for ot in off_topic_steps: idx = int(ot.get("step_index") or 0) if idx <= 0 or idx >= len(steps) - 1: continue phase = ot.get("expected_phase") or "vertiefung" add( { "source": "off_topic", "insert_after_index": idx - 1, "replace_step_index": idx, "gap": { "expected_phase": phase, "off_topic_title": ot.get("title"), "off_topic_exercise_id": ot.get("exercise_id"), }, "phase": phase, "title_hint": f"{topic} — {phase} (Ersatz für themenfremden Schritt)", "sketch": _default_sketch( goal_query=goal_query, brief=brief, step_a=steps[idx - 1], step_b=steps[idx + 1], phase=str(phase), rationale=f"Ersetzt themenfremden Schritt „{ot.get('title')}“.", ), "rationale": f"Schritt „{ot.get('title')}“ passt nicht zum Pfad-Thema.", } ) for spec in llm_specs: add(dict(spec)) return specs[:5] def build_gap_fill_goal_text( *, goal_query: str, brief: PlanningSemanticBrief, spec: Mapping[str, Any], step_a: Optional[Mapping[str, Any]] = None, step_b: Optional[Mapping[str, Any]] = None, ) -> str: """Ausführlicher Zieltext für KI-Neuanlage aus dem Pfad-Kontext.""" topic = (brief.primary_topic or "Technik").strip() phase = spec.get("phase") or "vertiefung" from_title = (step_a or {}).get("title") or spec.get("from_title") or "vorherigem Schritt" to_title = (step_b or {}).get("title") or spec.get("to_title") or "nächstem Schritt" arc = ", ".join(brief.development_arc or []) or "einstieg → grundlage → vertiefung → anwendung → perfektion" parts = [ f"Planungsziel (gesamter Pfad): {goal_query}", f"Hauptthema: {topic}", f"Entwicklungsphase dieser Übung: {phase}", f"Erwarteter Entwicklungsbogen: {arc}", f"Einordnung: didaktische Zwischenstufe zwischen „{from_title}“ und „{to_title}“.", ] if spec.get("rationale"): parts.append(f"Qualitätsprüfung: {spec['rationale']}") if spec.get("sketch"): parts.append(f"Skizze: {spec['sketch']}") parts.append( "Die Übung muss einen klaren, trainierbaren Bezug zum Hauptthema haben — " "keine generische Kraftübung ohne Technikbezug. Konkrete Durchführung, Ziel und Trainerhinweise ausformulieren." ) return "\n\n".join(parts)[:8000] def build_gap_fill_offer( *, spec: Mapping[str, Any], steps: Sequence[Mapping[str, Any]], goal_query: str = "", brief: Optional[PlanningSemanticBrief] = None, proposal: Optional[Mapping[str, Any]] = None, ) -> Dict[str, Any]: idx = int(spec.get("insert_after_index") or 0) offer_id = f"{spec.get('source')}-{idx}-{uuid.uuid4().hex[:8]}" step_a = steps[idx] if idx < len(steps) else None step_b = steps[idx + 1] if idx + 1 < len(steps) else None goal_for_ai = "" if brief and goal_query: goal_for_ai = build_gap_fill_goal_text( goal_query=goal_query, brief=brief, spec=spec, step_a=step_a, step_b=step_b, ) offer: Dict[str, Any] = { "offer_id": offer_id, "source": spec.get("source"), "insert_after_index": idx, "replace_step_index": spec.get("replace_step_index"), "title_hint": spec.get("title_hint"), "sketch": spec.get("sketch"), "goal_for_ai": goal_for_ai or spec.get("sketch"), "phase": spec.get("phase"), "rationale": spec.get("rationale"), "has_ai_payload": False, "from_title": (step_a or {}).get("title"), "to_title": (step_b or {}).get("title"), "primary_topic": (brief.primary_topic if brief else None), } if proposal: offer["has_ai_payload"] = True offer["proposal_key"] = proposal.get("proposal_key") offer["ai_suggestion"] = proposal.get("ai_suggestion") offer["proposal_title"] = proposal.get("title") offer["proposal_summary"] = proposal.get("summary") return offer def apply_gap_fill_after_qa( cur, steps: List[Dict[str, Any]], specs: Sequence[Mapping[str, Any]], *, goal_query: str, brief: PlanningSemanticBrief, include_ai_calls: bool = True, max_ai_proposals: int = 3, auto_insert_proposals: bool = False, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, Any]]]: """ Erzeugt gap_fill_offers für die UI; optional KI-Vorschläge einfügen. Returns: (steps, ai_proposals, gap_fill_offers) """ if not specs: return steps, [], [] out = list(steps) proposals: List[Dict[str, Any]] = [] offers: List[Dict[str, Any]] = [] for spec in specs: idx = int(spec.get("insert_after_index") or 0) if idx < 0 or idx >= len(out) - 1: continue step_a = out[idx] step_b = out[idx + 1] if step_a.get("is_ai_proposal") or step_b.get("is_ai_proposal"): offer = build_gap_fill_offer( spec=spec, steps=out, goal_query=goal_query, brief=brief, proposal=None, ) offers.append(offer) continue gap = dict(spec.get("gap") or {}) if not gap.get("expected_phase"): gap["expected_phase"] = spec.get("phase") or "vertiefung" proposal: Optional[Dict[str, Any]] = None if include_ai_calls and len(proposals) < max_ai_proposals: proposal = try_suggest_ai_bridge_step( cur, goal_query=goal_query, brief=brief, step_a=step_a, step_b=step_b, gap=gap, title_hint=str(spec.get("title_hint") or ""), sketch_hint=str(spec.get("sketch") or ""), ) offer = build_gap_fill_offer( spec=spec, steps=out, goal_query=goal_query, brief=brief, proposal=proposal, ) offers.append(offer) if proposal and auto_insert_proposals: out.insert(idx + 1, proposal) proposals.append( { "inserted_after_index": idx, "proposal_key": proposal.get("proposal_key"), "proposal_title": proposal.get("title"), "gap": gap, "offer_id": offer.get("offer_id"), } ) return out, proposals, offers def insert_ai_proposals_for_gaps( cur, steps: list, unfilled_gaps: list, *, goal_query: str, brief: PlanningSemanticBrief, max_proposals: int = 2, ) -> tuple[list, list]: """Legacy: Fügt KI-Vorschläge für Lücken ein, wenn Bibliotheks-Brücke fehlte.""" specs = collect_gap_fill_specs( steps=steps, unfilled_gaps=unfilled_gaps, off_topic_steps=[], llm_specs=[], brief=brief, goal_query=goal_query, ) out, proposals, _offers = apply_gap_fill_after_qa( cur, steps, specs, goal_query=goal_query, brief=brief, include_ai_calls=True, max_ai_proposals=max_proposals, auto_insert_proposals=True, ) return out, proposals __all__ = [ "apply_gap_fill_after_qa", "build_gap_fill_goal_text", "build_gap_fill_offer", "collect_gap_fill_specs", "insert_ai_proposals_for_gaps", "try_suggest_ai_bridge_step", ]