# -*- coding: utf-8 -*- """ Plugin-Registry – das Baukasten-Prinzip. Ein neues Tool wird angelegt, indem ein Ordner unter plugins// abgelegt wird. Mehr ist nicht nötig: Beim Start liest die Suite jede plugin.json ein, importiert das darin genannte Backend und hängt dessen ASGI-App unter /plugins/ ein. Das Dashboard listet das Tool automatisch auf. plugins//plugin.json ------------------------- { "id": "mein-tool", Pflicht, identisch zum Ordnernamen "name": "Mein Tool", Pflicht, Anzeigename "summary": "Kurz in einer Zeile", "description": "Längerer Text für die Dashboard-Karte", "icon": "🔧", "category": "Tandoor", "version": "1.0.0", "entrypoint": "backend:create_app", Modul:Funktion, relativ zum Plugin-Ordner "mount": "/plugins/mein-tool", optional, Standard ist /plugins/ "requires": ["tandoor"], optional: tandoor | openai "features": ["Stichpunkt", "..."], optional, für die Dashboard-Karte "enabled": true } backend.py ---------- def create_app(ctx: PluginContext): ... return app # FastAPI-, Starlette- oder (via a2wsgi) WSGI-App """ from __future__ import annotations import importlib.util import json import sys import traceback from dataclasses import dataclass, field from pathlib import Path from typing import Any from .jobs import JobManager from .settings import Settings # --------------------------------------------------------------------- Modell @dataclass class PluginMeta: id: str name: str directory: Path summary: str = "" description: str = "" icon: str = "▪" category: str = "Allgemein" version: str = "1.0.0" entrypoint: str = "backend:create_app" mount: str = "" requires: list[str] = field(default_factory=list) features: list[str] = field(default_factory=list) docs: str = "" enabled: bool = True order: int = 100 @classmethod def from_file(cls, manifest: Path) -> "PluginMeta": raw: dict[str, Any] = json.loads(manifest.read_text(encoding="utf-8")) directory = manifest.parent pid = str(raw.get("id") or directory.name) meta = cls( id=pid, name=str(raw.get("name") or pid), directory=directory, summary=str(raw.get("summary", "")), description=str(raw.get("description", "")), icon=str(raw.get("icon", "▪")), category=str(raw.get("category", "Allgemein")), version=str(raw.get("version", "1.0.0")), entrypoint=str(raw.get("entrypoint", "backend:create_app")), mount=str(raw.get("mount") or f"/plugins/{pid}"), requires=list(raw.get("requires", []) or []), features=list(raw.get("features", []) or []), docs=str(raw.get("docs", "")), enabled=bool(raw.get("enabled", True)), order=int(raw.get("order", 100)), ) if not meta.mount.startswith("/"): meta.mount = "/" + meta.mount return meta def public(self) -> dict[str, Any]: return { "id": self.id, "name": self.name, "summary": self.summary, "description": self.description, "icon": self.icon, "category": self.category, "version": self.version, "mount": self.mount, "requires": self.requires, "features": self.features, "docs": self.docs, } @dataclass class PluginContext: """Wird jedem Plugin beim Start übergeben.""" meta: PluginMeta settings: Settings jobs: JobManager data_dir: Path shared_dir: Path @property def id(self) -> str: return self.meta.id @property def dir(self) -> Path: return self.meta.directory @property def mount(self) -> str: return self.meta.mount def path(self, *parts: str) -> Path: return self.meta.directory.joinpath(*parts) @dataclass class LoadedPlugin: meta: PluginMeta app: Any = None error: str = "" @property def ok(self) -> bool: return self.app is not None and not self.error # ------------------------------------------------------------------ Discovery def discover(plugins_dir: Path) -> list[PluginMeta]: metas: list[PluginMeta] = [] if not plugins_dir.is_dir(): return metas for entry in sorted(plugins_dir.iterdir()): manifest = entry / "plugin.json" if not entry.is_dir() or not manifest.is_file(): continue try: meta = PluginMeta.from_file(manifest) except Exception as exc: print(f"[plugins] {entry.name}: plugin.json unlesbar – {exc}", file=sys.stderr) continue if meta.enabled: metas.append(meta) metas.sort(key=lambda m: (m.order, m.name.casefold())) return metas def _import_module(meta: PluginMeta, module_name: str): """Importiert plugins//.py isoliert unter eindeutigem Namen.""" file = meta.directory / f"{module_name}.py" if not file.is_file(): raise FileNotFoundError(f"{file} fehlt") qualified = f"boehmitools_plugin_{meta.id.replace('-', '_')}_{module_name}" spec = importlib.util.spec_from_file_location(qualified, file) if spec is None or spec.loader is None: raise ImportError(f"{file} kann nicht geladen werden") module = importlib.util.module_from_spec(spec) sys.modules[qualified] = module # Das Plugin-Verzeichnis muss importierbar sein, damit die Original-Tools # ihre eigenen Module weiterhin schlicht per "import x" finden. directory = str(meta.directory) if directory not in sys.path: sys.path.insert(0, directory) spec.loader.exec_module(module) return module def load(meta: PluginMeta, ctx: PluginContext) -> LoadedPlugin: try: module_name, _, func_name = meta.entrypoint.partition(":") module = _import_module(meta, module_name or "backend") factory = getattr(module, func_name or "create_app", None) if factory is None: raise AttributeError(f"{meta.entrypoint} nicht gefunden") app = factory(ctx) if app is None: raise ValueError("create_app() hat keine App zurückgegeben") return LoadedPlugin(meta=meta, app=app) except Exception as exc: detail = traceback.format_exc(limit=6) print(f"[plugins] {meta.id}: Laden fehlgeschlagen\n{detail}", file=sys.stderr) return LoadedPlugin(meta=meta, error=f"{type(exc).__name__}: {exc}") def unload(meta: PluginMeta) -> None: """ Entfernt alle Module, die aus dem Ordner dieses Plugins geladen wurden, aus ``sys.modules`` — sowohl das qualifizierte Backend als auch Untermodule, die das Plugin per ``import x`` aus seinem eigenen Ordner gezogen hat. Nötig, damit ein erneutes ``load`` geänderten Code wirklich neu einliest und nicht die zwischengespeicherte Fassung wiederverwendet. Kernmodule (``core``) werden bewusst NICHT angefasst — die laufen ja gerade. """ prefix = str(meta.directory.resolve()) for name, module in list(sys.modules.items()): datei = getattr(module, "__file__", None) if not datei: continue try: unter_plugin = str(Path(datei).resolve()).startswith(prefix) except (OSError, ValueError): continue if unter_plugin: del sys.modules[name]