Spaces:
Running on Zero
Running on Zero
File size: 3,702 Bytes
f7e32a5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """Plugin manager — loads, registers, and manages plugins."""
import json
import logging
from typing import Optional
from pathlib import Path
from config import config
from database.repositories import PluginRepo
logger = logging.getLogger("synapse.plugins")
class PluginAPI:
"""API surface exposed to plugins."""
def __init__(self):
self.tools: dict = {}
self.hooks: dict = {}
def register_tool(self, name: str, description: str, handler, parameters: dict = None):
self.tools[name] = {
"name": name,
"description": description,
"handler": handler,
"parameters": parameters or {},
}
logger.info(f"Plugin registered tool: {name}")
def register_hook(self, event: str, handler):
if event not in self.hooks:
self.hooks[event] = []
self.hooks[event].append(handler)
logger.info(f"Plugin registered hook: {event}")
async def trigger_hook(self, event: str, **kwargs):
results = []
for handler in self.hooks.get(event, []):
try:
if hasattr(handler, '__call__'):
result = handler(**kwargs)
results.append(result)
except Exception as e:
logger.error(f"Hook {event} error: {e}")
return results
class PluginManager:
"""Manages plugin lifecycle."""
def __init__(self):
self.api = PluginAPI()
self._loaded_plugins: dict[str, object] = {}
async def load_enabled_plugins(self):
plugins = await PluginRepo.get_enabled()
for plugin in plugins:
await self._load_plugin(plugin)
async def _load_plugin(self, plugin_data: dict):
plugin_id = plugin_data["id"]
name = plugin_data["name"]
try:
plugin_dir = Path(f"plugins/builtin/{name}")
if plugin_dir.exists():
import importlib.util
spec = importlib.util.spec_from_file_location(
f"plugins.builtin.{name}", plugin_dir / "main.py"
)
if spec and spec.loader:
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
if hasattr(module, "register"):
module.register(self.api)
self._loaded_plugins[plugin_id] = module
logger.info(f"Loaded plugin: {name}")
except Exception as e:
logger.error(f"Failed to load plugin {name}: {e}")
async def install_plugin(self, name: str, description: str = "",
author: str = "") -> dict:
from database.repositories import PluginRepo
plugin_id = PluginRepo.new_id()
from database import engine as db_engine
await db_engine.execute(
"INSERT INTO plugins (id, name, description, author) VALUES (?, ?, ?, ?)",
(plugin_id, name, description, author),
)
return {"id": plugin_id, "name": name}
async def toggle_plugin(self, plugin_id: str) -> dict:
await PluginRepo.toggle(plugin_id)
plugin = await PluginRepo.get_by_id(plugin_id)
return plugin
async def get_all_plugins(self) -> list[dict]:
from database import engine as db_engine
return await db_engine.fetch_all("SELECT * FROM plugins ORDER BY name")
def get_registered_tools(self) -> list[dict]:
return [
{"name": t["name"], "description": t["description"], "parameters": t["parameters"]}
for t in self.api.tools.values()
]
plugin_manager = PluginManager()
|