Trixy7/trixy_core/plugins/registry/discovery.py

# -*- coding: utf-8 -*-
"""
Plugin-Erkennung und -Scanning.

Scannt Verzeichnisse nach Plugins und extrahiert Metadaten.
"""
from __future__ import annotations

import importlib.util
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, TYPE_CHECKING

from trixy_core.plugins.registry.manifest import (
    PluginManifest,
    ManifestError,
    load_manifest,
)

if TYPE_CHECKING:
    from trixy_core.plugins.trixy_plugin import TrixyPlugin


@dataclass
class DiscoveryResult:
    """
    Ergebnis einer Plugin-Suche.

    Attributes:
        path: Pfad zum Plugin
        manifest: Plugin-Manifest (wenn vorhanden)
        valid: Ist das Plugin valide
        errors: Gefundene Fehler
        warnings: Warnungen
    """
    path: Path
    manifest: PluginManifest | None = None
    valid: bool = True
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

    @property
    def name(self) -> str:
        """Plugin-Name."""
        if self.manifest:
            return self.manifest.name
        return self.path.name

    @property
    def version(self) -> str:
        """Plugin-Version."""
        if self.manifest:
            return self.manifest.version
        return "unknown"

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "path": str(self.path),
            "name": self.name,
            "version": self.version,
            "valid": self.valid,
            "errors": self.errors,
            "warnings": self.warnings,
            "manifest": self.manifest.to_dict() if self.manifest else None,
        }


class PluginDiscovery:
    """
    Erkennt und validiert Plugins in Verzeichnissen.

    Scannt Plugin-Verzeichnisse, lädt Manifeste und prüft
    grundlegende Anforderungen.

    Example:
        discovery = PluginDiscovery()

        # Alle Plugins im Verzeichnis finden
        plugins = discovery.scan(Path("plugins"))

        for result in plugins:
            if result.valid:
                print(f"Gefunden: {result.name} v{result.version}")
            else:
                print(f"Ungültig: {result.name}: {result.errors}")

        # Einzelnes Plugin prüfen
        result = discovery.check_plugin(Path("plugins/my_plugin"))

        # Plugin-Klasse extrahieren
        plugin_class = discovery.extract_plugin_class(result.path)
    """

    MANIFEST_FILES = ["manifest.json", "plugin.json"]
    ENTRY_POINTS = ["main.py", "plugin.py", "__init__.py"]
    CONFIG_FILES = ["config.json", "settings.json"]

    def __init__(self) -> None:
        """Initialisiert die Plugin-Erkennung."""
        self._cache: dict[Path, DiscoveryResult] = {}

    def scan(
        self,
        directory: Path,
        recursive: bool = False,
        use_cache: bool = True,
    ) -> list[DiscoveryResult]:
        """
        Scannt ein Verzeichnis nach Plugins.

        Args:
            directory: Zu scannendes Verzeichnis
            recursive: Auch Unterverzeichnisse scannen
            use_cache: Cache verwenden

        Returns:
            Liste der gefundenen Plugins
        """
        if not directory.exists():
            return []

        results = []

        for item in directory.iterdir():
            if not item.is_dir():
                continue

            # Versteckte und private Verzeichnisse ignorieren
            if item.name.startswith(".") or item.name.startswith("_"):
                continue

            # Cache prüfen
            if use_cache and item in self._cache:
                results.append(self._cache[item])
                continue

            result = self.check_plugin(item)
            results.append(result)

            if use_cache:
                self._cache[item] = result

            # Rekursiv scannen
            if recursive and not result.valid:
                results.extend(self.scan(item, recursive=True, use_cache=use_cache))

        return results

    def check_plugin(self, path: Path) -> DiscoveryResult:
        """
        Prüft ein einzelnes Plugin-Verzeichnis.

        Args:
            path: Pfad zum Plugin

        Returns:
            DiscoveryResult
        """
        errors: list[str] = []
        warnings: list[str] = []
        manifest: PluginManifest | None = None

        if not path.exists():
            return DiscoveryResult(
                path=path,
                valid=False,
                errors=["Verzeichnis existiert nicht"],
            )

        if not path.is_dir():
            return DiscoveryResult(
                path=path,
                valid=False,
                errors=["Kein Verzeichnis"],
            )

        # Manifest suchen
        manifest = self._find_manifest(path, errors, warnings)

        # Entry-Point prüfen
        entry_point = self._find_entry_point(path, manifest)
        if entry_point is None:
            errors.append(
                f"Kein Entry-Point gefunden (erwartet: {', '.join(self.ENTRY_POINTS)})"
            )

        # Config prüfen
        config_path = self._find_config(path)
        if config_path is None and manifest is None:
            warnings.append("Keine Konfigurationsdatei gefunden")

        # Syntax prüfen
        if entry_point:
            syntax_errors = self._check_syntax(entry_point)
            errors.extend(syntax_errors)

        valid = len(errors) == 0

        return DiscoveryResult(
            path=path,
            manifest=manifest,
            valid=valid,
            errors=errors,
            warnings=warnings,
        )

    def _find_manifest(
        self,
        path: Path,
        errors: list[str],
        warnings: list[str],
    ) -> PluginManifest | None:
        """Sucht und lädt das Manifest."""
        for manifest_name in self.MANIFEST_FILES:
            manifest_path = path / manifest_name
            if manifest_path.exists():
                try:
                    return load_manifest(manifest_path)
                except ManifestError as e:
                    errors.append(f"Manifest-Fehler: {e}")
                    return None

        # Fallback: Aus main.py extrahieren
        main_file = path / "main.py"
        if main_file.exists():
            try:
                return self._extract_manifest_from_class(main_file, path.name)
            except Exception:
                pass

        warnings.append("Kein Manifest gefunden, verwende Standardwerte")
        return PluginManifest(name=path.name)

    def _extract_manifest_from_class(
        self,
        file_path: Path,
        default_name: str,
    ) -> PluginManifest | None:
        """Extrahiert Manifest-Informationen aus der Plugin-Klasse."""
        try:
            # Modul laden ohne auszuführen
            import ast

            with open(file_path, "r", encoding="utf-8") as f:
                tree = ast.parse(f.read())

            for node in ast.walk(tree):
                if isinstance(node, ast.ClassDef):
                    # Klassenattribute extrahieren
                    info = {"name": default_name}

                    for item in node.body:
                        if isinstance(item, ast.Assign):
                            for target in item.targets:
                                if isinstance(target, ast.Name):
                                    if target.id == "NAME" and isinstance(item.value, ast.Constant):
                                        info["name"] = item.value.value
                                    elif target.id == "VERSION" and isinstance(item.value, ast.Constant):
                                        info["version"] = item.value.value
                                    elif target.id == "DESCRIPTION" and isinstance(item.value, ast.Constant):
                                        info["description"] = item.value.value
                                    elif target.id == "AUTHOR" and isinstance(item.value, ast.Constant):
                                        info["author"] = item.value.value

                    return PluginManifest(**info)

        except Exception:
            pass

        return None

    def _find_entry_point(
        self,
        path: Path,
        manifest: PluginManifest | None,
    ) -> Path | None:
        """Findet den Entry-Point."""
        if manifest and manifest.entry_point:
            entry = path / manifest.entry_point
            if entry.exists():
                return entry

        for entry_name in self.ENTRY_POINTS:
            entry = path / entry_name
            if entry.exists():
                return entry

        return None

    def _find_config(self, path: Path) -> Path | None:
        """Findet die Konfigurationsdatei."""
        for config_name in self.CONFIG_FILES:
            config = path / config_name
            if config.exists():
                return config
        return None

    def _check_syntax(self, file_path: Path) -> list[str]:
        """Prüft die Python-Syntax."""
        errors = []

        try:
            with open(file_path, "r", encoding="utf-8") as f:
                source = f.read()
            compile(source, str(file_path), "exec")
        except SyntaxError as e:
            errors.append(f"Syntax-Fehler in {file_path.name}: {e.msg} (Zeile {e.lineno})")

        return errors

    def extract_plugin_class(
        self,
        path: Path,
    ) -> type["TrixyPlugin"] | None:
        """
        Extrahiert die Plugin-Klasse aus einem Verzeichnis.

        Args:
            path: Pfad zum Plugin

        Returns:
            Die Plugin-Klasse oder None
        """
        from trixy_core.plugins.trixy_plugin import TrixyPlugin

        result = self.check_plugin(path)
        if not result.valid:
            return None

        entry_point = self._find_entry_point(path, result.manifest)
        if entry_point is None:
            return None

        try:
            spec = importlib.util.spec_from_file_location(
                f"trixy_plugins.{path.name}",
                entry_point
            )

            if spec is None or spec.loader is None:
                return None

            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            # Plugin-Klasse finden
            for attr_name in dir(module):
                if attr_name.startswith("_"):
                    continue

                attr = getattr(module, attr_name)

                if (
                    isinstance(attr, type) and
                    issubclass(attr, TrixyPlugin) and
                    attr is not TrixyPlugin
                ):
                    return attr

        except Exception:
            pass

        return None

    def get_valid_plugins(
        self,
        directory: Path,
    ) -> list[DiscoveryResult]:
        """
        Gibt nur valide Plugins zurück.

        Args:
            directory: Zu scannendes Verzeichnis

        Returns:
            Liste valider Plugins
        """
        results = self.scan(directory)
        return [r for r in results if r.valid]

    def clear_cache(self, path: Path | None = None) -> None:
        """
        Löscht den Cache.

        Args:
            path: Optionaler spezifischer Pfad
        """
        if path:
            self._cache.pop(path, None)
        else:
            self._cache.clear()