# -*- coding: utf-8 -*- """ Text-Input-Handler für Server. Verarbeitet TextInput-Commands vom Client und integriert sie in das Conversation- und NLP-System. Flow: 1. Client sendet TextInput Command 2. Server empfängt text_input_received Event 3. Handler erstellt/fortsetzt Conversation 4. Handler emittiert speech_recognized Event 5. NLP Pipeline verarbeitet (wie bei echtem STT) 6. Bei Follow-Up: Server sendet FollowUpRequest zurück """ import uuid from typing import TYPE_CHECKING from trixy_core.events.decorators import TrixyEvent from trixy_core.events.event_data.basic import SpeechRecognized from trixy_core.network.cmd import TextInputAck, FollowUpRequest from trixy_core.utils.debug import pinfo, pdebug, perror if TYPE_CHECKING: from trixy_core.application import IApplication from trixy_core.conversation.manager import ConversationManager class TextInputHandler: """ Handler für Text-Input vom Client. Integriert Keyboard-Eingaben in das Conversation-System und leitet sie an die NLP-Pipeline weiter. Usage: handler = TextInputHandler(app) handler.register() # Registriert Event-Listener """ def __init__(self, application: "IApplication") -> None: """ Initialisiert den Handler. Args: application: Die Server-Anwendung """ self._app = application self._active_sessions: dict[str, str] = {} # session_id -> satellite_id def register(self) -> None: """Registriert die Event-Handler.""" # Event-Handler für text_input_received self._app.events.register("text_input_received", self._on_text_input) # Event-Handler für intent_handled (für Follow-Up-Erkennung) self._app.events.register("intent_handled", self._on_intent_handled) # Event-Handler für output_text_created (Follow-Up mit expects_response) self._app.events.register("output_text_created", self._on_output_created) pdebug("TextInputHandler registriert") def unregister(self) -> None: """Entfernt die Event-Handler.""" self._app.events.unregister("text_input_received", self._on_text_input) self._app.events.unregister("intent_handled", self._on_intent_handled) self._app.events.unregister("output_text_created", self._on_output_created) async def _on_text_input(self, event_name: str, data: dict) -> None: """ Verarbeitet eingehende Text-Eingaben. Args: event_name: Event-Name data: Event-Daten mit TextInput-Feldern """ # Daten extrahieren satellite_id = data.get("satellite_id", "") text = data.get("text", "") session_id = data.get("session_id", "") room_id = data.get("satellite_room", data.get("room_id", "")) language = data.get("language", "de") is_follow_up = data.get("is_follow_up", False) if not text: perror("TextInput ohne Text empfangen") return pdebug(f"TextInput von {satellite_id}: '{text[:50]}...' (session={session_id})") # Session-ID generieren falls keine vorhanden if not session_id: session_id = f"txt-{uuid.uuid4().hex[:8]}" # Session tracken self._active_sessions[session_id] = satellite_id # Bestätigung senden await self._send_ack(satellite_id, session_id) # SpeechRecognized Event emittieren (wie bei echtem STT) speech_data = SpeechRecognized( satellite_id=satellite_id, room_id=room_id, text=text, confidence=1.0, language=language, is_final=True, source="keyboard", ) # Zusätzliche Felder für Conversation-Tracking speech_data.session_id = session_id # type: ignore speech_data.is_follow_up = is_follow_up # type: ignore await self._app.events.trigger("speech_recognized", speech_data) async def _send_ack(self, satellite_id: str, session_id: str) -> None: """ Sendet Bestätigung an den Client. Args: satellite_id: ID des Satellites session_id: Session-ID """ try: # Satellite finden satellites = getattr(self._app, "satellites", None) if not satellites: return satellite = satellites.get(satellite_id) if not satellite: # Versuche über Alias for sat in satellites.all(): if sat.alias == satellite_id: satellite = sat break if satellite: ack = TextInputAck( session_id=session_id, received=True, ) await satellite.send_command(ack) pdebug(f"TextInputAck gesendet an {satellite_id}") except Exception as e: perror(f"Fehler beim Senden von TextInputAck: {e}") async def _on_intent_handled(self, event_name: str, data) -> None: """ Prüft ob ein Follow-Up erwartet wird. Args: event_name: Event-Name data: IntentHandled-Event-Daten """ # Prüfen ob Follow-Up nötig needs_followup = getattr(data, "needs_followup", False) if isinstance(data, dict): needs_followup = data.get("needs_followup", False) if not needs_followup: return # Session-ID und Satellite ermitteln session_id = getattr(data, "session_id", "") or "" satellite_id = getattr(data, "satellite_id", "") or "" followup_prompt = getattr(data, "followup_prompt", "") or "" if isinstance(data, dict): session_id = data.get("session_id", "") satellite_id = data.get("satellite_id", "") followup_prompt = data.get("followup_prompt", "") # Nur für Keyboard-Sessions if session_id not in self._active_sessions: return pdebug(f"Follow-Up für TextInput-Session: {session_id}") # FollowUpRequest wird nach TTS gesendet (in _on_output_created) async def _on_output_created(self, event_name: str, data) -> None: """ Prüft ob nach Output ein Follow-Up-Input erwartet wird. Args: event_name: Event-Name data: OutputTextCreated-Event-Daten """ expects_response = getattr(data, "expects_response", False) if isinstance(data, dict): expects_response = data.get("expects_response", False) if not expects_response: return # Session-ID und Satellite ermitteln session_id = getattr(data, "session_id", "") or "" satellite_id = getattr(data, "satellite_id", "") or "" text = getattr(data, "text", "") or "" if isinstance(data, dict): session_id = data.get("session_id", "") satellite_id = data.get("satellite_id", "") text = data.get("text", "") # Nur für Keyboard-Sessions if session_id not in self._active_sessions: return # FollowUpRequest an Client senden await self._send_follow_up_request( satellite_id=satellite_id or self._active_sessions.get(session_id, ""), session_id=session_id, question=text, ) async def _send_follow_up_request( self, satellite_id: str, session_id: str, question: str, ) -> None: """ Sendet Follow-Up-Request an den Client. Args: satellite_id: ID des Satellites session_id: Session-ID question: Die Rückfrage """ try: satellites = getattr(self._app, "satellites", None) if not satellites: return satellite = satellites.get(satellite_id) if not satellite: for sat in satellites.all(): if sat.alias == satellite_id: satellite = sat break if satellite: request = FollowUpRequest( session_id=session_id, question=question, ) await satellite.send_command(request) pdebug(f"FollowUpRequest gesendet an {satellite_id}: {question[:50]}...") except Exception as e: perror(f"Fehler beim Senden von FollowUpRequest: {e}") def end_session(self, session_id: str) -> None: """ Beendet eine Text-Input-Session. Args: session_id: Die Session-ID """ self._active_sessions.pop(session_id, None) def setup_text_input_handler(application: "IApplication") -> TextInputHandler: """ Erstellt und registriert den TextInputHandler. Args: application: Die Server-Anwendung Returns: Der registrierte Handler """ handler = TextInputHandler(application) handler.register() return handler