Trixy7/trixy_core/stt/layers/phonetic_layer.py

# -*- coding: utf-8 -*-
"""
Phonetische Korrekturschicht.

Nutzt Koelner Phonetik um phonetisch aehnliche Woerter zu erkennen
und Domain-Vokabular (Kuenstlernamen, Befehle) zu schuetzen.
"""

from trixy_core.stt.layers.base import CorrectionLayer
from trixy_core.utils.debug import pdebug, pwarn

try:
    import cologne_phonetics
    _HAS_COLOGNE = True
except ImportError:
    _HAS_COLOGNE = False


class PhoneticLayer(CorrectionLayer):
    """Koelner-Phonetik-basierte Korrektur fuer Domain-Vokabular."""

    NAME = "phonetic"

    def __init__(self) -> None:
        # phonetischer Code → bevorzugtes Wort
        self._phonetic_index: dict[str, str] = {}
        self._protected_words: set[str] = set()

    def is_available(self) -> bool:
        """Prueft ob cologne_phonetics installiert ist."""
        return _HAS_COLOGNE

    def initialize(self, config: dict, language: str, protected_words: list[str]) -> bool:
        """Baut phonetischen Index aus geschuetzten Woertern auf."""
        if not _HAS_COLOGNE:
            return False

        try:
            self._protected_words = {w.lower() for w in protected_words}

            # Phonetischen Index aufbauen
            for word in protected_words:
                codes = cologne_phonetics.encode(word)
                if codes:
                    # cologne_phonetics.encode gibt Liste von (wort, code) Tupeln
                    for _, code in codes:
                        if code:
                            self._phonetic_index[code] = word.lower()

            pdebug(f"[Phonetic] Index aufgebaut mit {len(self._phonetic_index)} Eintraegen")
            return True

        except Exception as e:
            pwarn(f"[Phonetic] Initialisierung fehlgeschlagen: {e}")
            return False

    def correct(self, text: str) -> str:
        """Ersetzt Woerter die phonetisch einem geschuetzten Wort entsprechen."""
        if not text.strip():
            return text

        try:
            words = text.split()
            corrected_words = []
            changed = False

            for word in words:
                replacement = self._find_phonetic_match(word)
                if replacement and replacement != word.lower():
                    corrected_words.append(replacement)
                    changed = True
                else:
                    corrected_words.append(word)

            if changed:
                result = " ".join(corrected_words)
                pdebug(f"[Phonetic] '{text}' → '{result}'")
                return result

            return text

        except Exception as e:
            pdebug(f"[Phonetic] Fehler bei Korrektur: {e}")
            return text

    def _find_phonetic_match(self, word: str) -> str | None:
        """Sucht phonetische Uebereinstimmung fuer ein Wort."""
        if word.lower() in self._protected_words:
            return word.lower()

        try:
            codes = cologne_phonetics.encode(word)
            if not codes:
                return None

            for _, code in codes:
                if code and code in self._phonetic_index:
                    candidate = self._phonetic_index[code]
                    # Nur ersetzen wenn Edit-Distanz klein genug
                    if self._edit_distance(word.lower(), candidate) <= 3:
                        return candidate

            return None

        except Exception:
            return None

    @staticmethod
    def _edit_distance(s1: str, s2: str) -> int:
        """Berechnet Levenshtein-Distanz zwischen zwei Strings."""
        if len(s1) < len(s2):
            return PhoneticLayer._edit_distance(s2, s1)

        if len(s2) == 0:
            return len(s1)

        prev_row = range(len(s2) + 1)
        for i, c1 in enumerate(s1):
            curr_row = [i + 1]
            for j, c2 in enumerate(s2):
                # Einfuegen, Loeschen, Ersetzen
                insertions = prev_row[j + 1] + 1
                deletions = curr_row[j] + 1
                substitutions = prev_row[j] + (c1 != c2)
                curr_row.append(min(insertions, deletions, substitutions))
            prev_row = curr_row

        return prev_row[-1]