Trixy7/trixy_core/nlp/decorators.py

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

Dekoratoren zum Markieren von Plugin-Methoden als Intent-Handler.

Verwendung:
    class WeatherPlugin(TrixyPlugin):
        @intent("current_weather", description="Aktuelles Wetter abfragen")
        @pattern("(wie ist|wie wird) [denn] {das|} wetter [draussen|gerade]")
        @pattern("wetter [in] {city}")
        @example("Wie ist das Wetter?", "Wie warm ist es?")
        async def handle_weather(self, data, city: str = "") -> IntentResult:
            ...

Slot-Erkennung aus Funktionssignatur:
    - Parameter mit Default-Wert → optionaler Slot
    - Parameter ohne Default-Wert → erforderlicher Slot
    - 'self' und 'data' werden uebersprungen
"""

import inspect
from functools import wraps
from typing import Any, Callable, TypeVar

from trixy_core.nlp.intent_registry import IntentSlot

# Type variable fuer generische Funktionssignaturen
F = TypeVar("F", bound=Callable[..., Any])

# Attributnamen fuer Metadaten
INTENT_METADATA_ATTR = "_trixy_intent_metadata"
_PATTERN_LIST_ATTR = "_trixy_patterns"
_EXAMPLE_LIST_ATTR = "_trixy_examples"

# Parameter die bei der Slot-Erkennung uebersprungen werden
_SKIP_PARAMS = {"self", "data", "return", "kwargs", "args"}


def pattern(pattern_str: str) -> Callable[[F], F]:
    """
    Dekorator zum Hinzufuegen eines Matching-Patterns zu einem Intent-Handler.

    Kann mehrfach angewendet werden (stackbar).

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

    Args:
        pattern_str: Pattern-String in Snips-aehnlicher Syntax

    Beispiel:
        @intent("current_weather")
        @pattern("(wie ist|wie wird) [denn] wetter")
        @pattern("wetter [in] {city}")
        async def handle_weather(self, data, city: str = ""):
            ...
    """
    def decorator(func: F) -> F:
        # Pattern-Liste an der Funktion sammeln (LIFO → umkehren bei Auslesen)
        patterns: list[str] = getattr(func, _PATTERN_LIST_ATTR, [])
        patterns.append(pattern_str)
        setattr(func, _PATTERN_LIST_ATTR, patterns)
        return func

    return decorator


def example(*phrases: str) -> Callable[[F], F]:
    """
    Dekorator zum Hinzufuegen von Beispiel-Phrasen zu einem Intent-Handler.

    Kann mehrfach angewendet werden (stackbar).

    Args:
        *phrases: Beispiel-Saetze fuer den Intent

    Beispiel:
        @intent("current_weather")
        @example("Wie ist das Wetter?", "Wie warm ist es?")
        @example("Temperatur draussen")
        async def handle_weather(self, data):
            ...
    """
    def decorator(func: F) -> F:
        examples: list[str] = getattr(func, _EXAMPLE_LIST_ATTR, [])
        examples.extend(phrases)
        setattr(func, _EXAMPLE_LIST_ATTR, examples)
        return func

    return decorator


def admin_only(func: F) -> F:
    """
    Dekorator der einen Intent als Admin-Only markiert.

    Admin-Intents werden nur ausgefuehrt wenn das Wakeword
    als Admin-Wakeword (system_command) erkannt wurde.

    Beispiel:
        @intent("fritzbox_status", description="Netzwerkstatus")
        @admin_only
        async def handle_status(self, data):
            ...
    """
    setattr(func, "_trixy_admin_only", True)
    return func


def intent(
    name: str,
    *,
    slots: dict[str, type | IntentSlot] | None = None,
    examples: list[str] | None = None,
    patterns: list[str] | None = None,
    description: str = "",
    priority: int = 0,
) -> Callable[[F], F]:
    """
    Dekorator zum Markieren einer Methode als Intent-Handler.

    Der PluginManager entdeckt diese Methoden automatisch und
    registriert sie in der IntentRegistry.

    Slots werden automatisch aus der Funktionssignatur erkannt:
    - Parameter mit Default-Wert → optionaler Slot
    - Parameter ohne Default-Wert → erforderlicher Slot
    - 'self' und 'data' werden uebersprungen

    Explizit uebergebene slots ueberschreiben signatur-basierte.

    Args:
        name: Eindeutiger Intent-Name
        slots: Explizite Slot-Definitionen (ueberschreibt Signatur)
        examples: Beispiel-Phrasen (alternativ: @example Dekorator)
        patterns: Pattern-Strings (alternativ: @pattern Dekorator)
        description: Beschreibung des Intents
        priority: Prioritaet (hoeher = wird bevorzugt)

    Beispiel:
        @intent("current_weather", description="Wetter abfragen")
        @pattern("(wie ist|wie wird) wetter")
        @example("Wie ist das Wetter?")
        async def handle_weather(self, data, city: str = "") -> IntentResult:
            # city ist optionaler Slot (hat Default-Wert)
            ...
    """

    def decorator(func: F) -> F:
        # 1. Slots aus Funktionssignatur erkennen
        sig_slots = _extract_slots_from_signature(func)

        # 2. Explizite Slots normalisieren und mergen (haben Vorrang)
        normalized_slots: dict[str, IntentSlot] = dict(sig_slots)
        if slots:
            for slot_name, slot_value in slots.items():
                if isinstance(slot_value, IntentSlot):
                    normalized_slots[slot_name] = slot_value
                elif isinstance(slot_value, type):
                    normalized_slots[slot_name] = IntentSlot(
                        name=slot_name,
                        slot_type=slot_value,
                    )
                else:
                    normalized_slots[slot_name] = IntentSlot(
                        name=slot_name,
                        slot_type=str,
                    )

        # 3. Patterns aus @pattern Dekoratoren sammeln + explizite
        all_patterns: list[str] = list(patterns or [])
        deco_patterns: list[str] = getattr(func, _PATTERN_LIST_ATTR, [])
        if deco_patterns:
            all_patterns.extend(deco_patterns)

        # 4. Examples aus @example Dekoratoren sammeln + explizite
        all_examples: list[str] = list(examples or [])
        deco_examples: list[str] = getattr(func, _EXAMPLE_LIST_ATTR, [])
        if deco_examples:
            all_examples.extend(deco_examples)

        # 5. Admin-Only Flag von @admin_only Dekorator uebernehmen
        is_admin = getattr(func, "_trixy_admin_only", False)

        # 6. Metadaten an Funktion anhaengen
        metadata = {
            "name": name,
            "slots": normalized_slots,
            "examples": all_examples,
            "patterns": all_patterns,
            "description": description,
            "priority": priority,
            "admin_only": is_admin,
        }

        setattr(func, INTENT_METADATA_ATTR, metadata)

        @wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> Any:
            return await func(*args, **kwargs)

        # Metadaten auch auf Wrapper uebertragen
        setattr(wrapper, INTENT_METADATA_ATTR, metadata)

        return wrapper  # type: ignore

    return decorator


def _extract_slots_from_signature(func: Callable[..., Any]) -> dict[str, IntentSlot]:
    """
    Extrahiert Slot-Definitionen aus der Funktionssignatur.

    Parameter mit Default-Wert → optionaler Slot.
    Parameter ohne Default-Wert → erforderlicher Slot.
    'self', 'data', '*args', '**kwargs' werden uebersprungen.

    Args:
        func: Die zu analysierende Funktion

    Returns:
        Dict von Slot-Name → IntentSlot
    """
    slots: dict[str, IntentSlot] = {}

    try:
        sig = inspect.signature(func)
    except (ValueError, TypeError):
        return slots

    for param_name, param in sig.parameters.items():
        # Ueberspringe self, data, *args, **kwargs
        if param_name in _SKIP_PARAMS:
            continue
        if param.kind in (
            inspect.Parameter.VAR_POSITIONAL,
            inspect.Parameter.VAR_KEYWORD,
        ):
            continue

        # Typ bestimmen
        annotation = param.annotation
        if annotation is inspect.Parameter.empty:
            slot_type = str
        elif isinstance(annotation, type):
            slot_type = annotation
        else:
            slot_type = str

        # Required bestimmen: kein Default = required
        has_default = param.default is not inspect.Parameter.empty
        required = not has_default

        slots[param_name] = IntentSlot(
            name=param_name,
            slot_type=slot_type,
            required=required,
        )

    return slots


def get_intent_metadata(func: Callable[..., Any]) -> dict[str, Any] | None:
    """
    Gibt die Intent-Metadaten einer Funktion zurueck.

    Args:
        func: Funktion zum Pruefen

    Returns:
        Metadaten-Dictionary oder None
    """
    return getattr(func, INTENT_METADATA_ATTR, None)


def is_intent_handler(func: Callable[..., Any]) -> bool:
    """
    Prueft ob eine Funktion ein Intent-Handler ist.

    Args:
        func: Funktion zum Pruefen

    Returns:
        True wenn Intent-Handler
    """
    return hasattr(func, INTENT_METADATA_ATTR)


def discover_intent_handlers(
    obj: object,
) -> list[tuple[str, Callable[..., Any], dict[str, Any]]]:
    """
    Entdeckt alle Intent-Handler in einem Objekt.

    Durchsucht alle Methoden des Objekts nach @intent Dekoratoren.

    Args:
        obj: Objekt zum Durchsuchen (z.B. Plugin-Instanz)

    Returns:
        Liste von (intent_name, handler_func, metadata) Tupeln
    """
    handlers = []

    for attr_name in dir(obj):
        if attr_name.startswith("_"):
            continue

        try:
            attr = getattr(obj, attr_name)
        except Exception:
            continue

        if not callable(attr):
            continue

        metadata = get_intent_metadata(attr)
        if metadata is not None:
            handlers.append((metadata["name"], attr, metadata))

    return handlers