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 _apply_aliases(self, text: str) -> str:
        """Wendet benutzerdefinierte Wort-Aliases an."""
        if not self._custom_aliases:
            return text

        words = text.split()
        changed = False

        for i, word in enumerate(words):
            replacement = self._custom_aliases.get(word.lower())
            if 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