"""插件热插拔管理:运行时加载/卸载/重载插件(仅超管) 通过直接导入模块并自行维护注册表,绕开 NoneBot PluginManager 的 重复检查限制(load_plugin 每次都会向全局 _managers 追加 manager, 重复加载会触发 "Plugin already exists" 错误)。 """ import importlib import sys from types import ModuleType from nonebot import get_driver, logger, on_command, on_message, require from nonebot.adapters.onebot.v11 import Bot, Message, MessageEvent from nonebot.internal.matcher import matchers as matchers_registry from nonebot.params import CommandArg from nonebot.permission import SUPERUSER from nonebot.plugin import ( Plugin, _plugins, get_loaded_plugins, get_plugin_by_module_name, ) from nonebot.rule import Rule, to_me require("nonebot_plugin_alconna") from nonebot_plugin_alconna import UniMessage from hexi.core.message_utils import send_forward_msg from . import plugin_control # 插件热拔插管理（hexi.core 核心模块，非 NoneBot 插件） def _forward_sender_name() -> str: """合并转发节点的发送者昵称(用机器人配置昵称,插件名写进 content)""" nickname = get_driver().config.nickname return next(iter(nickname), "Bot") if nickname else "Bot" def _match_help(event: MessageEvent) -> bool: """匹配 help/帮助/菜单 开头(支持 at bot 后跟命令,可带插件名参数) on_command 的命令规则只匹配第一段为文本的消息, "at bot + help" 第一段是 at 段无法触发,故用 on_message 自定义规则。 """ text = event.message.extract_plain_text().strip() return text in ("help", "帮助", "菜单") or text.startswith( ("help ", "帮助 ", "菜单 ") ) help_cmd = on_message(rule=Rule(_match_help) & to_me(), block=True) plugin_list_cmd = on_command("插件列表", permission=SUPERUSER, block=True) plugin_load_cmd = on_command("加载插件", permission=SUPERUSER, block=True) plugin_unload_cmd = on_command("卸载插件", permission=SUPERUSER, block=True) plugin_reload_cmd = on_command("重载插件", permission=SUPERUSER, block=True) # 自管理的热插拔插件注册表:模块名 -> 模块 _hot_plugins: dict[str, ModuleType] = {} def _registered_matchers(module_name: str) -> list: """获取某模块注册的全部 matcher(含子模块,注册表按优先级分组)""" return [ m for group in matchers_registry.values() for m in group if getattr(m, "module_name", None) and ( m.module_name == module_name or m.module_name.startswith(module_name + ".") ) ] def _collect_module_names(plugin: Plugin) -> list[str]: """收集插件及其全部子插件的模块名(用于查 matcher 与清缓存)""" names = [plugin.module_name] for sub in plugin.sub_plugins: names.extend(_collect_module_names(sub)) return names def _unregister_plugin(plugin: Plugin) -> None: """从 NoneBot 插件注册表移除插件(递归移除子插件)""" for sub in list(plugin.sub_plugins): _unregister_plugin(sub) _plugins.pop(plugin.id_, None) if parent := plugin.parent_plugin: parent.sub_plugins.discard(plugin) def _is_library(name: str) -> bool: """判断模块是否归属「非 application」插件（library / 未声明类别）。 只有 application 插件可热拔插；library（含根包未声明类别但被依赖的核心/框架） 都应保护，热拔插会破坏依赖它的插件运行态。 优先级：NoneBot 插件注册表 > sys.modules 回溯（含父包 __plugin_meta__）。 """ plugin = get_plugin_by_module_name(name) if plugin is not None: meta = plugin.metadata return not (meta and meta.type == "application") parts = name.split(".") for i in range(len(parts), 0, -1): module = sys.modules.get(".".join(parts[:i])) if module is not None: meta = getattr(module, "__plugin_meta__", None) if meta is not None: return meta.type != "application" return False def hot_load(name: str) -> bool: """热加载插件模块(直接 import,matcher 随之注册)""" if _is_library(name): logger.warning(f"插件 {name} 是非 application（library/未声明）插件，不支持热加载") return False if name in _hot_plugins: return False if any(p.module_name == name for p in get_loaded_plugins()): return False importlib.import_module(name) _hot_plugins[name] = sys.modules[name] logger.info(f"插件已热加载: {name}") try: plugin_control.instrument_plugin_gate() except Exception as e: # noqa: BLE001 logger.warning(f"热加载后注入 gateway 失败: {type(e).__name__}: {e}") return True def hot_unload(name: str) -> bool: """热卸载插件:移除 matcher、清理 NoneBot 注册表与模块缓存 不限于热加载的插件:启动时加载的插件(hexi/plugins 与 pyproject.toml 声明)同样支持卸载,以 NoneBot 插件注册表为准查找。 安全限制：非 application 插件（library/未声明类别，常为核心库/框架）被其他 插件依赖，热卸载会让依赖方引用悬空并导致运行混乱，故拒绝卸载。 """ if _is_library(name): logger.warning(f"插件 {name} 是非 application（library/未声明）插件，热卸载会使依赖它的插件运行混乱，已拒绝") return False plugin = get_plugin_by_module_name(name) if plugin is None and name not in _hot_plugins: return False module_names = _collect_module_names(plugin) if plugin else [name] for mod_name in module_names: for m in _registered_matchers(mod_name): group = matchers_registry.get(m.priority) if group and m in group: group.remove(m) if plugin is not None: _unregister_plugin(plugin) # 清理模块缓存(含子模块),保证重载拿到新模块 for mod_name in [ n for n in sys.modules if any(n == base or n.startswith(base + ".") for base in module_names) ]: sys.modules.pop(mod_name, None) # 清理自管理注册表(含以子模块名登记的情况) for key in list(_hot_plugins): if any(key == base or key.startswith(base + ".") for base in module_names): _hot_plugins.pop(key, None) logger.info(f"插件已热卸载: {name}") return True def hot_reload(name: str) -> bool: """热重载插件:先卸载再加载""" if _is_library(name): logger.warning(f"插件 {name} 是非 application（library/未声明）插件，热重载会使依赖它的插件运行混乱，已拒绝") return False hot_unload(name) return hot_load(name) def _plugin_owner(mod_name: str) -> tuple[str, object] | None: """找到模块归属的插件(自身或回溯父包直到有 meta),返回 (归属模块名, meta)""" parts = mod_name.split(".") for i in range(len(parts), 0, -1): cand = ".".join(parts[:i]) # 优先 NoneBot 注册表中的 metadata for p in get_loaded_plugins(): if p.module_name == cand: meta = p.metadata break else: module = sys.modules.get(cand) meta = getattr(module, "__plugin_meta__", None) if module else None if meta is not None: return cand, meta return None def _visible_plugins() -> list[tuple[str, object]]: """可显示的应用型插件列表 [(模块名, meta), ...] 以全局 matcher 注册表为准(正规插件、hexi_core 子模块、热加载插件一网打尽), 子模块 matcher 回溯到所属插件包,仅保留 type='application',排除 alconna 等 library。 """ seen: dict[str, object] = {} for group in matchers_registry.values(): for m in group: mod = getattr(m, "module_name", None) if not mod: continue owner = _plugin_owner(mod) if owner and owner[0] not in seen: seen[owner[0]] = owner[1] result = [] for mod in sorted(seen): meta = seen[mod] if meta.type != "application": continue result.append((mod, meta)) return result def _find_plugin(name: str): """按 模块名/meta名(含模糊) 查找应用型插件的 meta""" for mod, meta in _visible_plugins(): if mod == name or meta.name == name or name in meta.name: return meta return None @help_cmd.handle() async def help_handle(bot: Bot, ev: MessageEvent): text = ev.message.extract_plain_text().strip() # 提取插件名参数 arg = "" for kw in ("help", "帮助", "菜单"): if text.startswith(kw): arg = text[len(kw) :].strip() break if arg: # 单插件详细帮助 meta = _find_plugin(arg) if meta is None: await UniMessage.text( f"未找到插件「{arg}」,发送「帮助」查看所有插件" ).send() return content = f"【{meta.name}】{meta.description}" if meta.usage: content += f"\n{meta.usage}" await send_forward_msg( bot, ev, [(meta.name, content)], content, sender_name=_forward_sender_name(), ) return # 总览(仅显示有触发命令的应用型插件,内容以插件名开头) entries: list[tuple[str, str]] = [] fallback_lines = ["===== 插件菜单 ====="] for _, meta in _visible_plugins(): content = f"【{meta.name}】{meta.description}" entries.append((meta.name, content)) fallback_lines.append(content) entries.append(("提示", "发送「帮助 插件名」查看插件详细用法")) fallback_lines.append("发送「帮助 插件名」查看插件详细用法") await send_forward_msg( bot, ev, entries, "\n".join(fallback_lines), sender_name=_forward_sender_name(), ) @plugin_list_cmd.handle() async def plugin_list_handle(bot: Bot, ev: MessageEvent): entries: list[tuple[str, str]] = [] fallback_lines = ["已加载插件:"] for mod, meta in _visible_plugins(): content = f"{meta.name}\n{mod}" entries.append((meta.name, content)) fallback_lines.append(content) await send_forward_msg( bot, ev, entries, "\n".join(fallback_lines), sender_name=_forward_sender_name(), ) @plugin_load_cmd.handle() async def plugin_load_handle(ev: MessageEvent, arg: Message = CommandArg()): name = arg.extract_plain_text().strip() if not name: await UniMessage.text( "用法:加载插件 <模块名>,如 hexi.plugins.nonebot_plugin_dice" ).send() return try: if hot_load(name): await UniMessage.text(f"插件 {name} 加载成功").send() else: await UniMessage.text(f"插件 {name} 已加载或模块不存在").send() except Exception as e: logger.error(f"加载插件 {name} 失败", exc_info=True) await UniMessage.text(f"加载失败: {type(e).__name__}: {e}").send() @plugin_unload_cmd.handle() async def plugin_unload_handle(ev: MessageEvent, arg: Message = CommandArg()): name = arg.extract_plain_text().strip() if not name: await UniMessage.text( "用法:卸载插件 <模块名>,如 hexi.plugins.nonebot_plugin_dice" ).send() return if hot_unload(name): await UniMessage.text(f"插件 {name} 已卸载").send() else: await UniMessage.text(f"未找到插件 {name}(未加载或模块名有误)").send() @plugin_reload_cmd.handle() async def plugin_reload_handle(ev: MessageEvent, arg: Message = CommandArg()): name = arg.extract_plain_text().strip() if not name: await UniMessage.text( "用法:重载插件 <模块名>,如 hexi.plugins.nonebot_plugin_dice" ).send() return try: if hot_reload(name): await UniMessage.text(f"插件 {name} 重载成功").send() else: await UniMessage.text(f"插件 {name} 重载失败(模块不存在)").send() except Exception as e: logger.error(f"重载插件 {name} 失败", exc_info=True) await UniMessage.text(f"重载失败: {type(e).__name__}: {e}").send()