Trixy7/trixy_core/nlp/intent_registry.py

# -*- coding: utf-8 -*-
"""
Intent Registry.

Globale Registry für alle verfügbaren Intents im System.
Wird vom NLP-Provider verwendet, um verfügbare Intents zu ermitteln.
"""

from dataclasses import dataclass, field
from typing import Any, Callable, TYPE_CHECKING
import threading

if TYPE_CHECKING:
    from trixy_core.plugins.trixy_plugin import TrixyPlugin


@dataclass
class IntentSlot:
    """
    Definition eines Intent-Slots.

    Attributes:
        name: Slot-Name
        slot_type: Datentyp des Slots
        required: Ist der Slot erforderlich?
        description: Beschreibung des Slots
        examples: Beispielwerte
    """
    name: str
    slot_type: type = str
    required: bool = False
    description: str = ""
    examples: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert den Slot in ein Dictionary."""
        return {
            "name": self.name,
            "type": self.slot_type.__name__,
            "required": self.required,
            "description": self.description,
            "examples": self.examples,
        }


@dataclass
class IntentDefinition:
    """
    Definition eines Intents.

    Attributes:
        name: Eindeutiger Intent-Name
        handler: Handler-Funktion
        plugin_name: Name des registrierenden Plugins
        slots: Slot-Definitionen
        examples: Beispiel-Phrasen für den Intent
        description: Beschreibung des Intents
        priority: Priorität (höher = wird bevorzugt bei Konflikten)
        enabled: Ist der Intent aktiv?
    """
    name: str
    handler: Callable[..., Any]
    plugin_name: str = ""
    slots: dict[str, IntentSlot] = field(default_factory=dict)
    examples: list[str] = field(default_factory=list)
    description: str = ""
    priority: int = 0
    enabled: bool = True
    patterns: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """
        Konvertiert den Intent in ein Dictionary für NLP-Provider.

        Returns:
            Dictionary mit Intent-Informationen
        """
        result = {
            "name": self.name,
            "description": self.description,
            "slots": {name: slot.to_dict() for name, slot in self.slots.items()},
            "examples": self.examples,
            "plugin": self.plugin_name,
        }
        if self.patterns:
            result["patterns"] = self.patterns
        return result


class IntentRegistry:
    """
    Singleton Registry für alle Intents im System.

    Plugins registrieren ihre Intents hier, und NLP-Provider
    fragen die Registry ab, um verfügbare Intents zu ermitteln.

    Verwendung:
        # Intent registrieren
        registry = IntentRegistry.get_instance()
        registry.register(IntentDefinition(
            name="turn_on_light",
            handler=my_handler,
            plugin_name="smarthome",
            slots={"device": IntentSlot("device", str, required=True)},
            examples=["Licht an", "Lampe einschalten"]
        ))

        # Intents für NLP abfragen
        intents = registry.get_all_as_dict()

        # Handler aufrufen
        handler = registry.get_handler("turn_on_light")
        if handler:
            await handler(intent_data)
    """

    _instance: "IntentRegistry | None" = None
    _lock = threading.Lock()

    def __new__(cls) -> "IntentRegistry":
        """Singleton-Pattern."""
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._initialized = False
        return cls._instance

    def __init__(self) -> None:
        """Initialisiert die Registry."""
        if self._initialized:
            return

        self._intents: dict[str, IntentDefinition] = {}
        self._plugin_intents: dict[str, list[str]] = {}  # Plugin -> Intent-Namen
        self._initialized = True

    @classmethod
    def get_instance(cls) -> "IntentRegistry":
        """Gibt die Singleton-Instanz zurück."""
        return cls()

    def register(self, intent: IntentDefinition) -> bool:
        """
        Registriert einen Intent.

        Args:
            intent: Intent-Definition

        Returns:
            True bei Erfolg, False wenn Intent bereits existiert
        """
        if intent.name in self._intents:
            existing = self._intents[intent.name]
            # Bei gleichem Plugin ersetzen
            if existing.plugin_name == intent.plugin_name:
                self._intents[intent.name] = intent
                return True
            # Bei höherer Priorität ersetzen
            if intent.priority > existing.priority:
                self._intents[intent.name] = intent
                self._update_plugin_intents(existing.plugin_name, intent.plugin_name, intent.name)
                return True
            return False

        self._intents[intent.name] = intent

        # Plugin-Zuordnung aktualisieren
        if intent.plugin_name:
            if intent.plugin_name not in self._plugin_intents:
                self._plugin_intents[intent.plugin_name] = []
            self._plugin_intents[intent.plugin_name].append(intent.name)

        return True

    def _update_plugin_intents(
        self, old_plugin: str, new_plugin: str, intent_name: str
    ) -> None:
        """Aktualisiert die Plugin-Intent-Zuordnung."""
        if old_plugin and old_plugin in self._plugin_intents:
            if intent_name in self._plugin_intents[old_plugin]:
                self._plugin_intents[old_plugin].remove(intent_name)

        if new_plugin:
            if new_plugin not in self._plugin_intents:
                self._plugin_intents[new_plugin] = []
            if intent_name not in self._plugin_intents[new_plugin]:
                self._plugin_intents[new_plugin].append(intent_name)

    def unregister(self, intent_name: str) -> bool:
        """
        Entfernt einen Intent aus der Registry.

        Args:
            intent_name: Name des Intents

        Returns:
            True bei Erfolg
        """
        if intent_name not in self._intents:
            return False

        intent = self._intents.pop(intent_name)

        # Plugin-Zuordnung entfernen
        if intent.plugin_name and intent.plugin_name in self._plugin_intents:
            if intent_name in self._plugin_intents[intent.plugin_name]:
                self._plugin_intents[intent.plugin_name].remove(intent_name)

        return True

    def unregister_plugin(self, plugin_name: str) -> int:
        """
        Entfernt alle Intents eines Plugins.

        Args:
            plugin_name: Plugin-Name

        Returns:
            Anzahl entfernter Intents
        """
        if plugin_name not in self._plugin_intents:
            return 0

        intent_names = self._plugin_intents.pop(plugin_name, []).copy()
        count = 0

        for name in intent_names:
            if name in self._intents:
                del self._intents[name]
                count += 1

        return count

    def get(self, intent_name: str) -> IntentDefinition | None:
        """
        Gibt eine Intent-Definition zurück.

        Args:
            intent_name: Name des Intents

        Returns:
            IntentDefinition oder None
        """
        return self._intents.get(intent_name)

    def resolve_intent(self, intent_name: str) -> IntentDefinition | None:
        """
        Loest einen Intent-Namen auf (exakt oder Suffix-Match).

        Sucht zuerst nach exaktem Match. Falls nicht gefunden,
        wird ein Suffix-Match versucht (z.B. "weather" findet
        "current_weather"). Nur bei genau einem Treffer wird
        der Suffix-Match akzeptiert.

        Args:
            intent_name: Name des Intents (ggf. verkuerzt vom LLM)

        Returns:
            IntentDefinition oder None
        """
        # Exakter Match
        intent = self._intents.get(intent_name)
        if intent and intent.enabled:
            return intent

        # Suffix-Match: LLM gibt manchmal verkuerzte Namen zurueck
        candidates = [
            i for i in self._intents.values()
            if i.enabled and i.name.endswith(f"_{intent_name}")
        ]
        if len(candidates) == 1:
            return candidates[0]

        return None

    def get_handler(self, intent_name: str) -> Callable[..., Any] | None:
        """
        Gibt den Handler für einen Intent zurück.

        Nutzt resolve_intent() fuer exakten und Suffix-Match.

        Args:
            intent_name: Name des Intents

        Returns:
            Handler-Funktion oder None
        """
        intent = self.resolve_intent(intent_name)
        if intent:
            return intent.handler
        return None

    def get_all(self) -> list[IntentDefinition]:
        """
        Gibt alle registrierten Intents zurück.

        Returns:
            Liste aller IntentDefinitions
        """
        return list(self._intents.values())

    def get_enabled(self) -> list[IntentDefinition]:
        """
        Gibt alle aktivierten Intents zurück.

        Returns:
            Liste aktivierter IntentDefinitions
        """
        return [i for i in self._intents.values() if i.enabled]

    def get_all_as_dict(self) -> list[dict[str, Any]]:
        """
        Gibt alle Intents als Dictionary-Liste zurück.

        Wird vom NLP-Provider verwendet, um verfügbare Intents
        in den System-Prompt aufzunehmen.

        Returns:
            Liste von Intent-Dictionaries
        """
        return [i.to_dict() for i in self._intents.values() if i.enabled]

    def get_plugin_intents(self, plugin_name: str) -> list[IntentDefinition]:
        """
        Gibt alle Intents eines Plugins zurück.

        Args:
            plugin_name: Plugin-Name

        Returns:
            Liste der Plugin-Intents
        """
        intent_names = self._plugin_intents.get(plugin_name, [])
        return [self._intents[name] for name in intent_names if name in self._intents]

    def has_intent(self, intent_name: str) -> bool:
        """Prüft ob ein Intent registriert ist."""
        return intent_name in self._intents

    def enable(self, intent_name: str) -> bool:
        """
        Aktiviert einen Intent.

        Args:
            intent_name: Name des Intents

        Returns:
            True bei Erfolg
        """
        if intent_name in self._intents:
            self._intents[intent_name].enabled = True
            return True
        return False

    def disable(self, intent_name: str) -> bool:
        """
        Deaktiviert einen Intent.

        Args:
            intent_name: Name des Intents

        Returns:
            True bei Erfolg
        """
        if intent_name in self._intents:
            self._intents[intent_name].enabled = False
            return True
        return False

    def clear(self) -> None:
        """Löscht alle registrierten Intents."""
        self._intents.clear()
        self._plugin_intents.clear()

    def count(self) -> int:
        """Anzahl registrierter Intents."""
        return len(self._intents)

    def __len__(self) -> int:
        """Anzahl registrierter Intents."""
        return len(self._intents)

    def __contains__(self, intent_name: str) -> bool:
        """Prüft ob ein Intent registriert ist."""
        return intent_name in self._intents

    def __iter__(self):
        """Iterator über alle Intent-Definitionen."""
        return iter(self._intents.values())