""" Gemeinsame Intent-Anreicherung für Planungs-Retrieval. Progressionsgraph (Roadmap stage_specs) und später Trainingsplanung (Abschnitt/Slot) nutzen dieselben Bausteine: Intent-Kontext bauen → Specs finalisieren → Matching-Gates. """ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Any, Dict, List, Mapping, Optional, Sequence from planning_exercise_semantics import ( PlanningSemanticBrief, resolve_path_anti_patterns, technique_sibling_excludes, ) _NEGATION_CLAUSE_RE = re.compile( r"\b(?:ohne|kein(?:e|en|er|em)?|nicht)\s+[^,.;\n]+", flags=re.IGNORECASE, ) def extract_explicit_exclusions(*texts: Optional[str]) -> List[str]: """Lesbare Negationsklauseln aus Freitext (ohne Themen-Raten).""" out: List[str] = [] for raw in texts: s = (raw or "").strip() if not s: continue for m in _NEGATION_CLAUSE_RE.finditer(s): clause = m.group(0).strip().rstrip(".,;") if clause and clause.lower() not in {x.lower() for x in out}: out.append(clause[:220]) return out[:12] @dataclass class PlanningIntentContext: """Pfad-/Abschnittsweiter Planungs-Intent — domänenneutral.""" source_query: str = "" primary_topic: str = "" path_anti_patterns: List[str] = field(default_factory=list) path_success_criteria: List[str] = field(default_factory=list) explicit_exclusions: List[str] = field(default_factory=list) context_notes: str = "" topic_type: str = "general" technique_sibling_excludes: List[str] = field(default_factory=list) def to_api_dict(self) -> Dict[str, Any]: return { "source_query": self.source_query, "primary_topic": self.primary_topic, "topic_type": self.topic_type, "path_anti_patterns": self.path_anti_patterns[:16], "path_success_criteria": self.path_success_criteria[:10], "explicit_exclusions": self.explicit_exclusions[:10], "technique_sibling_excludes": self.technique_sibling_excludes[:16], "context_notes": self.context_notes[:1200] or None, } def build_planning_intent_context( goal_query: str, *, semantic_brief: Optional[PlanningSemanticBrief] = None, goal_analysis: Optional[Mapping[str, Any]] = None, extra_context: Optional[str] = None, primary_topic: Optional[str] = None, ) -> PlanningIntentContext: """Intent aus Anfrage, Zielanalyse und optionalem Kontext — ohne Sonderregeln pro Thema.""" ga = dict(goal_analysis or {}) notes_parts = [extra_context or ""] constraints = ga.get("constraints") if isinstance(ga.get("constraints"), dict) else {} if isinstance(constraints, dict): trainer_notes = str(constraints.get("trainer_notes") or "").strip() if trainer_notes: notes_parts.append(trainer_notes) combined_notes = " ".join(p.strip() for p in notes_parts if p and p.strip()) explicit = extract_explicit_exclusions(goal_query, combined_notes or None) ga_excluded = constraints.get("excluded_themes") if isinstance(constraints, dict) else None if isinstance(ga_excluded, list): for item in ga_excluded: s = str(item or "").strip() if s and s.lower() not in {x.lower() for x in explicit}: explicit.append(s[:220]) path_anti = resolve_path_anti_patterns( goal_query, semantic_brief=semantic_brief, extra_context=combined_notes or None, ) path_success: List[str] = [] for item in ga.get("success_criteria") or []: s = str(item or "").strip() if s and s not in path_success: path_success.append(s[:240]) target = str(ga.get("target_state") or "").strip() if target and len(target) >= 8: line = f"Zielzustand erreichbar: {target[:200]}" if line not in path_success: path_success.append(line) topic = (primary_topic or ga.get("primary_topic") or "").strip() topic_type = "general" siblings: List[str] = [] if semantic_brief: if not topic: topic = (semantic_brief.primary_topic or "").strip() topic_type = (semantic_brief.topic_type or "general").strip().lower() if topic_type == "technique" and topic: siblings = technique_sibling_excludes(topic) for raw in semantic_brief.exclude_phrases or []: s = str(raw or "").strip() if s and s.lower() not in {x.lower() for x in siblings}: siblings.append(s[:120]) if topic_type == "technique" and topic: line = f"Haupttechnik {topic} in Kurzbeschreibung oder Übungsziel erkennbar" if line not in path_success: path_success.insert(0, line) return PlanningIntentContext( source_query=(goal_query or "").strip(), primary_topic=topic, topic_type=topic_type, path_anti_patterns=path_anti, path_success_criteria=path_success, explicit_exclusions=explicit, technique_sibling_excludes=siblings[:16], context_notes=combined_notes[:1200], ) def _dedupe_preserve(items: Sequence[str], *, limit: int = 14) -> List[str]: out: List[str] = [] seen: set[str] = set() for raw in items: s = str(raw or "").strip() if not s: continue key = s.lower() if key in seen: continue seen.add(key) out.append(s[:240]) if len(out) >= limit: break return out def finalize_stage_spec_artifact( spec: "StageSpecArtifact", *, major_step: Optional["MajorStep"] = None, intent: PlanningIntentContext, ) -> "StageSpecArtifact": """Pfad-Intent in eine Stufenspezifikation mergen (LLM oder heuristisch).""" from planning_progression_roadmap import MajorStep, StageSpecArtifact learning_goal = (spec.learning_goal or (major_step.learning_goal if major_step else "")).strip() phase = (major_step.phase if major_step else "").strip().lower() anti = _dedupe_preserve( [ *(spec.anti_patterns or []), *intent.explicit_exclusions, *intent.path_anti_patterns, *intent.technique_sibling_excludes, ( f"andere Technik als {intent.primary_topic}" if intent.topic_type == "technique" and intent.primary_topic else "" ), ], limit=14, ) stage_start = (spec.start_state or "").strip() stage_target = (spec.target_state or "").strip() success = _dedupe_preserve( [ *(spec.success_criteria or []), *intent.path_success_criteria, (f"Soll-Start der Stufe erreichbar: {stage_start[:180]}" if stage_start else ""), (f"Stufen-Ziel erreichbar: {stage_target[:180]}" if stage_target else ""), ( f"Übung liefert messbar: {learning_goal[:160]}" if learning_goal else "" ), ( f"Kurzbeschreibung und Übungsziel passen zur Phase {phase}" if phase else "Kurzbeschreibung und Übungsziel passen zum Stufen-Lernziel" ), ], limit=8, ) idx = spec.major_step_index if major_step is not None: idx = major_step.index return StageSpecArtifact( major_step_index=idx, learning_goal=learning_goal, load_profile=list(spec.load_profile or []), exercise_type=(spec.exercise_type or "").strip(), success_criteria=success, anti_patterns=anti, ) def finalize_stage_specs_with_intent( specs: Sequence["StageSpecArtifact"], major_steps: Sequence["MajorStep"], *, intent: PlanningIntentContext, fallback_specs: Optional[Sequence["StageSpecArtifact"]] = None, ) -> List["StageSpecArtifact"]: """Alle Stufen mit gleichem Pfad-Intent anreichern; fehlende Indizes aus Fallback.""" from planning_progression_roadmap import MajorStep, StageSpecArtifact by_idx = {int(s.major_step_index): s for s in specs} fallback_by_idx = {int(s.major_step_index): s for s in (fallback_specs or [])} out: List[StageSpecArtifact] = [] for major in major_steps: raw = by_idx.get(major.index) or fallback_by_idx.get(major.index) if raw is None: raw = StageSpecArtifact( major_step_index=major.index, learning_goal=major.learning_goal, ) out.append(finalize_stage_spec_artifact(raw, major_step=major, intent=intent)) return out __all__ = [ "PlanningIntentContext", "build_planning_intent_context", "extract_explicit_exclusions", "finalize_stage_spec_artifact", "finalize_stage_specs_with_intent", ]