Trixy7/trixy_core/plugins/hooks/registry.py

# -*- coding: utf-8 -*-
"""
Hook-Registry für zentrale Hook-Verwaltung.

Ermöglicht das Registrieren, Finden und Verwalten von Hooks
über alle Plugins hinweg.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, TypeVar

from trixy_core.plugins.hooks.hook import (
    PluginHook,
    HookPriority,
    HookCallback,
    HookHandler,
)

T = TypeVar("T")


@dataclass
class HookInfo:
    """
    Informationen über einen registrierten Hook.

    Attributes:
        name: Hook-Name
        description: Beschreibung
        handler_count: Anzahl Handler
        total_calls: Gesamtaufrufe
        handlers: Handler-Details
    """
    name: str
    description: str
    handler_count: int
    total_calls: int
    handlers: list[dict[str, Any]] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "name": self.name,
            "description": self.description,
            "handler_count": self.handler_count,
            "total_calls": self.total_calls,
            "handlers": self.handlers,
        }


class HookRegistry:
    """
    Zentrale Registry für alle Plugin-Hooks.

    Verwaltet Hooks und ermöglicht deren dynamische
    Erstellung und Abfrage.

    Example:
        registry = HookRegistry()

        # Hook erstellen oder abrufen
        process_hook = registry.get_or_create("process_text")

        # Handler registrieren
        registry.register_handler(
            "process_text",
            "my_plugin",
            lambda text: text.upper()
        )

        # Hook anwenden
        result = registry.apply("process_text", "hello")
        print(result.value)  # "HELLO"

        # Alle Hooks eines Plugins entfernen
        registry.remove_plugin("my_plugin")

        # Statistiken
        for info in registry.get_all_info():
            print(f"{info.name}: {info.handler_count} handlers")
    """

    def __init__(self) -> None:
        """Initialisiert die Registry."""
        self._hooks: dict[str, PluginHook] = {}
        self._plugin_hooks: dict[str, set[str]] = {}  # plugin_name -> set(hook_names)

    @property
    def hook_count(self) -> int:
        """Anzahl registrierter Hooks."""
        return len(self._hooks)

    @property
    def total_handlers(self) -> int:
        """Gesamtanzahl aller Handler."""
        return sum(hook.handler_count for hook in self._hooks.values())

    def create(
        self,
        name: str,
        description: str = "",
        default_value: Any = None,
    ) -> PluginHook:
        """
        Erstellt einen neuen Hook.

        Args:
            name: Eindeutiger Hook-Name
            description: Beschreibung
            default_value: Standardwert

        Returns:
            Der erstellte Hook

        Raises:
            ValueError: Wenn Hook bereits existiert
        """
        if name in self._hooks:
            raise ValueError(f"Hook existiert bereits: {name}")

        hook: PluginHook = PluginHook(name, description, default_value)
        self._hooks[name] = hook
        return hook

    def get(self, name: str) -> PluginHook | None:
        """Gibt einen Hook zurück."""
        return self._hooks.get(name)

    def get_or_create(
        self,
        name: str,
        description: str = "",
        default_value: Any = None,
    ) -> PluginHook:
        """
        Gibt einen Hook zurück oder erstellt ihn.

        Args:
            name: Hook-Name
            description: Beschreibung (nur bei Erstellung)
            default_value: Standardwert (nur bei Erstellung)

        Returns:
            Der Hook
        """
        if name not in self._hooks:
            self._hooks[name] = PluginHook(name, description, default_value)
        return self._hooks[name]

    def remove(self, name: str) -> PluginHook | None:
        """
        Entfernt einen Hook.

        Args:
            name: Hook-Name

        Returns:
            Der entfernte Hook oder None
        """
        hook = self._hooks.pop(name, None)

        # Aus Plugin-Mapping entfernen
        for hooks in self._plugin_hooks.values():
            hooks.discard(name)

        return hook

    def register_handler(
        self,
        hook_name: str,
        plugin_name: str,
        callback: HookCallback,
        priority: HookPriority = HookPriority.NORMAL,
        once: bool = False,
    ) -> bool:
        """
        Registriert einen Handler für einen Hook.

        Erstellt den Hook automatisch wenn nicht vorhanden.

        Args:
            hook_name: Hook-Name
            plugin_name: Plugin-Name
            callback: Callback-Funktion
            priority: Ausführungspriorität
            once: Nur einmal ausführen

        Returns:
            True bei Erfolg
        """
        hook = self.get_or_create(hook_name)
        hook.add_handler(callback, plugin_name, priority, once)

        # Plugin-Mapping aktualisieren
        if plugin_name not in self._plugin_hooks:
            self._plugin_hooks[plugin_name] = set()
        self._plugin_hooks[plugin_name].add(hook_name)

        return True

    def unregister_handler(
        self,
        hook_name: str,
        plugin_name: str,
    ) -> int:
        """
        Entfernt Handler eines Plugins von einem Hook.

        Args:
            hook_name: Hook-Name
            plugin_name: Plugin-Name

        Returns:
            Anzahl entfernter Handler
        """
        hook = self._hooks.get(hook_name)
        if hook is None:
            return 0

        removed = hook.remove_handler(plugin_name)

        # Plugin-Mapping aktualisieren
        if plugin_name in self._plugin_hooks:
            self._plugin_hooks[plugin_name].discard(hook_name)

        return removed

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

        Args:
            plugin_name: Plugin-Name

        Returns:
            Gesamtanzahl entfernter Handler
        """
        total_removed = 0

        if plugin_name in self._plugin_hooks:
            for hook_name in list(self._plugin_hooks[plugin_name]):
                hook = self._hooks.get(hook_name)
                if hook:
                    total_removed += hook.remove_handler(plugin_name)

            del self._plugin_hooks[plugin_name]

        return total_removed

    def apply(
        self,
        hook_name: str,
        value: Any,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """
        Wendet einen Hook an.

        Args:
            hook_name: Hook-Name
            value: Zu verarbeitender Wert
            *args: Zusätzliche Argumente
            **kwargs: Keyword-Argumente

        Returns:
            HookResult oder der Originalwert
        """
        hook = self._hooks.get(hook_name)
        if hook is None:
            from trixy_core.plugins.hooks.hook import HookResult
            return HookResult(value=value)

        return hook.apply(value, *args, **kwargs)

    async def apply_async(
        self,
        hook_name: str,
        value: Any,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """Wendet einen Hook asynchron an."""
        hook = self._hooks.get(hook_name)
        if hook is None:
            from trixy_core.plugins.hooks.hook import HookResult
            return HookResult(value=value)

        return await hook.apply_async(value, *args, **kwargs)

    def trigger(
        self,
        hook_name: str,
        *args: Any,
        **kwargs: Any,
    ) -> list[Any]:
        """
        Triggert einen Hook als Action.

        Args:
            hook_name: Hook-Name
            *args: Argumente
            **kwargs: Keyword-Argumente

        Returns:
            Liste der Ergebnisse
        """
        hook = self._hooks.get(hook_name)
        if hook is None:
            return []

        return hook.trigger(*args, **kwargs)

    async def trigger_async(
        self,
        hook_name: str,
        *args: Any,
        **kwargs: Any,
    ) -> list[Any]:
        """Triggert einen Hook asynchron."""
        hook = self._hooks.get(hook_name)
        if hook is None:
            return []

        return await hook.trigger_async(*args, **kwargs)

    def get_hooks_for_plugin(self, plugin_name: str) -> list[str]:
        """
        Gibt alle Hooks zurück, für die ein Plugin Handler hat.

        Args:
            plugin_name: Plugin-Name

        Returns:
            Liste der Hook-Namen
        """
        return list(self._plugin_hooks.get(plugin_name, set()))

    def get_plugins_for_hook(self, hook_name: str) -> list[str]:
        """
        Gibt alle Plugins zurück, die Handler für einen Hook haben.

        Args:
            hook_name: Hook-Name

        Returns:
            Liste der Plugin-Namen
        """
        hook = self._hooks.get(hook_name)
        if hook is None:
            return []

        return list(set(h.plugin_name for h in hook.get_handlers()))

    def get_info(self, hook_name: str) -> HookInfo | None:
        """
        Gibt Informationen über einen Hook zurück.

        Args:
            hook_name: Hook-Name

        Returns:
            HookInfo oder None
        """
        hook = self._hooks.get(hook_name)
        if hook is None:
            return None

        handlers = []
        total_calls = 0

        for handler in hook.get_handlers():
            total_calls += handler.call_count
            handlers.append({
                "plugin": handler.plugin_name,
                "priority": handler.priority.name,
                "is_async": handler.is_async,
                "enabled": handler.enabled,
                "call_count": handler.call_count,
            })

        return HookInfo(
            name=hook.name,
            description=hook.description,
            handler_count=hook.handler_count,
            total_calls=total_calls,
            handlers=handlers,
        )

    def get_all_info(self) -> list[HookInfo]:
        """Gibt Informationen über alle Hooks zurück."""
        return [
            info for name in self._hooks
            if (info := self.get_info(name)) is not None
        ]

    def list_hooks(self) -> list[str]:
        """Gibt alle Hook-Namen zurück."""
        return list(self._hooks.keys())

    def has_hook(self, name: str) -> bool:
        """Prüft, ob ein Hook existiert."""
        return name in self._hooks

    def clear(self) -> None:
        """Entfernt alle Hooks und Handler."""
        self._hooks.clear()
        self._plugin_hooks.clear()

    def __contains__(self, name: str) -> bool:
        return name in self._hooks

    def __getitem__(self, name: str) -> PluginHook:
        if name not in self._hooks:
            raise KeyError(f"Hook nicht gefunden: {name}")
        return self._hooks[name]

    def __len__(self) -> int:
        return len(self._hooks)