216 lines
7.3 KiB
Python
216 lines
7.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Plugin-Registry – das Baukasten-Prinzip.
|
||
|
||
Ein neues Tool wird angelegt, indem ein Ordner unter plugins/<id>/ 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/<id> ein. Das Dashboard listet das Tool automatisch auf.
|
||
|
||
plugins/<id>/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/<id>
|
||
"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/<id>/<module>.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]
|