""" 插件管理插件 提供插件的热重载、启用、禁用等管理功能 支持的指令： /插件列表 - 查看所有插件状态 /重载插件 <名称> - 重载指定插件 /重载所有插件 - 重载所有插件 /启用插件 <名称> - 启用指定插件 /禁用插件 <名称> - 禁用指定插件 /刷新插件 - 扫描并发现新插件 /插件帮助 - 显示帮助信息 """ import sys import tomllib from pathlib import Path from loguru import logger from utils.plugin_base import PluginBase from utils.decorators import on_text_message from utils.plugin_manager import PluginManager class ManagePlugin(PluginBase): """插件管理插件""" # 插件元数据 description = "插件管理，支持热重载、启用、禁用" author = "ShiHao" version = "2.0.0" # 最高加载优先级，确保最先加载 load_priority = 100 def __init__(self): super().__init__() self.config = None self.admins = [] async def async_init(self): """插件异步初始化""" # 从主配置文件读取管理员列表 with open("main_config.toml", "rb") as f: main_config = tomllib.load(f) self.admins = main_config.get("Bot", {}).get("admins", []) logger.info(f"插件管理插件已加载，管理员: {self.admins}") def _check_admin(self, message: dict) -> bool: """检查是否是管理员""" sender_wxid = message.get("SenderWxid", "") from_wxid = message.get("FromWxid", "") is_group = message.get("IsGroup", False) # 私聊时 sender_wxid 可能为空，使用 from_wxid user_wxid = sender_wxid if is_group else from_wxid return user_wxid in self.admins @on_text_message(priority=99) async def handle_command(self, bot, message: dict): """处理管理命令""" content = message.get("Content", "").strip() from_wxid = message.get("FromWxid", "") # 检查权限 if not self._check_admin(message): return True # 继续传递给其他插件 # 插件帮助 if content == "/插件帮助" or content == "/plugin help": await self._show_help(bot, from_wxid) return False # 插件列表 elif content == "/插件列表" or content == "/plugins": await self._list_plugins(bot, from_wxid) return False # 重载所有插件 elif content == "/重载所有插件" or content == "/reload all": await self._reload_all_plugins(bot, from_wxid) return False # 重载插件 elif content.startswith("/重载插件 ") or content.startswith("/reload "): plugin_name = content.split(maxsplit=1)[1].strip() await self._reload_plugin(bot, from_wxid, plugin_name) return False # 启用插件 elif content.startswith("/启用插件 ") or content.startswith("/enable "): plugin_name = content.split(maxsplit=1)[1].strip() await self._enable_plugin(bot, from_wxid, plugin_name) return False # 禁用插件 elif content.startswith("/禁用插件 ") or content.startswith("/disable "): plugin_name = content.split(maxsplit=1)[1].strip() await self._disable_plugin(bot, from_wxid, plugin_name) return False # 刷新插件（发现新插件） elif content == "/刷新插件" or content == "/refresh": await self._refresh_plugins(bot, from_wxid) return False # 加载新插件（从目录加载全新插件） elif content.startswith("/加载插件 ") or content.startswith("/load "): plugin_name = content.split(maxsplit=1)[1].strip() await self._load_new_plugin(bot, from_wxid, plugin_name) return False return True # 不是管理命令，继续传递 async def _list_plugins(self, bot, to_wxid: str): """列出所有插件""" try: pm = PluginManager() plugins = pm.get_plugin_info() if not plugins: await bot.send_text(to_wxid, "暂无插件") return lines = ["📦 插件列表\n"] for plugin in plugins: status = "✅ 已启用" if plugin.get("enabled", False) else "❌ 已禁用" lines.append( f"{status} {plugin.get('name', 'Unknown')}\n" f" 版本: {plugin.get('version', 'Unknown')}\n" f" 作者: {plugin.get('author', 'Unknown')}\n" ) response = "\n".join(lines) await bot.send_text(to_wxid, response) logger.info(f"已发送插件列表给 {to_wxid}") except Exception as e: logger.error(f"列出插件失败: {e}") await bot.send_text(to_wxid, f"❌ 获取插件列表失败: {e}") async def _reload_plugin(self, bot, to_wxid: str, plugin_name: str): """重载插件""" pm = PluginManager() if plugin_name == "ManagePlugin": await bot.send_text(to_wxid, "❌ ManagePlugin 不能被重载") return success = await pm.reload_plugin(plugin_name) if success: await bot.send_text(to_wxid, f"✅ 插件 {plugin_name} 重载成功") logger.info(f"插件 {plugin_name} 已被重载") else: await bot.send_text(to_wxid, f"❌ 插件 {plugin_name} 重载失败") async def _enable_plugin(self, bot, to_wxid: str, plugin_name: str): """启用插件""" pm = PluginManager() # 检查插件是否存在 plugin_info = pm.get_plugin_info(plugin_name) if not plugin_info: await bot.send_text(to_wxid, f"❌ 插件 {plugin_name} 不存在") return # 检查是否已启用 if plugin_info["enabled"]: await bot.send_text(to_wxid, f"ℹ️ 插件 {plugin_name} 已经是启用状态") return # 加载插件 success = await pm.load_plugin(plugin_name) if success: await bot.send_text(to_wxid, f"✅ 插件 {plugin_name} 已启用") logger.info(f"插件 {plugin_name} 已被启用") else: await bot.send_text(to_wxid, f"❌ 插件 {plugin_name} 启用失败") async def _disable_plugin(self, bot, to_wxid: str, plugin_name: str): """禁用插件""" pm = PluginManager() if plugin_name == "ManagePlugin": await bot.send_text(to_wxid, "❌ ManagePlugin 不能被禁用") return # 检查插件是否存在 plugin_info = pm.get_plugin_info(plugin_name) if not plugin_info: await bot.send_text(to_wxid, f"❌ 插件 {plugin_name} 不存在") return # 检查是否已禁用 if not plugin_info["enabled"]: await bot.send_text(to_wxid, f"ℹ️ 插件 {plugin_name} 已经是禁用状态") return # 卸载插件 success = await pm.unload_plugin(plugin_name) if success: await bot.send_text(to_wxid, f"✅ 插件 {plugin_name} 已禁用") logger.info(f"插件 {plugin_name} 已被禁用") else: await bot.send_text(to_wxid, f"❌ 插件 {plugin_name} 禁用失败") async def _show_help(self, bot, to_wxid: str): """显示帮助信息""" help_text = """📦 插件管理帮助 /插件列表 - 查看所有插件状态 /插件帮助 - 显示此帮助信息 /加载插件 <名称> - 加载新插件（无需重启） /重载插件 <名称> - 热重载指定插件 /重载所有插件 - 热重载所有插件 /启用插件 <名称> - 启用已禁用的插件 /禁用插件 <名称> - 禁用指定插件 /刷新插件 - 扫描发现新插件 示例： /加载插件 NewPlugin /重载插件 AIChat /禁用插件 Weather""" await bot.send_text(to_wxid, help_text) async def _reload_all_plugins(self, bot, to_wxid: str): """重载所有插件""" pm = PluginManager() await bot.send_text(to_wxid, "⏳ 正在重载所有插件...") try: # 清理插件相关的模块缓存 modules_to_remove = [ name for name in sys.modules.keys() if name.startswith('plugins.') and 'ManagePlugin' not in name ] for module_name in modules_to_remove: del sys.modules[module_name] # 重载所有插件 reloaded = await pm.reload_plugins() if reloaded: await bot.send_text( to_wxid, f"✅ 重载完成\n已加载 {len(reloaded)} 个插件:\n" + "\n".join(f" • {name}" for name in reloaded) ) logger.success(f"已重载 {len(reloaded)} 个插件") else: await bot.send_text(to_wxid, "⚠️ 没有插件被重载") except Exception as e: logger.error(f"重载所有插件失败: {e}") await bot.send_text(to_wxid, f"❌ 重载失败: {e}") async def _refresh_plugins(self, bot, to_wxid: str): """刷新插件列表，发现新插件""" pm = PluginManager() try: # 记录刷新前的插件数量 old_count = len(pm.plugin_info) # 刷新插件列表 await pm.refresh_plugins() # 计算新发现的插件 new_count = len(pm.plugin_info) new_plugins = new_count - old_count if new_plugins > 0: # 获取新发现的插件名称 new_plugin_names = [ info["name"] for info in pm.plugin_info.values() if not info.get("enabled", False) ][-new_plugins:] await bot.send_text( to_wxid, f"✅ 发现 {new_plugins} 个新插件:\n" + "\n".join(f" • {name}" for name in new_plugin_names) + "\n\n使用 /启用插件 <名称> 来启用" ) logger.info(f"发现 {new_plugins} 个新插件: {new_plugin_names}") else: await bot.send_text(to_wxid, "ℹ️ 没有发现新插件") except Exception as e: logger.error(f"刷新插件失败: {e}") await bot.send_text(to_wxid, f"❌ 刷新失败: {e}") async def _load_new_plugin(self, bot, to_wxid: str, plugin_name: str): """加载全新的插件（支持插件类名或目录名）""" import os import importlib import inspect pm = PluginManager() # 检查是否已加载 if plugin_name in pm.plugins: await bot.send_text(to_wxid, f"ℹ️ 插件 {plugin_name} 已经加载，如需重载请使用 /重载插件") return try: # 尝试查找插件 found = False plugin_class = None actual_plugin_name = None for dirname in os.listdir("plugins"): dirpath = f"plugins/{dirname}" if not os.path.isdir(dirpath) or not os.path.exists(f"{dirpath}/main.py"): continue # 支持通过目录名或类名查找 if dirname == plugin_name or dirname.lower() == plugin_name.lower(): # 通过目录名匹配 module_name = f"plugins.{dirname}.main" # 清理旧的模块缓存 if module_name in sys.modules: del sys.modules[module_name] # 导入模块 module = importlib.import_module(module_name) # 查找插件类 for name, obj in inspect.getmembers(module): if (inspect.isclass(obj) and issubclass(obj, PluginBase) and obj != PluginBase): plugin_class = obj actual_plugin_name = obj.__name__ found = True break if found: break # 尝试通过类名匹配 try: module_name = f"plugins.{dirname}.main" if module_name in sys.modules: del sys.modules[module_name] module = importlib.import_module(module_name) for name, obj in inspect.getmembers(module): if (inspect.isclass(obj) and issubclass(obj, PluginBase) and obj != PluginBase and (obj.__name__ == plugin_name or obj.__name__.lower() == plugin_name.lower())): plugin_class = obj actual_plugin_name = obj.__name__ found = True break if found: break except: continue if not found or not plugin_class: await bot.send_text( to_wxid, f"❌ 未找到插件 {plugin_name}\n" f"请确认：\n" f"1. plugins/{plugin_name}/main.py 存在\n" f"2. main.py 中有继承 PluginBase 的类" ) return # 检查是否已加载（用实际类名再检查一次） if actual_plugin_name in pm.plugins: await bot.send_text(to_wxid, f"ℹ️ 插件 {actual_plugin_name} 已经加载") return # 加载插件 success = await pm._load_plugin_class(plugin_class) if success: await bot.send_text( to_wxid, f"✅ 插件加载成功\n" f"名称: {actual_plugin_name}\n" f"版本: {plugin_class.version}\n" f"作者: {plugin_class.author}" ) logger.success(f"新插件 {actual_plugin_name} 已加载") else: await bot.send_text(to_wxid, f"❌ 插件 {actual_plugin_name} 加载失败") except Exception as e: import traceback logger.error(f"加载新插件失败: {e}\n{traceback.format_exc()}") await bot.send_text(to_wxid, f"❌ 加载失败: {e}")