# -*- coding: utf-8 -*- """ STT-Korrektur-Service. Mehrstufiger Post-Processor fuer STT-Ausgaben. Registriert sich mit HIGH-Prioritaet auf speech_recognized und korrigiert den Text in-place, bevor das NLP-Plugin (NORMAL-Prioritaet) ihn verarbeitet. """ from trixy_core.service.iservice import IService from trixy_core.service.enums import ServicePriority, ServiceGroup from trixy_core.events.decorators import TrixyEvent from trixy_core.events.enums import EventPriority from trixy_core.events.event_data.basic import SpeechRecognized from trixy_core.stt.layers.base import CorrectionLayer from trixy_core.utils.debug import pinfo, pdebug, pwarn class STTCorrectorService(IService): """ Service fuer STT-Nachkorrektur. Wendet konfigurierbare Korrektur-Schichten sequenziell auf erkannten Text an, bevor er vom NLP-Plugin verarbeitet wird. """ PRIORITY = ServicePriority.OPTIONAL GROUP = ServiceGroup.CONVERSATION DEPENDENCIES: list[str] = [] NAME = "STTCorrectorService" def __init__(self, application) -> None: super().__init__(application) self._active_layers: list[CorrectionLayer] = [] self._custom_aliases: dict[str, str] = {} self._enabled: bool = False async def start(self) -> None: """Startet den STT-Korrektur-Service.""" pinfo("Starte STTCorrectorService...") # Konfiguration lesen config = self._get_config() if config is None: pwarn("[STT-Korrektur] Keine Konfiguration gefunden") return if not config.enabled: pinfo("[STT-Korrektur] Deaktiviert per Konfiguration") return self._enabled = True self._custom_aliases = {k.lower(): v for k, v in config.custom_aliases.items()} # Config als dict fuer Layer-Initialisierung config_dict = { "symspell_max_edit_distance": config.symspell_max_edit_distance, "symspell_dictionary_path": config.symspell_dictionary_path, "kenlm_model_path": config.kenlm_model_path, "jamspell_model_path": config.jamspell_model_path, "hunspell_dict_path": config.hunspell_dict_path, "languagetool_language": config.languagetool_language, } # Layers in konfigurierter Reihenfolge initialisieren from trixy_core.stt.layers import LAYER_REGISTRY for layer_name in config.layers: layer_class = LAYER_REGISTRY.get(layer_name) if layer_class is None: pwarn(f"[STT-Korrektur] Unbekannte Schicht: '{layer_name}'") continue layer = layer_class() if not layer.is_available(): pinfo(f"[STT-Korrektur] Schicht '{layer_name}' nicht verfuegbar (Library fehlt)") continue if layer.initialize(config_dict, config.language, config.protected_words): self._active_layers.append(layer) pinfo(f"[STT-Korrektur] Schicht '{layer_name}' aktiviert") else: pinfo(f"[STT-Korrektur] Schicht '{layer_name}' konnte nicht initialisiert werden") if self._active_layers: layer_names = [l.NAME for l in self._active_layers] pinfo(f"[STT-Korrektur] Gestartet mit Schichten: {layer_names}") elif self._custom_aliases: pinfo(f"[STT-Korrektur] Gestartet nur mit Alias-Korrektur ({len(self._custom_aliases)} Aliases)") else: pinfo("[STT-Korrektur] Keine Schichten verfuegbar, nur Passthrough") async def stop(self) -> None: """Stoppt den STT-Korrektur-Service.""" for layer in self._active_layers: try: layer.shutdown() except Exception as e: pdebug(f"[STT-Korrektur] Fehler beim Shutdown von '{layer.NAME}': {e}") self._active_layers.clear() self._custom_aliases.clear() self._enabled = False pinfo("STTCorrectorService gestoppt") @TrixyEvent(["speech_recognized"], priority=EventPriority.HIGH) async def on_speech_recognized(self, event_name: str, event_data: SpeechRecognized) -> None: """Korrigiert STT-Text bevor das NLP-Plugin ihn verarbeitet.""" if not self._enabled: return # Keyboard-Input nicht korrigieren if getattr(event_data, "source", "") == "keyboard": return text = event_data.text if not text or not text.strip(): return corrected = text # 1. Custom-Aliases anwenden (einfache Wort-Ersetzung) corrected = self._apply_aliases(corrected) # 2. Korrektur-Schichten durchlaufen for layer in self._active_layers: try: corrected = layer.correct(corrected) except Exception as e: pdebug(f"[STT-Korrektur] Fehler in Schicht '{layer.NAME}': {e}") # In-place Modifikation des EventData-Objekts if corrected != text: pinfo(f"[STT-Korrektur] '{text}' → '{corrected}'") event_data.text = corrected def _load_plugin_aliases(self) -> None: """ Laedt Plugin-eigene STT-Aliases aus plugins/*/stt_aliases.json. Jedes Plugin kann eine stt_aliases.json bereitstellen mit Korrekturen die nur relevant sind wenn das Plugin aktiv ist. Format: {"falsche erkennung": "korrekt", ...} """ import json from pathlib import Path plugins_dir = Path("plugins") if not plugins_dir.is_dir(): return loaded_count = 0 for alias_file in sorted(plugins_dir.glob("*/stt_aliases.json")): plugin_name = alias_file.parent.name # Nur laden wenn Plugin aktiviert ist (config.json → enabled: true) plugin_config = alias_file.parent / "config.json" if plugin_config.is_file(): try: with open(plugin_config) as f: pcfg = json.load(f) if not pcfg.get("enabled", True): continue except Exception: pass try: with open(alias_file, encoding="utf-8") as f: aliases = json.load(f) count = 0 for key, value in aliases.items(): if key.startswith("_"): # _comment etc. ueberspringen continue self._custom_aliases[key.lower()] = value count += 1 if count > 0: loaded_count += count pdebug(f"[STT-Korrektur] {count} Aliases aus Plugin '{plugin_name}' geladen") except (json.JSONDecodeError, OSError) as e: pdebug(f"[STT-Korrektur] Fehler beim Laden von {alias_file}: {e}") if loaded_count > 0: pinfo(f"[STT-Korrektur] {loaded_count} Plugin-Aliases geladen") def _apply_aliases(self, text: str) -> str: """ Wendet benutzerdefinierte Wort-Aliases an. Unterstuetzt Einzelwort- und Mehrwort-Phrasen: - "shop" → "stopp" (Einzelwort) - "nach utopie" → "naruto" (Mehrwort-Phrase) - "etc anime" → "ecchi anime" (Mehrwort-Phrase) """ if not self._custom_aliases: return text result = text changed = False # Erst Mehrwort-Phrasen ersetzen (laengere zuerst) multi_word = {k: v for k, v in self._custom_aliases.items() if " " in k} for phrase in sorted(multi_word, key=len, reverse=True): if phrase in result.lower(): # Position finden und ersetzen (Case-insensitive) lower = result.lower() idx = lower.find(phrase) while idx >= 0: result = result[:idx] + multi_word[phrase] + result[idx + len(phrase):] changed = True lower = result.lower() idx = lower.find(phrase, idx + len(multi_word[phrase])) # Dann Einzelwort-Ersetzungen words = result.split() for i, word in enumerate(words): replacement = self._custom_aliases.get(word.lower()) if replacement and " " not in replacement: words[i] = replacement changed = True if changed: result = " ".join(words) pdebug(f"[STT-Korrektur] Alias: '{text}' → '{result}'") return result return text def _get_config(self): """Liest die STT-Korrektur-Konfiguration aus der Anwendung.""" app = self._application # Standalone-Modus if hasattr(app, "standalone_config"): return app.standalone_config.stt_correction # Server-Modus if hasattr(app, "server_config"): return app.server_config.stt_correction return None