Dani
48585acb6f
Added life log, persona, and plugins manager. changed it so that any new .json files aren't uploaded
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
import os
|
|
import importlib.util
|
|
import sys
|
|
from typing import Any, Callable, Dict
|
|
|
|
PLUGINS_DIR = "plugins"
|
|
os.makedirs(PLUGINS_DIR, exist_ok=True)
|
|
|
|
|
|
class PluginManager:
|
|
"""
|
|
Dynamically loads Python modules from plugins/ and
|
|
exposes their functions in a registry.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.registry: Dict[str, Callable[..., Any]] = {}
|
|
self._load_all()
|
|
|
|
def _load_all(self):
|
|
"""Scan plugins/ and import every .py as a module."""
|
|
for fname in os.listdir(PLUGINS_DIR):
|
|
if not fname.endswith(".py"):
|
|
continue
|
|
path = os.path.join(PLUGINS_DIR, fname)
|
|
name = os.path.splitext(fname)[0]
|
|
self._load_module(name, path)
|
|
|
|
def _load_module(self, name: str, path: str):
|
|
"""Load a single plugin module and register its callables."""
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
if spec and spec.loader:
|
|
mod = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = mod
|
|
spec.loader.exec_module(mod) # type: ignore
|
|
for attr in dir(mod):
|
|
if attr.startswith("_"):
|
|
continue
|
|
obj = getattr(mod, attr)
|
|
if callable(obj):
|
|
key = f"{name}.{attr}"
|
|
self.registry[key] = obj
|
|
|
|
def register_plugin(self, code: str, name: str):
|
|
"""
|
|
Persist a new plugin file (name.py), load it immediately,
|
|
and add its functions to the registry.
|
|
"""
|
|
path = os.path.join(PLUGINS_DIR, f"{name}.py")
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(code)
|
|
# replace any existing module
|
|
if name in sys.modules:
|
|
del sys.modules[name]
|
|
self._load_module(name, path)
|