# -*- 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()