# -*- coding: utf-8 -*- """ Plugin-Registry und Marketplace. Zentrale Verwaltung von Plugin-Informationen und -Versionen. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Iterator import json from trixy_core.plugins.registry.manifest import PluginManifest @dataclass class PluginInfo: """ Kurzinformationen zu einem Plugin. Attributes: name: Plugin-Name version: Aktuelle Version description: Beschreibung author: Autor installed: Ist installiert enabled: Ist aktiviert path: Installationspfad """ name: str version: str description: str = "" author: str = "" installed: bool = False enabled: bool = False path: Path | None = None def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "name": self.name, "version": self.version, "description": self.description, "author": self.author, "installed": self.installed, "enabled": self.enabled, "path": str(self.path) if self.path else None, } @dataclass class RegistryEntry: """ Vollständiger Registry-Eintrag für ein Plugin. Attributes: manifest: Plugin-Manifest available_versions: Verfügbare Versionen installed_version: Installierte Version install_path: Installationspfad installed_at: Installationszeitpunkt updated_at: Letztes Update enabled: Ist aktiviert download_count: Download-Zähler rating: Bewertung (0-5) tags: Tags """ manifest: PluginManifest available_versions: list[str] = field(default_factory=list) installed_version: str | None = None install_path: Path | None = None installed_at: datetime | None = None updated_at: datetime | None = None enabled: bool = True download_count: int = 0 rating: float = 0.0 tags: list[str] = field(default_factory=list) @property def name(self) -> str: """Plugin-Name.""" return self.manifest.name @property def is_installed(self) -> bool: """Ist das Plugin installiert?""" return self.installed_version is not None @property def has_update(self) -> bool: """Ist ein Update verfügbar?""" if not self.installed_version or not self.available_versions: return False from trixy_core.plugins.versioning.semver import parse_version try: installed = parse_version(self.installed_version) latest = parse_version(self.available_versions[-1]) return latest > installed except ValueError: return False def to_info(self) -> PluginInfo: """Konvertiert zu PluginInfo.""" return PluginInfo( name=self.manifest.name, version=self.installed_version or self.manifest.version, description=self.manifest.description, author=self.manifest.author, installed=self.is_installed, enabled=self.enabled, path=self.install_path, ) def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "manifest": self.manifest.to_dict(), "available_versions": self.available_versions, "installed_version": self.installed_version, "install_path": str(self.install_path) if self.install_path else None, "installed_at": self.installed_at.isoformat() if self.installed_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "enabled": self.enabled, "download_count": self.download_count, "rating": self.rating, "tags": self.tags, } class PluginRegistry: """ Zentrale Plugin-Registry. Verwaltet alle bekannten Plugins, deren Versionen und Installationsstatus. Example: registry = PluginRegistry() # Plugin registrieren registry.register(manifest) # Als installiert markieren registry.mark_installed("my_plugin", "1.0.0", Path("plugins/my_plugin")) # Suchen results = registry.search("audio") # Nach Kategorie filtern audio_plugins = registry.get_by_category("audio") # Updates prüfen updates = registry.get_available_updates() # Persistenz registry.save(Path("registry.json")) registry.load(Path("registry.json")) """ def __init__(self) -> None: """Initialisiert die Registry.""" self._entries: dict[str, RegistryEntry] = {} @property def count(self) -> int: """Anzahl registrierter Plugins.""" return len(self._entries) @property def installed_count(self) -> int: """Anzahl installierter Plugins.""" return sum(1 for e in self._entries.values() if e.is_installed) def register( self, manifest: PluginManifest, versions: list[str] | None = None, ) -> RegistryEntry: """ Registriert ein Plugin. Args: manifest: Plugin-Manifest versions: Verfügbare Versionen Returns: Der Registry-Eintrag """ versions = versions or [manifest.version] entry = RegistryEntry( manifest=manifest, available_versions=sorted(versions), tags=manifest.tags.copy(), ) self._entries[manifest.name] = entry return entry def unregister(self, name: str) -> RegistryEntry | None: """ Entfernt ein Plugin aus der Registry. Args: name: Plugin-Name Returns: Der entfernte Eintrag oder None """ return self._entries.pop(name, None) def get(self, name: str) -> RegistryEntry | None: """Gibt einen Eintrag zurück.""" return self._entries.get(name) def get_info(self, name: str) -> PluginInfo | None: """Gibt Plugin-Info zurück.""" entry = self._entries.get(name) return entry.to_info() if entry else None def get_all(self) -> list[RegistryEntry]: """Gibt alle Einträge zurück.""" return list(self._entries.values()) def get_installed(self) -> list[RegistryEntry]: """Gibt alle installierten Plugins zurück.""" return [e for e in self._entries.values() if e.is_installed] def get_enabled(self) -> list[RegistryEntry]: """Gibt alle aktivierten Plugins zurück.""" return [e for e in self._entries.values() if e.is_installed and e.enabled] def mark_installed( self, name: str, version: str, path: Path, ) -> bool: """ Markiert ein Plugin als installiert. Args: name: Plugin-Name version: Installierte Version path: Installationspfad Returns: True bei Erfolg """ entry = self._entries.get(name) if entry is None: return False entry.installed_version = version entry.install_path = path entry.installed_at = datetime.now() entry.updated_at = datetime.now() return True def mark_uninstalled(self, name: str) -> bool: """ Markiert ein Plugin als deinstalliert. Args: name: Plugin-Name Returns: True bei Erfolg """ entry = self._entries.get(name) if entry is None: return False entry.installed_version = None entry.install_path = None entry.enabled = False return True def set_enabled(self, name: str, enabled: bool) -> bool: """ Aktiviert/Deaktiviert ein Plugin. Args: name: Plugin-Name enabled: Aktivieren oder Deaktivieren Returns: True bei Erfolg """ entry = self._entries.get(name) if entry is None or not entry.is_installed: return False entry.enabled = enabled return True def add_version(self, name: str, version: str) -> bool: """ Fügt eine verfügbare Version hinzu. Args: name: Plugin-Name version: Neue Version Returns: True bei Erfolg """ entry = self._entries.get(name) if entry is None: return False if version not in entry.available_versions: entry.available_versions.append(version) entry.available_versions.sort() return True def search( self, query: str, installed_only: bool = False, ) -> list[RegistryEntry]: """ Sucht Plugins nach Name, Beschreibung oder Tags. Args: query: Suchbegriff installed_only: Nur installierte Returns: Liste passender Einträge """ query = query.lower() results = [] for entry in self._entries.values(): if installed_only and not entry.is_installed: continue # Im Namen suchen if query in entry.manifest.name.lower(): results.append(entry) continue # In Beschreibung suchen if query in entry.manifest.description.lower(): results.append(entry) continue # In Tags suchen if any(query in tag.lower() for tag in entry.tags): results.append(entry) continue # In Keywords suchen if any(query in kw.lower() for kw in entry.manifest.keywords): results.append(entry) return results def get_by_category(self, category: str) -> list[RegistryEntry]: """Filtert nach Kategorie.""" return [ e for e in self._entries.values() if e.manifest.category == category ] def get_by_tag(self, tag: str) -> list[RegistryEntry]: """Filtert nach Tag.""" return [e for e in self._entries.values() if tag in e.tags] def get_available_updates(self) -> list[RegistryEntry]: """Gibt Plugins mit verfügbaren Updates zurück.""" return [e for e in self._entries.values() if e.has_update] def get_dependencies_of(self, name: str) -> list[str]: """ Gibt die Abhängigkeiten eines Plugins zurück. Args: name: Plugin-Name Returns: Liste der Abhängigkeitsnamen """ entry = self._entries.get(name) if entry is None: return [] return list(entry.manifest.dependencies.keys()) def get_dependents_of(self, name: str) -> list[str]: """ Gibt Plugins zurück, die von diesem abhängen. Args: name: Plugin-Name Returns: Liste der abhängigen Plugin-Namen """ dependents = [] for entry in self._entries.values(): if name in entry.manifest.dependencies: dependents.append(entry.name) return dependents def save(self, path: Path) -> None: """ Speichert die Registry in eine Datei. Args: path: Zielpfad """ data = { "version": "1.0", "entries": { name: entry.to_dict() for name, entry in self._entries.items() } } with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) def load(self, path: Path) -> int: """ Lädt die Registry aus einer Datei. Args: path: Quellpfad Returns: Anzahl geladener Einträge """ if not path.exists(): return 0 with open(path, "r", encoding="utf-8") as f: data = json.load(f) for name, entry_data in data.get("entries", {}).items(): try: manifest = PluginManifest.from_dict(entry_data["manifest"]) entry = RegistryEntry( manifest=manifest, available_versions=entry_data.get("available_versions", []), installed_version=entry_data.get("installed_version"), install_path=Path(entry_data["install_path"]) if entry_data.get("install_path") else None, enabled=entry_data.get("enabled", True), download_count=entry_data.get("download_count", 0), rating=entry_data.get("rating", 0.0), tags=entry_data.get("tags", []), ) if entry_data.get("installed_at"): entry.installed_at = datetime.fromisoformat(entry_data["installed_at"]) if entry_data.get("updated_at"): entry.updated_at = datetime.fromisoformat(entry_data["updated_at"]) self._entries[name] = entry except Exception: continue return len(self._entries) def __contains__(self, name: str) -> bool: return name in self._entries def __iter__(self) -> Iterator[RegistryEntry]: return iter(self._entries.values()) def __len__(self) -> int: return len(self._entries)