Trixy7/trixy_core/stt/corrector.py

# -*- 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