Trixy7/trixy_core/nlp/keyword_matcher.py

# -*- coding: utf-8 -*-
"""
Keyword Intent Matcher — Snips-aehnlicher Pre-Filter vor dem LLM.

Erkennt eindeutige Befehle in <5ms ueber Pattern-Matching und
Fuzzy-Token-Vergleich. Nur bei niedrigem Confidence-Score wird
das LLM als Fallback aufgerufen.

Pattern-Syntax:
    wort          — Pflicht-Keyword
    [wort]        — Optionales Wort (0 oder 1)
    [a|b|c]       — Optionale Alternativen
    (a|b)         — Pflicht-Alternativen (genau 1)
    {slot_name}   — Slot-Extraktion (Rest-Wort)
    {a|b|}        — Inline-Alternativen (letztes leer = optional)
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from typing import Any

from trixy_core.utils.debug import pdebug

# Optionaler Import — Entity-System ist nicht zwingend erforderlich
try:
    from trixy_core.nlp.entities import EntityRegistry, EntityResolver, ResolvedEntity
    _HAS_ENTITIES = True
except ImportError:
    _HAS_ENTITIES = False


# ============================================================================
# Datenklassen
# ============================================================================

@dataclass
class PatternToken:
    """Ein kompiliertes Pattern-Element."""
    kind: str           # "literal", "optional", "alternatives", "slot"
    values: list[str] = field(default_factory=list)
    slot_name: str = ""
    entity_type: str = ""   # Entity-Typ fuer Validierung (z.B. "datum", "stadt")
    required: bool = True


@dataclass
class CompiledPattern:
    """Ein kompiliertes Pattern mit Quell-String."""
    tokens: list[PatternToken] = field(default_factory=list)
    source: str = ""


@dataclass
class MatchResult:
    """Ergebnis eines erfolgreichen Matches."""
    intent: str = ""
    confidence: float = 0.0
    slots: dict[str, Any] = field(default_factory=dict)
    sentiment: dict[str, Any] = field(default_factory=dict)
    matched_by: str = ""   # "pattern" oder "example"


# ============================================================================
# Sentiment-Woerter
# ============================================================================

POLITE_WORDS: set[str] = {
    "bitte", "danke", "freundlicherweise", "koenntest", "wuerdest",
    "koennten", "wuerden", "gerne", "bitteschoen",
}

RUDE_WORDS: set[str] = {
    "verdammt", "scheisse", "mist", "verflucht", "bloed", "dumm",
    "idiot", "doof", "kacke", "scheiße",
}

URGENT_WORDS: set[str] = {
    "sofort", "schnell", "jetzt", "dringend", "eilig", "hurtig",
}

# Woerter die beim Matching ignoriert werden (Fuellwoerter)
_FILLER_WORDS: set[str] = {
    "mal", "mir", "du", "kannst", "koenntest", "wuerdest",
    "eigentlich", "doch", "auch", "noch",
}

# Pattern-Syntax: Regex zum Erkennen der Elemente
_PATTERN_RE = re.compile(
    r"""
    \(  ([^)]+)  \)   |   # (a|b) — Pflicht-Alternativen
    \[  ([^\]]+) \]   |   # [a|b] — Optionale Alternativen / optionales Wort
    \{  ([^}]+)  \}   |   # {slot} oder {a|b|} — Slot / Inline-Alternativen
    (\S+)                 # Pflicht-Wort
    """,
    re.VERBOSE,
)


class KeywordIntentMatcher:
    """
    Snips-aehnlicher Keyword-Matcher als LLM-Pre-Filter.

    Kompiliert Intent-Patterns einmal beim Laden und matcht
    eingehenden Text in <5ms gegen alle Patterns.
    """

    def __init__(self, config: dict[str, Any] | None = None) -> None:
        config = config or {}
        self._confidence_threshold: float = config.get("confidence_threshold", 0.82)
        self._fuzzy_threshold: float = config.get("fuzzy_threshold", 0.8)
        self._compiled: dict[str, list[CompiledPattern]] = {}
        self._examples: dict[str, list[list[str]]] = {}  # intent → tokenized examples
        self._entity_registry: Any | None = None  # EntityRegistry (optional)
        self._entity_resolver: Any | None = None  # EntityResolver (optional)

    # ========================================================================
    # Kompilierung
    # ========================================================================

    def set_entity_registry(self, registry: Any) -> None:
        """
        Setzt die EntityRegistry fuer Entity-Typ-Validierung.

        Args:
            registry: EntityRegistry-Instanz
        """
        self._entity_registry = registry
        if _HAS_ENTITIES and registry is not None:
            self._entity_resolver = EntityResolver(registry)
        else:
            self._entity_resolver = None

    def compile_intents(self, intents: list[dict[str, Any]]) -> None:
        """
        Kompiliert Intent-Patterns und Examples aus den available_intents.

        Args:
            intents: Liste von Intent-Dicts (aus Registry.get_all_as_dict())
        """
        self._compiled.clear()
        self._examples.clear()

        for intent_dict in intents:
            name = intent_dict.get("name", "")
            if not name:
                continue

            # Patterns kompilieren
            patterns = intent_dict.get("patterns", [])
            if patterns:
                compiled = []
                for pattern_str in patterns:
                    parsed = self._parse_pattern(pattern_str)
                    if parsed:
                        compiled.append(CompiledPattern(tokens=parsed, source=pattern_str))
                if compiled:
                    self._compiled[name] = compiled

            # Examples tokenisieren
            examples = intent_dict.get("examples", [])
            if examples:
                self._examples[name] = [
                    self._tokenize(ex) for ex in examples
                ]

        total_patterns = sum(len(v) for v in self._compiled.values())
        total_examples = sum(len(v) for v in self._examples.values())
        pdebug(
            f"[KeywordMatcher] Kompiliert: {len(self._compiled)} Intents "
            f"mit {total_patterns} Patterns, {total_examples} Examples"
        )

    # ========================================================================
    # Matching
    # ========================================================================

    def match(self, text: str) -> MatchResult | None:
        """
        Matcht Text gegen alle kompilierten Patterns und Examples.

        Returns:
            MatchResult wenn Confidence >= Schwelle, sonst None
        """
        if not text or not text.strip():
            return None

        # Sentiment extrahieren und Tokens bereinigen
        all_tokens = self._tokenize(text)
        sentiment = self._extract_sentiment(all_tokens)
        clean_tokens = self._remove_sentiment_words(all_tokens)

        best: MatchResult | None = None

        # 1. Pattern-Match (hoehere Praezision)
        for intent_name, patterns in self._compiled.items():
            for pattern in patterns:
                result = self._match_pattern(clean_tokens, pattern)
                if result is not None:
                    confidence = result["confidence"]
                    if confidence >= self._confidence_threshold:
                        if best is None or confidence > best.confidence:
                            best = MatchResult(
                                intent=intent_name,
                                confidence=confidence,
                                slots=result.get("slots", {}),
                                sentiment=sentiment,
                                matched_by="pattern",
                            )

        # 2. Example-Fuzzy-Match (Fallback)
        for intent_name, example_lists in self._examples.items():
            for ex_tokens in example_lists:
                score = self._score_example_match(clean_tokens, ex_tokens)
                if score >= self._confidence_threshold:
                    if best is None or score > best.confidence:
                        best = MatchResult(
                            intent=intent_name,
                            confidence=score,
                            slots={},
                            sentiment=sentiment,
                            matched_by="example",
                        )

        if best is not None:
            best.sentiment = sentiment

        return best

    # ========================================================================
    # Tokenisierung
    # ========================================================================

    def _tokenize(self, text: str) -> list[str]:
        """Tokenisiert Text in Kleinbuchstaben-Woerter."""
        # Interpunktion entfernen, lowercase, splitten
        cleaned = re.sub(r"[^\w\süöäß]", "", text.lower())
        return cleaned.split()

    # ========================================================================
    # Pattern-Parsing
    # ========================================================================

    def _parse_pattern(self, pattern: str) -> list[PatternToken]:
        """
        Parst einen Pattern-String in eine Liste von PatternTokens.

        Beispiel:
            "(wie ist|wie wird) [denn] wetter [in] {city}"
            →  [alternatives(required), optional, literal, optional, slot]
        """
        tokens: list[PatternToken] = []

        for m in _PATTERN_RE.finditer(pattern):
            alternatives_group = m.group(1)   # (a|b)
            optional_group = m.group(2)        # [a|b]
            slot_group = m.group(3)            # {slot} oder {a|b|}
            literal_group = m.group(4)         # wort

            if alternatives_group is not None:
                # (a|b) — Pflicht-Alternativen (koennen Mehrwort sein)
                values = [v.strip().lower() for v in alternatives_group.split("|")]
                tokens.append(PatternToken(
                    kind="alternatives",
                    values=values,
                    required=True,
                ))

            elif optional_group is not None:
                # [wort] oder [a|b|c]
                if "|" in optional_group:
                    values = [v.strip().lower() for v in optional_group.split("|")]
                else:
                    values = [optional_group.strip().lower()]
                tokens.append(PatternToken(
                    kind="optional",
                    values=values,
                    required=False,
                ))

            elif slot_group is not None:
                # {slot_name} oder {slot_name:entity_type} oder {a|b|} (Inline-Alternativen)
                if "|" in slot_group:
                    # Inline-Alternativen: {das|} → letztes leer = optional
                    values = [v.strip().lower() for v in slot_group.split("|")]
                    has_empty = "" in values
                    values = [v for v in values if v]  # leere entfernen
                    tokens.append(PatternToken(
                        kind="optional" if has_empty else "alternatives",
                        values=values,
                        required=not has_empty,
                    ))
                elif ":" in slot_group:
                    # Slot mit Entity-Typ: {date:datum}
                    slot_name, entity_type = slot_group.split(":", 1)
                    tokens.append(PatternToken(
                        kind="slot",
                        slot_name=slot_name.strip(),
                        entity_type=entity_type.strip(),
                        required=False,
                    ))
                else:
                    # Slot-Extraktion ohne Entity-Typ
                    tokens.append(PatternToken(
                        kind="slot",
                        slot_name=slot_group.strip(),
                        required=False,
                    ))

            elif literal_group is not None:
                # Pflicht-Keyword
                tokens.append(PatternToken(
                    kind="literal",
                    values=[literal_group.strip().lower()],
                    required=True,
                ))

        return tokens

    # ========================================================================
    # Pattern-Matching
    # ========================================================================

    def _match_pattern(
        self, tokens: list[str], pattern: CompiledPattern
    ) -> dict[str, Any] | None:
        """
        Matcht eine Token-Liste gegen ein kompiliertes Pattern.

        Returns:
            Dict mit "confidence" und "slots", oder None bei Nicht-Match
        """
        input_tokens = list(tokens)  # Kopie
        pos = 0
        matched_count = 0
        total_required = 0
        slots: dict[str, Any] = {}

        for pt in pattern.tokens:
            if pt.kind == "literal":
                total_required += 1
                if pos < len(input_tokens):
                    sim = self._fuzzy_compare(input_tokens[pos], pt.values[0])
                    if sim >= self._fuzzy_threshold:
                        matched_count += 1
                        pos += 1
                    else:
                        return None  # Pflicht-Keyword fehlt
                else:
                    return None

            elif pt.kind == "alternatives":
                if pt.required:
                    total_required += 1

                found = False
                for value in pt.values:
                    # Mehrwort-Alternativen (z.B. "wie ist")
                    alt_tokens = value.split()
                    if pos + len(alt_tokens) <= len(input_tokens):
                        all_match = True
                        for i, alt_tok in enumerate(alt_tokens):
                            sim = self._fuzzy_compare(input_tokens[pos + i], alt_tok)
                            if sim < self._fuzzy_threshold:
                                all_match = False
                                break
                        if all_match:
                            matched_count += 1
                            pos += len(alt_tokens)
                            found = True
                            break

                if not found:
                    if pt.required:
                        return None

            elif pt.kind == "optional":
                # Optional: matchen falls vorhanden, sonst ueberspringen
                if pos < len(input_tokens):
                    for value in pt.values:
                        alt_tokens = value.split()
                        if pos + len(alt_tokens) <= len(input_tokens):
                            all_match = True
                            for i, alt_tok in enumerate(alt_tokens):
                                sim = self._fuzzy_compare(
                                    input_tokens[pos + i], alt_tok
                                )
                                if sim < self._fuzzy_threshold:
                                    all_match = False
                                    break
                            if all_match:
                                matched_count += 1
                                pos += len(alt_tokens)
                                break

            elif pt.kind == "slot":
                # Slot: Restliche Tokens bis zum naechsten Pattern-Element aufnehmen
                # Oder ein Token wenn noch Pattern-Elemente folgen
                after_this = pattern.tokens[pattern.tokens.index(pt) + 1:]
                remaining_pattern = [
                    p for p in after_this
                    if p.required or p.kind == "slot"
                ]

                has_entity = bool(pt.entity_type and self._entity_registry)

                if remaining_pattern:
                    if pos < len(input_tokens):
                        # Bei Entity-Typ: Mehrwort-Match versuchen
                        if has_entity:
                            mw_result = self._entity_registry.match_multi_word(
                                input_tokens, pos, pt.entity_type
                            )
                            if mw_result is not None:
                                mw_value, new_pos = mw_result
                                slot_val = self._resolve_slot(mw_value, pt.entity_type)
                                slots[pt.slot_name] = slot_val
                                matched_count += 1
                                pos = new_pos
                            elif self._entity_registry.matches(input_tokens[pos], pt.entity_type):
                                slot_val = self._resolve_slot(input_tokens[pos], pt.entity_type)
                                slots[pt.slot_name] = slot_val
                                matched_count += 1
                                pos += 1
                            else:
                                # Strikte Entity: Wert passt nicht → kein Match
                                entity_def = self._entity_registry.get(pt.entity_type)
                                if entity_def is not None and not entity_def.open:
                                    return None
                                # Open Entity: Wert akzeptieren
                                slot_val = self._resolve_slot(input_tokens[pos], pt.entity_type)
                                slots[pt.slot_name] = slot_val
                                matched_count += 1
                                pos += 1
                        else:
                            # Kein Entity-Typ: wie bisher
                            slots[pt.slot_name] = input_tokens[pos]
                            matched_count += 1
                            pos += 1
                else:
                    # Rest als Slot
                    if pos < len(input_tokens):
                        slot_value = " ".join(input_tokens[pos:])
                        if has_entity:
                            # Mehrwort-Match ab pos versuchen
                            mw_result = self._entity_registry.match_multi_word(
                                input_tokens, pos, pt.entity_type
                            )
                            if mw_result is not None:
                                mw_value, new_pos = mw_result
                                slot_val = self._resolve_slot(mw_value, pt.entity_type)
                                slots[pt.slot_name] = slot_val
                                matched_count += 1
                                pos = new_pos
                            elif self._entity_registry.matches(slot_value, pt.entity_type):
                                slot_val = self._resolve_slot(slot_value, pt.entity_type)
                                slots[pt.slot_name] = slot_val
                                matched_count += 1
                                pos = len(input_tokens)
                            else:
                                entity_def = self._entity_registry.get(pt.entity_type)
                                if entity_def is not None and not entity_def.open:
                                    return None
                                slot_val = self._resolve_slot(slot_value, pt.entity_type)
                                slots[pt.slot_name] = slot_val
                                matched_count += 1
                                pos = len(input_tokens)
                        else:
                            slots[pt.slot_name] = slot_value
                            matched_count += 1
                            pos = len(input_tokens)

        # Confidence berechnen
        if total_required == 0:
            return None

        # Basis-Confidence: Anteil gematchter Tokens
        token_coverage = pos / max(len(input_tokens), 1)
        required_coverage = min(matched_count / max(total_required, 1), 1.0)

        # Gewichtete Confidence
        confidence = 0.6 * required_coverage + 0.4 * token_coverage

        # Bonus fuer vollstaendigen Match (alle Tokens konsumiert)
        if pos >= len(input_tokens):
            confidence = min(confidence + 0.05, 0.98)

        # Penalty fuer unkonsumierte Tokens
        unconsumed = len(input_tokens) - pos
        if unconsumed >= 2:
            confidence *= 0.7
        elif unconsumed == 1:
            confidence *= 0.85

        # Bonus fuer Slots die gegen strikte (nicht-offene) Entity-Typen
        # gematcht wurden — bevorzugt spezifischere Matches
        if self._entity_registry is not None:
            for pt in pattern.tokens:
                if pt.kind == "slot" and pt.entity_type and pt.slot_name in slots:
                    entity_def = self._entity_registry.get(pt.entity_type)
                    if entity_def is not None and not entity_def.open:
                        confidence += 0.03

        if confidence < self._confidence_threshold:
            return None

        return {"confidence": round(confidence, 3), "slots": slots}

    # ========================================================================
    # Fuzzy-Vergleich
    # ========================================================================

    def _fuzzy_compare(self, a: str, b: str) -> float:
        """
        Vergleicht zwei Tokens mit SequenceMatcher.

        Exakte Treffer werden direkt zurueckgegeben (Performance).
        """
        if a == b:
            return 1.0
        return SequenceMatcher(None, a, b).ratio()

    # ========================================================================
    # Example-Matching
    # ========================================================================

    def _score_example_match(
        self, tokens: list[str], example_tokens: list[str]
    ) -> float:
        """
        Bewertet die Aehnlichkeit zwischen Input-Tokens und Example-Tokens.

        Nutzt Fuzzy-Token-Matching fuer STT-Fehlertoleranz.
        """
        if not tokens or not example_tokens:
            return 0.0

        # Exakter Match (Sonderfall fuer Einwort-Befehle)
        if tokens == example_tokens:
            return 0.98

        # Token-fuer-Token Fuzzy-Matching
        matched = 0
        used_indices: set[int] = set()

        for input_tok in tokens:
            # Besten Match in den Example-Tokens finden
            best_score = 0.0
            best_idx = -1
            for idx, ex_tok in enumerate(example_tokens):
                if idx in used_indices:
                    continue
                score = self._fuzzy_compare(input_tok, ex_tok)
                if score > best_score:
                    best_score = score
                    best_idx = idx

            if best_score >= self._fuzzy_threshold and best_idx >= 0:
                matched += 1
                used_indices.add(best_idx)

        # Score: Anteil gematchter Tokens am laengeren der beiden
        max_len = max(len(tokens), len(example_tokens))
        score = matched / max_len

        # Bonus fuer gleiche Laenge (exakterer Match)
        if len(tokens) == len(example_tokens):
            score = min(score + 0.03, 0.98)

        return round(score, 3)

    # ========================================================================
    # Sentiment-Analyse
    # ========================================================================

    def _extract_sentiment(self, tokens: list[str]) -> dict[str, Any]:
        """
        Extrahiert Sentiment-Informationen aus den Tokens.

        Returns:
            Dict mit tone, is_urgent, markers
        """
        markers: list[str] = []
        tone = "neutral"
        is_urgent = False

        for tok in tokens:
            if tok in POLITE_WORDS:
                markers.append(tok)
                if tone != "rude":
                    tone = "polite"
            elif tok in RUDE_WORDS:
                markers.append(tok)
                tone = "rude"
            elif tok in URGENT_WORDS:
                markers.append(tok)
                is_urgent = True

        return {
            "tone": tone,
            "is_urgent": is_urgent,
            "markers": markers,
        }

    def _remove_sentiment_words(self, tokens: list[str]) -> list[str]:
        """Entfernt Sentiment- und Fuellwoerter aus den Tokens fuer saubereres Matching."""
        all_sentiment = POLITE_WORDS | RUDE_WORDS | URGENT_WORDS | _FILLER_WORDS
        cleaned = [t for t in tokens if t not in all_sentiment]
        # Falls alles entfernt wurde, Original zurueckgeben
        return cleaned if cleaned else tokens

    def _resolve_slot(self, raw_value: str, entity_type: str) -> Any:
        """Loest einen Slot-Wert ueber den EntityResolver auf, falls verfuegbar."""
        if self._entity_resolver is not None:
            return self._entity_resolver.resolve(raw_value, entity_type)
        return raw_value