wehub-resource-sync
67 lines
1.5 KiB
Python
67 lines
1.5 KiB
Python
import os
|
|
import sys
|
|
import importlib
|
|
|
|
# 模块级注册表: event_name -> [callback, ...]
|
|
_registry = {}
|
|
_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
def register(event):
|
|
def decorator(fn):
|
|
_registry.setdefault(event, []).append(fn)
|
|
return fn
|
|
return decorator
|
|
|
|
|
|
def trigger(event, ctx: dict):
|
|
for fn in _registry.get(event, []):
|
|
try:
|
|
r = fn(ctx)
|
|
if isinstance(r, dict):
|
|
ctx = r
|
|
except Exception as e:
|
|
sys.stderr.write(f"[hooks] {event} callback error: {e}\n")
|
|
return ctx
|
|
|
|
|
|
def unregister(event, fn):
|
|
try:
|
|
_registry[event] = [f for f in _registry[event] if f is not fn]
|
|
except KeyError:
|
|
pass
|
|
|
|
|
|
def clear(event=None):
|
|
if event:
|
|
_registry.pop(event, None)
|
|
else:
|
|
_registry.clear()
|
|
|
|
|
|
def has(event):
|
|
return bool(_registry.get(event))
|
|
|
|
|
|
def discover_and_load(plugin_dir=None):
|
|
if plugin_dir is None:
|
|
plugin_dir = os.path.join(_PROJECT_ROOT, 'plugins')
|
|
if not os.path.isdir(plugin_dir):
|
|
return
|
|
parent = os.path.dirname(plugin_dir)
|
|
if parent not in sys.path:
|
|
sys.path.insert(0, parent)
|
|
for fn in sorted(os.listdir(plugin_dir)):
|
|
if fn.startswith('_') or not fn.endswith('.py'):
|
|
continue
|
|
name = fn[:-3]
|
|
load(name)
|
|
|
|
|
|
def load(name):
|
|
try:
|
|
importlib.import_module(f'plugins.{name}')
|
|
return True
|
|
except Exception as e:
|
|
sys.stderr.write(f"[hooks] plugin '{name}' load failed: {e}\n")
|
|
return False |