diff --git a/plugins/plugin_manager/__init__.py b/plugins/plugin_manager/__init__.py new file mode 100644 index 0000000..e3ce26c --- /dev/null +++ b/plugins/plugin_manager/__init__.py @@ -0,0 +1,7 @@ +# 从当前包的main模块导入PluginManagerPlugin类 +from .main import PluginManagerPlugin + +# 提供get_plugin函数,返回插件实例 +def get_plugin(): + """获取插件实例""" + return PluginManagerPlugin() \ No newline at end of file diff --git a/plugins/plugin_manager/config.toml b/plugins/plugin_manager/config.toml new file mode 100644 index 0000000..889dcec --- /dev/null +++ b/plugins/plugin_manager/config.toml @@ -0,0 +1,14 @@ +[PluginManager] +enable = true +command = ["插件", "plugin", "插件管理"] +command-format = """ +-----Bot----- +🔧插件管理指令: +插件 列表 - 查看所有插件 +插件 启用 [插件名] - 启用插件 +插件 禁用 [插件名] - 禁用插件 +插件 重载 [插件名] - 重新加载插件 +插件 卸载 [插件名] - 卸载插件 +插件 加载 [插件名] - 加载插件 +插件 信息 [插件名] - 查看插件详情 +""" \ No newline at end of file diff --git a/plugins/plugin_manager/main.py b/plugins/plugin_manager/main.py new file mode 100644 index 0000000..886d6bc --- /dev/null +++ b/plugins/plugin_manager/main.py @@ -0,0 +1,327 @@ +import logging +import os +import sys +from typing import Dict, Any, List, Optional, Tuple + +from plugin_common.message_plugin_interface import MessagePluginInterface +from plugin_common.plugin_interface import PluginStatus +from plugin_common.plugin_registry import PluginRegistry +from plugin_common.plugin_manager import PluginManager +from plugins.stats_collector.decorators import plugin_stats_decorator +from robot_cmd.robot_command import GroupBotManager + + +class PluginManagerPlugin(MessagePluginInterface): + """插件管理插件""" + + @property + def name(self) -> str: + return "插件管理" + + @property + def version(self) -> str: + return "1.0.0" + + @property + def description(self) -> str: + return "提供插件的启用、禁用、重载、卸载和加载功能,实现插件热更新" + + @property + def author(self) -> str: + return "Trae AI" + + @property + def command_prefix(self) -> Optional[str]: + return "" # 不需要前缀,直接匹配命令 + + @property + def commands(self) -> List[str]: + return self._commands + + def __init__(self): + super().__init__() + self.plugin_manager = None + + def initialize(self, context: Dict[str, Any]) -> bool: + """初始化插件""" + self.LOG = logging.getLogger(f"Plugin.{self.name}") + self.LOG.info(f"正在初始化 {self.name} 插件...") + + # 保存上下文对象 + self.wcf = context.get("wcf") + self.event_system = context.get("event_system") + self.message_util = context.get("message_util") + + # 保存插件注册表引用,用于管理插件 + self.plugin_registry: PluginRegistry = context.get("plugin_registry") + if not self.plugin_registry: + self.LOG.error("无法获取插件注册表,插件管理功能将无法正常工作") + return False + + # 创建插件管理器实例 + self.plugin_manager = PluginManager() + # 设置系统上下文 + self.plugin_manager.set_system_context(context) + + self._commands = self._config.get("PluginManager", {}).get("command", ["插件", "plugin", "插件管理"]) + self.command_format = self._config.get("PluginManager", {}).get("command-format", "插件 列表") + self.enable = self._config.get("PluginManager", {}).get("enable", True) + + self.LOG.info(f"[{self.name}] 插件初始化完成,指令:{self._commands}") + return True + + def start(self) -> bool: + """启动插件""" + self.LOG.info(f"[{self.name}] 插件已启动") + self.status = PluginStatus.RUNNING + return True + + def stop(self) -> bool: + """停止插件""" + self.LOG.info(f"[{self.name}] 插件已停止") + self.status = PluginStatus.STOPPED + return True + + def can_process(self, message: Dict[str, Any]) -> bool: + """检查是否可以处理该消息""" + if not self.enable: + return False + + content = str(message.get("content", "")).strip() + command = content.split(" ")[0] + + return command in self._commands + + def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]: + """处理消息""" + content = str(message.get("content", "")).strip() + self.LOG.info(f"插件执行: {self.name}:{content}") + + sender = message.get("sender") + roomid = message.get("roomid", "") + wcf = message.get("wcf") + gbm = message.get("gbm") + + # 检查命令格式 + parts = content.split(" ") + if len(parts) == 1: + wcf.send_text(f"-----Bot-----\n❌命令格式错误!\n{self.command_format}", + (roomid if roomid else sender), sender) + return True, "命令格式错误" + + # 检查权限 (只允许管理员操作) + if not self._is_admin(sender, gbm): + wcf.send_text(f"-----Bot-----\n❌权限不足,只有管理员可以管理插件", + (roomid if roomid else sender), sender) + return True, "权限不足" + + # 解析子命令 + sub_command = parts[1] + plugin_name = parts[2] if len(parts) > 2 else "" + + try: + # 根据子命令执行相应操作 + if sub_command == "列表": + return self._list_plugins(wcf, sender, roomid) + elif sub_command == "启用" and plugin_name: + return self._enable_plugin(plugin_name, wcf, sender, roomid) + elif sub_command == "禁用" and plugin_name: + return self._disable_plugin(plugin_name, wcf, sender, roomid) + elif sub_command == "重载" and plugin_name: + return self._reload_plugin(plugin_name, wcf, sender, roomid) + elif sub_command == "卸载" and plugin_name: + return self._unload_plugin(plugin_name, wcf, sender, roomid) + elif sub_command == "加载" and plugin_name: + return self._load_plugin(plugin_name, wcf, sender, roomid) + elif sub_command == "信息" and plugin_name: + return self._plugin_info(plugin_name, wcf, sender, roomid) + else: + wcf.send_text(f"-----Bot-----\n❌未知命令或缺少参数!\n{self.command_format}", + (roomid if roomid else sender), sender) + return True, "未知命令" + except Exception as e: + self.LOG.error(f"处理插件管理请求出错: {e}") + wcf.send_text(f"-----Bot-----\n❌操作失败: {str(e)}", + (roomid if roomid else sender), sender) + return True, f"处理出错: {e}" + + def _is_admin(self, user_id: str, gbm: GroupBotManager) -> bool: + """检查用户是否为管理员""" + # 从配置中获取管理员列表 + admin_list = gbm.get_admin_list() if gbm else [] + return user_id in admin_list + + def _list_plugins(self, wcf, sender: str, roomid: str) -> Tuple[bool, str]: + """列出所有插件""" + plugins = self.plugin_registry.get_all_plugins().values() + + # 构建插件列表消息 + message = "-----Bot-----\n📋 插件列表:\n" + for plugin in plugins: + status = "✅ 已启用" if plugin.status == PluginStatus.RUNNING else "❌ 已禁用" + message += f"{plugin.name} v{plugin.version} - {status}\n" + + wcf.send_text(message, (roomid if roomid else sender), sender) + return True, "列出插件成功" + + def _enable_plugin(self, plugin_name: str, wcf, sender: str, roomid: str) -> Tuple[bool, str]: + """启用插件""" + plugin = self.plugin_registry.get_plugin(plugin_name) + if not plugin: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 不存在", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 不存在" + + if plugin.status == PluginStatus.RUNNING: + wcf.send_text(f"-----Bot-----\n⚠️插件 {plugin_name} 已经是启用状态", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 已经是启用状态" + + # 使用插件管理器启动插件 + success = self.plugin_manager.start_plugin(plugin_name) + if success: + wcf.send_text(f"-----Bot-----\n✅插件 {plugin_name} 启用成功", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 启用成功" + else: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 启用失败", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 启用失败" + + def _disable_plugin(self, plugin_name: str, wcf, sender: str, roomid: str) -> Tuple[bool, str]: + """禁用插件""" + # 不允许禁用自身 + if plugin_name == self.name: + wcf.send_text(f"-----Bot-----\n⚠️不能禁用插件管理插件自身", + (roomid if roomid else sender), sender) + return True, "不能禁用插件管理插件自身" + + plugin = self.plugin_registry.get_plugin(plugin_name) + if not plugin: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 不存在", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 不存在" + + if plugin.status == PluginStatus.STOPPED: + wcf.send_text(f"-----Bot-----\n⚠️插件 {plugin_name} 已经是禁用状态", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 已经是禁用状态" + + # 使用插件管理器停止插件 + success = self.plugin_manager.stop_plugin(plugin_name) + if success: + wcf.send_text(f"-----Bot-----\n✅插件 {plugin_name} 禁用成功", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 禁用成功" + else: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 禁用失败", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 禁用失败" + + def _reload_plugin(self, plugin_name: str, wcf, sender: str, roomid: str) -> Tuple[bool, str]: + """重新加载插件""" + # 不允许重载自身 + if plugin_name == self.name: + wcf.send_text(f"-----Bot-----\n⚠️不能重载插件管理插件自身", + (roomid if roomid else sender), sender) + return True, "不能重载插件管理插件自身" + + # 使用插件管理器重新加载插件 + plugin = self.plugin_manager.reload_plugin(plugin_name) + if plugin: + wcf.send_text(f"-----Bot-----\n✅插件 {plugin_name} 重载成功", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 重载成功" + else: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 重载失败", + (roomid if roomid else sender), sender) + return False, f"插件 {plugin_name} 重载失败" + + def _unload_plugin(self, plugin_name: str, wcf, sender: str, roomid: str, silent: bool = False) -> Tuple[bool, str]: + """卸载插件""" + # 不允许卸载自身 + if plugin_name == self.name: + if not silent: + wcf.send_text(f"-----Bot-----\n⚠️不能卸载插件管理插件自身", + (roomid if roomid else sender), sender) + return True, "不能卸载插件管理插件自身" + + plugin = self.plugin_registry.get_plugin(plugin_name) + if not plugin: + if not silent: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 不存在或已卸载", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 不存在或已卸载" + + # 使用插件管理器卸载插件 + success = self.plugin_manager.unload_plugin(plugin_name) + if success: + if not silent: + wcf.send_text(f"-----Bot-----\n✅插件 {plugin_name} 卸载成功", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 卸载成功" + else: + if not silent: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 卸载失败", + (roomid if roomid else sender), sender) + return False, f"插件 {plugin_name} 卸载失败" + + def _load_plugin(self, plugin_name: str, wcf, sender: str, roomid: str, silent: bool = False) -> Tuple[bool, str]: + """加载插件""" + # 检查插件是否已加载 + existing_plugin = self.plugin_registry.get_plugin(plugin_name) + if existing_plugin: + if not silent: + wcf.send_text(f"-----Bot-----\n⚠️插件 {plugin_name} 已经加载", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 已经加载" + + try: + # 检查插件目录是否存在 + plugin_dir = os.path.join("plugins", plugin_name) + if not os.path.exists(plugin_dir): + if not silent: + wcf.send_text(f"-----Bot-----\n❌插件目录 {plugin_dir} 不存在", + (roomid if roomid else sender), sender) + return False, f"插件目录 {plugin_dir} 不存在" + + # 使用插件管理器加载插件 + plugin = self.plugin_manager.load_plugin(plugin_name) + if plugin: + if not silent: + wcf.send_text(f"-----Bot-----\n✅插件 {plugin_name} 加载成功", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 加载成功" + else: + if not silent: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 加载失败", + (roomid if roomid else sender), sender) + return False, f"插件 {plugin_name} 加载失败" + except Exception as e: + self.LOG.error(f"加载插件 {plugin_name} 出错: {e}") + if not silent: + wcf.send_text(f"-----Bot-----\n❌加载插件出错: {str(e)}", + (roomid if roomid else sender), sender) + return False, f"加载插件出错: {e}" + + def _plugin_info(self, plugin_name: str, wcf, sender: str, roomid: str) -> Tuple[bool, str]: + """查看插件详情""" + plugin = self.plugin_registry.get_plugin(plugin_name) + if not plugin: + wcf.send_text(f"-----Bot-----\n❌插件 {plugin_name} 不存在", + (roomid if roomid else sender), sender) + return True, f"插件 {plugin_name} 不存在" + + # 构建插件详情消息 + status_text = "✅ 已启用" if plugin.status == PluginStatus.RUNNING else "❌ 已禁用" + message = f"""-----Bot----- +📦 插件详情:{plugin.name} +📝 描述:{plugin.description} +🔢 版本:{plugin.version} +👤 作者:{plugin.author} +⚙️ 状态:{status_text} +🔑 命令:{', '.join(plugin.commands) if hasattr(plugin, 'commands') else '无'} +""" + + wcf.send_text(message, (roomid if roomid else sender), sender) + return True, "查看插件详情成功" \ No newline at end of file diff --git a/robot_cmd/robot_command.py b/robot_cmd/robot_command.py index d6bf108..d82dc75 100644 --- a/robot_cmd/robot_command.py +++ b/robot_cmd/robot_command.py @@ -6,6 +6,8 @@ # 3.群AI能力 #启用群AI #关闭群AI # 4.群总结能力 #启用群总结 #关闭群总结 # 5.sehuatang PDF能力 #启用pdf #关闭pdf +from typing import List + import redis import json from enum import Enum @@ -147,12 +149,12 @@ class GroupBotManager: # 如果是GROUP_LIST指令,返回 group:list 清单 if command_str.strip().upper() == "群列表": return GroupBotManager.get_group_list() - + # 如果是清除群指令 if command_str.strip().startswith("清除群-"): target_group_id = command_str.strip().split("-")[1] return GroupBotManager.remove_group(target_group_id) - + if len(command_parts) < 2: return None @@ -241,18 +243,40 @@ class GroupBotManager: # 检查群是否在列表中 if group_id not in GroupBotManager.local_cache["group_list"]: return f"群 {group_id} 不在机器人管理列表中" - + # 从本地缓存中移除群组 GroupBotManager.local_cache["group_list"].remove(group_id) if group_id in GroupBotManager.local_cache["group_permissions"]: del GroupBotManager.local_cache["group_permissions"][group_id] - + # 从Redis中移除群组 r.srem("group:list", group_id) r.delete(f'group:{group_id}:permissions') - + return f"已成功清除群 {group_id} 的所有设置" + @staticmethod + def get_admin_list() -> List[str]: + """获取管理员列表 + + 返回系统管理员的微信ID列表 + """ + # 从配置文件中获取管理员列表 + config_admin_list = [] # self.config.get("admin_list", []) + + # 手动添加的管理员ID列表 + manual_admin_list = [ + "Jyunere", # 示例ID,请替换为实际的微信ID + "wxid_abcdef", # 示例ID,请替换为实际的微信ID + "filehelper" # 文件传输助手,方便自己测试 + ] + + # 合并所有管理员列表并去重 + all_admin_list = list(set(config_admin_list + manual_admin_list)) + + return all_admin_list + + # 示例命令 def simulate_commands(): # 加载本地缓存