Lib.core.PluginManager

  1"""
  2插件管理器
  3"""
  4
  5import dataclasses
  6import importlib
  7import inspect
  8import sys
  9import traceback
 10
 11from Lib.common import save_exc_dump
 12from Lib.constants import *
 13from Lib.core import ConfigManager
 14from Lib.core.EventManager import event_listener
 15from Lib.core.ListenerServer import EscalationEvent
 16from Lib.utils.Logger import get_logger
 17
 18logger = get_logger()
 19
 20plugins: list[dict] = []
 21found_plugins: list[dict] = []
 22has_main_func_plugins: list[dict] = []
 23
 24if not os.path.exists(PLUGINS_PATH):
 25    os.makedirs(PLUGINS_PATH)
 26
 27
 28class NotEnabledPluginException(Exception):
 29    """
 30    插件未启用的异常
 31    """
 32    pass
 33
 34
 35def load_plugin(plugin):
 36    """
 37    加载插件
 38    Args:
 39        plugin: 插件信息
 40    """
 41    name = plugin["name"]
 42    full_path = plugin["path"]
 43    is_package = os.path.isdir(full_path) and os.path.exists(os.path.join(full_path, "__init__.py"))
 44
 45    # 计算导入路径
 46    # 获取相对于 WORK_PATH 的路径，例如 "plugins/AIChat" 或 "plugins/single_file_plugin.py"
 47    relative_plugin_path = os.path.relpath(full_path, start=WORK_PATH)
 48
 49    # 将路径分隔符替换为点，例如 "plugins.AIChat" 或 "plugins.single_file_plugin"
 50    import_path = relative_plugin_path.replace(os.sep, '.')
 51    if not is_package and import_path.endswith('.py'):
 52        import_path = import_path[:-3]  # 去掉 .py 后缀
 53
 54    logger.debug(f"计算 {name} 得到的导入路径: {import_path}")
 55
 56    if WORK_PATH not in sys.path:
 57        logger.warning(f"项目根目录 {WORK_PATH} 不在 sys.path 中，正在添加。请检查执行环境。")
 58        sys.path.insert(0, WORK_PATH)  # 插入到前面，优先查找
 59
 60    try:
 61        logger.debug(f"尝试加载: {import_path}")
 62        module = importlib.import_module(import_path)
 63    except ImportError as e:
 64        logger.error(f"加载 {import_path} 失败: {repr(e)}\n"
 65                     f"{traceback.format_exc()}")
 66        raise
 67
 68    plugin_info = None
 69    try:
 70        if isinstance(module.plugin_info, PluginInfo):
 71            plugin_info = module.plugin_info
 72        else:
 73            logger.warning(f"插件 {name} 的 plugin_info 并非 PluginInfo 类型，无法获取插件信息")
 74    except AttributeError:
 75        logger.warning(f"插件 {name} 未定义 plugin_info 属性，无法获取插件信息")
 76
 77    return module, plugin_info
 78
 79
 80def load_plugins():
 81    """
 82    加载插件
 83    """
 84    global plugins, found_plugins
 85
 86    found_plugins = []
 87    # 获取插件目录下的所有文件
 88    for plugin in os.listdir(PLUGINS_PATH):
 89        if plugin == "__pycache__":
 90            continue
 91        full_path = os.path.join(PLUGINS_PATH, plugin)
 92        if (
 93                os.path.isdir(full_path) and
 94                os.path.exists(os.path.join(full_path, "__init__.py")) and
 95                os.path.isfile(os.path.join(full_path, "__init__.py"))
 96        ):
 97            file_path = os.path.join(os.path.join(full_path, "__init__.py"))
 98            name = plugin
 99        elif os.path.isfile(full_path) and full_path.endswith(".py"):
100            file_path = full_path
101            name = os.path.split(file_path)[1]
102        else:
103            logger.warning(f"{full_path} 不是一个有效的插件")
104            continue
105        logger.debug(f"找到插件 {file_path} 待加载")
106        plugin = {"name": name, "plugin": None, "info": None, "file_path": file_path, "path": full_path}
107        found_plugins.append(plugin)
108
109    plugins = []
110
111    for plugin in found_plugins:
112        name = plugin["name"]
113        full_path = plugin["path"]
114
115        if plugin["plugin"] is not None:
116            # 由于其他原因已被加载（例如插件依赖）
117            logger.debug(f"插件 {name} 已被加载，跳过加载")
118            continue
119
120        logger.debug(f"开始尝试加载插件 {full_path}")
121
122        try:
123            module, plugin_info = load_plugin(plugin)
124
125            plugin["info"] = plugin_info
126            plugin["plugin"] = module
127            plugins.append(plugin)
128        except NotEnabledPluginException:
129            logger.warning(f"插件 {name}({full_path}) 已被禁用，将不会被加载")
130            continue
131        except Exception as e:
132            if ConfigManager.GlobalConfig().debug.save_dump:
133                dump_path = save_exc_dump(f"尝试加载插件 {full_path} 时失败")
134            else:
135                dump_path = None
136            logger.error(f"尝试加载插件 {full_path} 时失败！ 原因:{repr(e)}\n"
137                         f"{"".join(traceback.format_exc())}"
138                         f"{f"\n已保存异常到 {dump_path}" if dump_path else ""}")
139            continue
140
141        logger.debug(f"插件 {name}({full_path}) 加载成功！")
142
143
144@dataclasses.dataclass
145class PluginInfo:
146    """
147    插件信息
148    """
149    NAME: str  # 插件名称
150    AUTHOR: str  # 插件作者
151    VERSION: str  # 插件版本
152    DESCRIPTION: str  # 插件描述
153    HELP_MSG: str  # 插件帮助
154    ENABLED: bool = True  # 插件是否启用
155    IS_HIDDEN: bool = False  # 插件是否隐藏（在/help命令中）
156    extra: dict | None = None  # 一个字典，可以用于存储任意信息。其他插件可以通过约定 extra 字典的键名来达成收集某些特殊信息的目的。
157
158    def __post_init__(self):
159        if self.ENABLED is not True:
160            raise NotEnabledPluginException
161        if self.extra is None:
162            self.extra = {}
163
164
165def requirement_plugin(plugin_name: str):
166    """
167    插件依赖
168    Args:
169        plugin_name: 插件的名称，如果依赖的是库形式的插件则是库文件夹的名称，如果依赖的是文件形式则是插件文件的名称（文件名称包含后缀）
170
171    Returns:
172        依赖的插件的信息
173    """
174    logger.debug(f"由于插件依赖，正在尝试加载插件 {plugin_name}")
175    for plugin in found_plugins:
176        if plugin["name"] == plugin_name:
177            if plugin not in plugins:
178                try:
179                    module, plugin_info = load_plugin(plugin)
180                    plugin["info"] = plugin_info
181                    plugin["plugin"] = module
182                    plugins.append(plugin)
183                except NotEnabledPluginException:
184                    logger.error(f"被依赖的插件 {plugin_name} 已被禁用，无法加载依赖")
185                    raise Exception(f"被依赖的插件 {plugin_name} 已被禁用，无法加载依赖")
186                except Exception as e:
187                    if ConfigManager.GlobalConfig().debug.save_dump:
188                        dump_path = save_exc_dump(f"尝试加载被依赖的插件 {plugin_name} 时失败！")
189                    else:
190                        dump_path = None
191                    logger.error(f"尝试加载被依赖的插件 {plugin_name} 时失败！ 原因:{repr(e)}\n"
192                                 f"{"".join(traceback.format_exc())}"
193                                 f"{f"\n已保存异常到 {dump_path}" if dump_path else ""}")
194                    raise e
195                logger.debug(f"由于插件依赖，插件 {plugin_name} 加载成功！")
196            else:
197                logger.debug(f"由于插件依赖，插件 {plugin_name} 已被加载，跳过加载")
198            return plugin
199    else:
200        raise FileNotFoundError(f"插件 {plugin_name} 不存在或不符合要求，无法加载依赖")
201
202
203# 该方法已被弃用
204def run_plugin_main(event_data):
205    """
206    运行插件的main函数
207    Args:
208        event_data: 事件数据
209    """
210    global has_main_func_plugins
211    for plugin in has_main_func_plugins:
212        logger.debug(f"执行插件: {plugin['name']}")
213        try:
214            plugin["plugin"].main(event_data, WORK_PATH)
215        except Exception as e:
216            logger.error(f"执行插件{plugin['name']}时发生错误: {repr(e)}")
217            continue
218
219
220@event_listener(EscalationEvent)
221def run_plugin_main_wrapper(event):
222    """
223    运行插件的main函数
224    Args:
225        event: 事件
226    """
227    run_plugin_main(event.event_data)
228
229
230def get_caller_plugin_data():
231    """
232    获取调用者的插件数据
233    :return:
234        plugin_data: dict | None
235    """
236
237    stack = inspect.stack()[1:]
238    for frame_info in stack:
239        filename = frame_info.filename
240
241        normalized_filename = os.path.normpath(filename)
242        normalized_plugins_path = os.path.normpath(PLUGINS_PATH)
243
244        if normalized_filename.startswith(normalized_plugins_path):
245            for plugin in found_plugins:
246                normalized_plugin_file_path = os.path.normpath(plugin["file_path"])
247                plugin_dir, plugin_file = os.path.split(normalized_plugin_file_path)
248
249                if plugin_dir == normalized_plugins_path:
250                    if normalized_plugin_file_path == normalized_filename:
251                        return plugin
252                else:
253                    if normalized_filename.startswith(plugin_dir):
254                        return plugin
255    return None