Trixy7/trixy_core/plugins/hooks/hook.py

# -*- coding: utf-8 -*-
"""
Plugin-Hook Implementierung.

Hooks ermöglichen Plugins, sich in definierte Erweiterungspunkte
einzuklinken und Daten zu modifizieren.
"""
from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import IntEnum, auto
from typing import Any, Callable, TypeVar, Generic

T = TypeVar("T")

# Hook-Callback-Typen
HookCallback = Callable[..., Any]
AsyncHookCallback = Callable[..., Any]  # Koroutine


class HookPriority(IntEnum):
    """Priorität für Hook-Callbacks."""
    FIRST = 0       # Wird zuerst ausgeführt
    HIGH = 25
    NORMAL = 50     # Standard
    LOW = 75
    LAST = 100      # Wird zuletzt ausgeführt


@dataclass
class HookResult(Generic[T]):
    """
    Ergebnis eines Hook-Aufrufs.

    Attributes:
        value: Der Ergebniswert
        modified: Wurde der Wert modifiziert
        stopped: Wurde die Verarbeitung gestoppt
        source: Welches Plugin hat modifiziert
        errors: Aufgetretene Fehler
    """
    value: T
    modified: bool = False
    stopped: bool = False
    source: str = ""
    errors: list[str] = field(default_factory=list)

    @property
    def success(self) -> bool:
        """Keine Fehler aufgetreten."""
        return len(self.errors) == 0


@dataclass
class HookHandler:
    """
    Ein registrierter Hook-Handler.

    Attributes:
        callback: Die Callback-Funktion
        plugin_name: Name des registrierenden Plugins
        priority: Ausführungspriorität
        is_async: Ist asynchron
        once: Nur einmal ausführen
        enabled: Ist aktiviert
        call_count: Anzahl Aufrufe
    """
    callback: HookCallback
    plugin_name: str
    priority: HookPriority = HookPriority.NORMAL
    is_async: bool = False
    once: bool = False
    enabled: bool = True
    call_count: int = 0
    created_at: datetime = field(default_factory=datetime.now)

    def __lt__(self, other: HookHandler) -> bool:
        """Sortierung nach Priorität."""
        return self.priority < other.priority


class PluginHook(Generic[T]):
    """
    Ein Plugin-Hook für Erweiterungspunkte.

    Hooks ermöglichen Plugins, sich in definierte Punkte einzuklinken
    und Daten zu verarbeiten oder zu modifizieren.

    Es gibt zwei Hauptmuster:
    1. Filter: Modifiziert einen Wert (z.B. Text-Transformation)
    2. Action: Führt eine Aktion aus (z.B. Logging)

    Example:
        # Hook definieren
        process_text = PluginHook[str]("process_text")

        # Handler registrieren
        @process_text.register("my_plugin")
        def uppercase(text: str) -> str:
            return text.upper()

        @process_text.register("other_plugin", priority=HookPriority.LAST)
        def add_period(text: str) -> str:
            if not text.endswith("."):
                return text + "."
            return text

        # Hook ausführen
        result = process_text.apply("hello world")
        print(result.value)  # "HELLO WORLD."

        # Async-Unterstützung
        @process_text.register_async("async_plugin")
        async def async_handler(text: str) -> str:
            await asyncio.sleep(0.1)
            return text.lower()

        result = await process_text.apply_async("HELLO")
    """

    def __init__(
        self,
        name: str,
        description: str = "",
        default_value: T | None = None,
    ) -> None:
        """
        Initialisiert den Hook.

        Args:
            name: Eindeutiger Hook-Name
            description: Beschreibung
            default_value: Standardwert wenn keine Handler
        """
        self._name = name
        self._description = description
        self._default_value = default_value
        self._handlers: list[HookHandler] = []

    @property
    def name(self) -> str:
        """Hook-Name."""
        return self._name

    @property
    def description(self) -> str:
        """Hook-Beschreibung."""
        return self._description

    @property
    def handler_count(self) -> int:
        """Anzahl registrierter Handler."""
        return len(self._handlers)

    def register(
        self,
        plugin_name: str,
        priority: HookPriority = HookPriority.NORMAL,
        once: bool = False,
    ) -> Callable[[HookCallback], HookCallback]:
        """
        Decorator zum Registrieren eines Handlers.

        Args:
            plugin_name: Name des Plugins
            priority: Ausführungspriorität
            once: Nur einmal ausführen

        Returns:
            Decorator-Funktion
        """
        def decorator(callback: HookCallback) -> HookCallback:
            handler = HookHandler(
                callback=callback,
                plugin_name=plugin_name,
                priority=priority,
                is_async=asyncio.iscoroutinefunction(callback),
                once=once,
            )
            self._handlers.append(handler)
            self._handlers.sort()
            return callback
        return decorator

    def register_async(
        self,
        plugin_name: str,
        priority: HookPriority = HookPriority.NORMAL,
        once: bool = False,
    ) -> Callable[[AsyncHookCallback], AsyncHookCallback]:
        """Decorator für async Handler."""
        return self.register(plugin_name, priority, once)

    def add_handler(
        self,
        callback: HookCallback,
        plugin_name: str,
        priority: HookPriority = HookPriority.NORMAL,
        once: bool = False,
    ) -> None:
        """Fügt einen Handler direkt hinzu."""
        handler = HookHandler(
            callback=callback,
            plugin_name=plugin_name,
            priority=priority,
            is_async=asyncio.iscoroutinefunction(callback),
            once=once,
        )
        self._handlers.append(handler)
        self._handlers.sort()

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

        Args:
            plugin_name: Plugin-Name

        Returns:
            Anzahl entfernter Handler
        """
        before = len(self._handlers)
        self._handlers = [h for h in self._handlers if h.plugin_name != plugin_name]
        return before - len(self._handlers)

    def apply(self, value: T, *args: Any, **kwargs: Any) -> HookResult[T]:
        """
        Wendet den Hook auf einen Wert an.

        Args:
            value: Der zu verarbeitende Wert
            *args: Zusätzliche Argumente
            **kwargs: Zusätzliche Keyword-Argumente

        Returns:
            HookResult mit dem verarbeiteten Wert
        """
        current_value = value
        modified = False
        errors: list[str] = []
        source = ""
        to_remove: list[HookHandler] = []

        for handler in self._handlers:
            if not handler.enabled:
                continue

            try:
                if handler.is_async:
                    # Async-Handler synchron ausführen (wenn möglich)
                    try:
                        loop = asyncio.get_event_loop()
                        result = loop.run_until_complete(
                            handler.callback(current_value, *args, **kwargs)
                        )
                    except RuntimeError:
                        # Kein Event-Loop - überspringen
                        errors.append(
                            f"{handler.plugin_name}: Async-Handler ohne Event-Loop"
                        )
                        continue
                else:
                    result = handler.callback(current_value, *args, **kwargs)

                handler.call_count += 1

                if result is not None:
                    current_value = result
                    modified = True
                    source = handler.plugin_name

                if handler.once:
                    to_remove.append(handler)

            except Exception as e:
                errors.append(f"{handler.plugin_name}: {str(e)}")

        # Once-Handler entfernen
        for handler in to_remove:
            self._handlers.remove(handler)

        return HookResult(
            value=current_value,
            modified=modified,
            source=source,
            errors=errors,
        )

    async def apply_async(
        self,
        value: T,
        *args: Any,
        **kwargs: Any,
    ) -> HookResult[T]:
        """
        Wendet den Hook asynchron an.

        Args:
            value: Der zu verarbeitende Wert
            *args: Zusätzliche Argumente
            **kwargs: Zusätzliche Keyword-Argumente

        Returns:
            HookResult mit dem verarbeiteten Wert
        """
        current_value = value
        modified = False
        errors: list[str] = []
        source = ""
        to_remove: list[HookHandler] = []

        for handler in self._handlers:
            if not handler.enabled:
                continue

            try:
                if handler.is_async:
                    result = await handler.callback(current_value, *args, **kwargs)
                else:
                    result = handler.callback(current_value, *args, **kwargs)

                handler.call_count += 1

                if result is not None:
                    current_value = result
                    modified = True
                    source = handler.plugin_name

                if handler.once:
                    to_remove.append(handler)

            except Exception as e:
                errors.append(f"{handler.plugin_name}: {str(e)}")

        # Once-Handler entfernen
        for handler in to_remove:
            self._handlers.remove(handler)

        return HookResult(
            value=current_value,
            modified=modified,
            source=source,
            errors=errors,
        )

    def trigger(self, *args: Any, **kwargs: Any) -> list[Any]:
        """
        Triggert den Hook als Action (ohne Rückgabewert-Verkettung).

        Args:
            *args: Argumente für Handler
            **kwargs: Keyword-Argumente

        Returns:
            Liste der Rückgabewerte
        """
        results = []

        for handler in self._handlers:
            if not handler.enabled:
                continue

            try:
                if handler.is_async:
                    continue  # Async-Handler überspringen

                result = handler.callback(*args, **kwargs)
                handler.call_count += 1
                results.append(result)

                if handler.once:
                    self._handlers.remove(handler)

            except Exception:
                pass

        return results

    async def trigger_async(self, *args: Any, **kwargs: Any) -> list[Any]:
        """Triggert den Hook asynchron."""
        results = []

        for handler in list(self._handlers):
            if not handler.enabled:
                continue

            try:
                if handler.is_async:
                    result = await handler.callback(*args, **kwargs)
                else:
                    result = handler.callback(*args, **kwargs)

                handler.call_count += 1
                results.append(result)

                if handler.once:
                    self._handlers.remove(handler)

            except Exception:
                pass

        return results

    def clear(self) -> int:
        """
        Entfernt alle Handler.

        Returns:
            Anzahl entfernter Handler
        """
        count = len(self._handlers)
        self._handlers.clear()
        return count

    def get_handlers(self) -> list[HookHandler]:
        """Gibt alle Handler zurück."""
        return self._handlers.copy()

    def __repr__(self) -> str:
        return f"PluginHook({self._name!r}, handlers={len(self._handlers)})"