diff --git a/plugins/take_picture_plugin/plugin.py b/plugins/take_picture_plugin/plugin.py index 1a84a824a..5be4bf438 100644 --- a/plugins/take_picture_plugin/plugin.py +++ b/plugins/take_picture_plugin/plugin.py @@ -456,12 +456,7 @@ class TakePicturePlugin(BasePlugin): # 配置Schema定义 config_schema = { "plugin": { - "name": ConfigField(type=str, default="take_picture_plugin", description="插件名称", required=True), - "version": ConfigField(type=str, default="1.0.0", description="插件版本号"), "enabled": ConfigField(type=bool, default=False, description="是否启用插件"), - "description": ConfigField( - type=str, default="提供生成自拍照和展示最近照片的功能", description="插件描述", required=True - ), }, "api": { "base_url": ConfigField( diff --git a/src/plugin_system/base/base_plugin.py b/src/plugin_system/base/base_plugin.py index 179c6c0bf..ca80f9ce0 100644 --- a/src/plugin_system/base/base_plugin.py +++ b/src/plugin_system/base/base_plugin.py @@ -237,7 +237,8 @@ class BasePlugin(ABC): return toml_str = f"# {self.plugin_name} - 自动生成的配置文件\n" - toml_str += f"# {self.plugin_description}\n\n" + plugin_description = self.get_manifest_info("description", "插件配置文件") + toml_str += f"# {plugin_description}\n\n" # 遍历每个配置节 for section, fields in self.config_schema.items(): @@ -285,8 +286,184 @@ class BasePlugin(ABC): except IOError as e: logger.error(f"{self.log_prefix} 保存默认配置文件失败: {e}", exc_info=True) + def _get_expected_config_version(self) -> str: + """获取插件期望的配置版本号""" + # 从config_schema的plugin.config_version字段获取 + if "plugin" in self.config_schema and isinstance(self.config_schema["plugin"], dict): + config_version_field = self.config_schema["plugin"].get("config_version") + if isinstance(config_version_field, ConfigField): + return config_version_field.default + return "1.0.0" + + def _get_current_config_version(self, config: Dict[str, Any]) -> str: + """从配置文件中获取当前版本号""" + if "plugin" in config and "config_version" in config["plugin"]: + return str(config["plugin"]["config_version"]) + # 如果没有config_version字段,视为最早的版本 + return "0.0.0" + + def _backup_config_file(self, config_file_path: str) -> str: + """备份配置文件""" + import shutil + import datetime + + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + backup_path = f"{config_file_path}.backup_{timestamp}" + + try: + shutil.copy2(config_file_path, backup_path) + logger.info(f"{self.log_prefix} 配置文件已备份到: {backup_path}") + return backup_path + except Exception as e: + logger.error(f"{self.log_prefix} 备份配置文件失败: {e}") + return "" + + def _migrate_config_values(self, old_config: Dict[str, Any], new_config: Dict[str, Any]) -> Dict[str, Any]: + """将旧配置值迁移到新配置结构中 + + Args: + old_config: 旧配置数据 + new_config: 基于新schema生成的默认配置 + + Returns: + Dict[str, Any]: 迁移后的配置 + """ + def migrate_section(old_section: Dict[str, Any], new_section: Dict[str, Any], section_name: str) -> Dict[str, Any]: + """迁移单个配置节""" + result = new_section.copy() + + for key, value in old_section.items(): + if key in new_section: + # 特殊处理:config_version字段总是使用新版本 + if section_name == "plugin" and key == "config_version": + # 保持新的版本号,不迁移旧值 + logger.debug(f"{self.log_prefix} 更新配置版本: {section_name}.{key} = {result[key]} (旧值: {value})") + continue + + # 键存在于新配置中,复制值 + if isinstance(value, dict) and isinstance(new_section[key], dict): + # 递归处理嵌套字典 + result[key] = migrate_section(value, new_section[key], f"{section_name}.{key}") + else: + result[key] = value + logger.debug(f"{self.log_prefix} 迁移配置: {section_name}.{key} = {value}") + else: + # 键在新配置中不存在,记录警告 + logger.warning(f"{self.log_prefix} 配置项 {section_name}.{key} 在新版本中已被移除") + + return result + + migrated_config = {} + + # 迁移每个配置节 + for section_name, new_section_data in new_config.items(): + if section_name in old_config and isinstance(old_config[section_name], dict) and isinstance(new_section_data, dict): + migrated_config[section_name] = migrate_section(old_config[section_name], new_section_data, section_name) + else: + # 新增的节或类型不匹配,使用默认值 + migrated_config[section_name] = new_section_data + if section_name in old_config: + logger.warning(f"{self.log_prefix} 配置节 {section_name} 结构已改变,使用默认值") + + # 检查旧配置中是否有新配置没有的节 + for section_name in old_config.keys(): + if section_name not in migrated_config: + logger.warning(f"{self.log_prefix} 配置节 {section_name} 在新版本中已被移除") + + return migrated_config + + def _generate_config_from_schema(self) -> Dict[str, Any]: + """根据schema生成配置数据结构(不写入文件)""" + if not self.config_schema: + return {} + + config_data = {} + + # 遍历每个配置节 + for section, fields in self.config_schema.items(): + if isinstance(fields, dict): + section_data = {} + + # 遍历节内的字段 + for field_name, field in fields.items(): + if isinstance(field, ConfigField): + section_data[field_name] = field.default + + config_data[section] = section_data + + return config_data + + def _save_config_to_file(self, config_data: Dict[str, Any], config_file_path: str): + """将配置数据保存为TOML文件(包含注释)""" + if not self.config_schema: + logger.debug(f"{self.log_prefix} 插件未定义config_schema,不生成配置文件") + return + + toml_str = f"# {self.plugin_name} - 配置文件\n" + plugin_description = self.get_manifest_info("description", "插件配置文件") + toml_str += f"# {plugin_description}\n" + + # 获取当前期望的配置版本 + expected_version = self._get_expected_config_version() + toml_str += f"# 配置版本: {expected_version}\n\n" + + # 遍历每个配置节 + for section, fields in self.config_schema.items(): + # 添加节描述 + if section in self.config_section_descriptions: + toml_str += f"# {self.config_section_descriptions[section]}\n" + + toml_str += f"[{section}]\n\n" + + # 遍历节内的字段 + if isinstance(fields, dict) and section in config_data: + section_data = config_data[section] + + for field_name, field in fields.items(): + if isinstance(field, ConfigField): + # 添加字段描述 + toml_str += f"# {field.description}" + if field.required: + toml_str += " (必需)" + toml_str += "\n" + + # 如果有示例值,添加示例 + if field.example: + toml_str += f"# 示例: {field.example}\n" + + # 如果有可选值,添加说明 + if field.choices: + choices_str = ", ".join(map(str, field.choices)) + toml_str += f"# 可选值: {choices_str}\n" + + # 添加字段值(使用迁移后的值) + value = section_data.get(field_name, field.default) + if isinstance(value, str): + toml_str += f'{field_name} = "{value}"\n' + elif isinstance(value, bool): + toml_str += f"{field_name} = {str(value).lower()}\n" + elif isinstance(value, list): + # 格式化列表 + if all(isinstance(item, str) for item in value): + formatted_list = "[" + ", ".join(f'"{item}"' for item in value) + "]" + else: + formatted_list = str(value) + toml_str += f"{field_name} = {formatted_list}\n" + else: + toml_str += f"{field_name} = {value}\n" + + toml_str += "\n" + toml_str += "\n" + + try: + with open(config_file_path, "w", encoding="utf-8") as f: + f.write(toml_str) + logger.info(f"{self.log_prefix} 配置文件已保存: {config_file_path}") + except IOError as e: + logger.error(f"{self.log_prefix} 保存配置文件失败: {e}", exc_info=True) + def _load_plugin_config(self): - """加载插件配置文件""" + """加载插件配置文件,支持版本检查和自动迁移""" if not self.config_file_name: logger.debug(f"{self.log_prefix} 未指定配置文件,跳过加载") return @@ -310,6 +487,7 @@ class BasePlugin(ABC): config_file_path = os.path.join(plugin_dir, self.config_file_name) + # 如果配置文件不存在,生成默认配置 if not os.path.exists(config_file_path): logger.info(f"{self.log_prefix} 配置文件 {config_file_path} 不存在,将生成默认配置。") self._generate_and_save_default_config(config_file_path) @@ -321,8 +499,39 @@ class BasePlugin(ABC): file_ext = os.path.splitext(self.config_file_name)[1].lower() if file_ext == ".toml": + # 加载现有配置 with open(config_file_path, "r", encoding="utf-8") as f: - self.config = toml.load(f) or {} + existing_config = toml.load(f) or {} + + # 检查配置版本 + current_version = self._get_current_config_version(existing_config) + + # 如果配置文件没有版本信息,跳过版本检查 + if current_version == "0.0.0": + logger.debug(f"{self.log_prefix} 配置文件无版本信息,跳过版本检查") + self.config = existing_config + else: + expected_version = self._get_expected_config_version() + + if current_version != expected_version: + logger.info(f"{self.log_prefix} 检测到配置版本需要更新: 当前=v{current_version}, 期望=v{expected_version}") + + # 生成新的默认配置结构 + new_config_structure = self._generate_config_from_schema() + + # 迁移旧配置值到新结构 + migrated_config = self._migrate_config_values(existing_config, new_config_structure) + + # 保存迁移后的配置 + self._save_config_to_file(migrated_config, config_file_path) + + logger.info(f"{self.log_prefix} 配置文件已从 v{current_version} 更新到 v{expected_version}") + + self.config = migrated_config + else: + logger.debug(f"{self.log_prefix} 配置版本匹配 (v{current_version}),直接加载") + self.config = existing_config + logger.debug(f"{self.log_prefix} 配置已从 {config_file_path} 加载") # 从配置中更新 enable_plugin diff --git a/src/plugins/built_in/mute_plugin/_manifest.json b/src/plugins/built_in/mute_plugin/_manifest.json index 32f848aeb..b8d919560 100644 --- a/src/plugins/built_in/mute_plugin/_manifest.json +++ b/src/plugins/built_in/mute_plugin/_manifest.json @@ -1,7 +1,7 @@ { "manifest_version": 1, "name": "群聊禁言管理插件 (Mute Plugin)", - "version": "2.0.0", + "version": "3.0.0", "description": "群聊禁言管理插件,提供智能禁言功能", "author": { "name": "MaiBot开发团队", diff --git a/src/plugins/built_in/mute_plugin/plugin.py b/src/plugins/built_in/mute_plugin/plugin.py index 3f5dc35de..72e5637f1 100644 --- a/src/plugins/built_in/mute_plugin/plugin.py +++ b/src/plugins/built_in/mute_plugin/plugin.py @@ -9,10 +9,11 @@ - 模板化消息:支持自定义禁言提示消息 - 参数验证:完整的输入参数验证和错误处理 - 配置文件支持:所有设置可通过配置文件调整 +- 权限管理:支持用户权限和群组权限控制 包含组件: -- 智能禁言Action - 基于LLM判断是否需要禁言 -- 禁言命令Command - 手动执行禁言操作 +- 智能禁言Action - 基于LLM判断是否需要禁言(支持群组权限控制) +- 禁言命令Command - 手动执行禁言操作(支持用户权限控制) """ from typing import List, Tuple, Type, Optional @@ -90,10 +91,45 @@ class MuteAction(BaseAction): # 关联类型 associated_types = ["text", "command"] + def _check_group_permission(self) -> Tuple[bool, Optional[str]]: + """检查当前群是否有禁言动作权限 + + Returns: + Tuple[bool, Optional[str]]: (是否有权限, 错误信息) + """ + # 如果不是群聊,直接返回False + if not self.is_group: + return False, "禁言动作只能在群聊中使用" + + # 获取权限配置 + allowed_groups = self.get_config("permissions.allowed_groups", []) + + # 如果配置为空,表示不启用权限控制 + if not allowed_groups: + logger.info(f"{self.log_prefix} 群组权限未配置,允许所有群使用禁言动作") + return True, None + + # 检查当前群是否在允许列表中 + current_group_key = f"{self.platform}:{self.group_id}" + for allowed_group in allowed_groups: + if allowed_group == current_group_key: + logger.info(f"{self.log_prefix} 群组 {current_group_key} 有禁言动作权限") + return True, None + + logger.warning(f"{self.log_prefix} 群组 {current_group_key} 没有禁言动作权限") + return False, f"当前群组没有使用禁言动作的权限" + async def execute(self) -> Tuple[bool, Optional[str]]: """执行智能禁言判定""" logger.info(f"{self.log_prefix} 执行智能禁言动作") + # 首先检查群组权限 + has_permission, permission_error = self._check_group_permission() + if not has_permission: + logger.error(f"{self.log_prefix} 权限检查失败: {permission_error}") + # 不发送错误消息,静默拒绝 + return False, permission_error + # 获取参数 target = self.action_data.get("target") duration = self.action_data.get("duration") @@ -238,9 +274,48 @@ class MuteCommand(BaseCommand): command_examples = ["/mute 用户名 300", "/mute 张三 600 刷屏", "/mute @某人 1800 违规内容"] intercept_message = True # 拦截消息处理 + def _check_user_permission(self) -> Tuple[bool, Optional[str]]: + """检查当前用户是否有禁言命令权限 + + Returns: + Tuple[bool, Optional[str]]: (是否有权限, 错误信息) + """ + # 获取当前用户信息 + chat_stream = self.message.chat_stream + if not chat_stream: + return False, "无法获取聊天流信息" + + current_platform = chat_stream.platform + current_user_id = str(chat_stream.user_info.user_id) + + # 获取权限配置 + allowed_users = self.get_config("permissions.allowed_users", []) + + # 如果配置为空,表示不启用权限控制 + if not allowed_users: + logger.info(f"{self.log_prefix} 用户权限未配置,允许所有用户使用禁言命令") + return True, None + + # 检查当前用户是否在允许列表中 + current_user_key = f"{current_platform}:{current_user_id}" + for allowed_user in allowed_users: + if allowed_user == current_user_key: + logger.info(f"{self.log_prefix} 用户 {current_user_key} 有禁言命令权限") + return True, None + + logger.warning(f"{self.log_prefix} 用户 {current_user_key} 没有禁言命令权限") + return False, f"你没有使用禁言命令的权限" + async def execute(self) -> Tuple[bool, Optional[str]]: """执行禁言命令""" try: + # 首先检查用户权限 + has_permission, permission_error = self._check_user_permission() + if not has_permission: + logger.error(f"{self.log_prefix} 权限检查失败: {permission_error}") + await self.send_text(f"❌ {permission_error}") + return False, permission_error + target = self.matched_groups.get("target") duration = self.matched_groups.get("duration") reason = self.matched_groups.get("reason", "管理员操作") @@ -352,8 +427,8 @@ class MutePlugin(BasePlugin): """禁言插件 提供智能禁言功能: - - 智能禁言Action:基于LLM判断是否需要禁言 - - 禁言命令Command:手动执行禁言操作 + - 智能禁言Action:基于LLM判断是否需要禁言(支持群组权限控制) + - 禁言命令Command:手动执行禁言操作(支持用户权限控制) """ # 插件基本信息 @@ -365,6 +440,7 @@ class MutePlugin(BasePlugin): config_section_descriptions = { "plugin": "插件基本信息配置", "components": "组件启用控制", + "permissions": "权限管理配置", "mute": "核心禁言功能配置", "smart_mute": "智能禁言Action的专属配置", "mute_command": "禁言命令Command的专属配置", @@ -374,17 +450,25 @@ class MutePlugin(BasePlugin): # 配置Schema定义 config_schema = { "plugin": { - "name": ConfigField(type=str, default="mute_plugin", description="插件名称", required=True), - "version": ConfigField(type=str, default="2.0.0", description="插件版本号"), "enabled": ConfigField(type=bool, default=False, description="是否启用插件"), - "description": ConfigField( - type=str, default="群聊禁言管理插件,提供智能禁言功能", description="插件描述", required=True - ), + "config_version": ConfigField(type=str, default="0.0.2", description="配置文件版本"), }, "components": { "enable_smart_mute": ConfigField(type=bool, default=True, description="是否启用智能禁言Action"), "enable_mute_command": ConfigField(type=bool, default=False, description="是否启用禁言命令Command"), }, + "permissions": { + "allowed_users": ConfigField( + type=list, + default=[], + description="允许使用禁言命令的用户列表,格式:['platform:user_id'],如['qq:123456789']。空列表表示不启用权限控制", + ), + "allowed_groups": ConfigField( + type=list, + default=[], + description="允许使用禁言动作的群组列表,格式:['platform:group_id'],如['qq:987654321']。空列表表示不启用权限控制", + ), + }, "mute": { "min_duration": ConfigField(type=int, default=60, description="最短禁言时长(秒)"), "max_duration": ConfigField(type=int, default=2592000, description="最长禁言时长(秒),默认30天"),