plugin_registry.py

"""v0.3 后处理插件框架。插件只接收已解析 Transaction，不进入读卡核心。"""
from __future__ import annotations

import importlib.util
import os
import traceback

import paths

_loaded = None
_errors = []
_runtime_errors = []


def plugin_dir():
    d = paths.data_path("plugins")
    os.makedirs(d, exist_ok=True)
    return d


def discover(force=False):
    global _loaded, _errors, _runtime_errors
    if _loaded is not None and not force:
        return _loaded
    _loaded = []; _errors = []
    if force:
        _runtime_errors = []
    for name in sorted(os.listdir(plugin_dir())):
        if not name.endswith(".py") or name.startswith("_"):
            continue
        path = os.path.join(plugin_dir(), name)
        try:
            spec = importlib.util.spec_from_file_location(f"neko_plugin_{name[:-3]}", path)
            module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
            hook = getattr(module, "postprocess", None)
            if callable(hook):
                _loaded.append({"name": getattr(module, "PLUGIN_NAME", name[:-3]),
                                "version": getattr(module, "PLUGIN_VERSION", "0"),
                                "path": path, "hook": hook})
        except Exception:
            _errors.append({"name": name, "error": traceback.format_exc(limit=3)})
    return _loaded


def apply(transactions, context=None):
    global _runtime_errors
    result = list(transactions); errors = []
    for plugin in discover():
        try:
            value = plugin["hook"](result, context or {})
            if value is not None:
                candidate = list(value)
                if not all(callable(getattr(tx, "to_dict", None)) for tx in candidate):
                    raise TypeError("postprocess 必须返回 Transaction 序列")
                result = candidate
        except Exception as exc:
            errors.append({"plugin": plugin["name"], "error": str(exc)})
    _runtime_errors = list(errors)
    return result, errors


def status():
    return {"directory": plugin_dir(),
            "plugins": [{k: p[k] for k in ("name", "version", "path")} for p in discover()],
            "errors": list(_errors) + [{"name": e["plugin"], "error": e["error"]}
                                        for e in _runtime_errors]}