gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🤖 自动格式化代码 [skip ci]

Browse Source
This commit is contained in:
github-actions[bot]
2025-06-20 03:57:03 +00:00
parent 02088e18c2
commit 16df1661b1
2 changed files with 55 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
src/plugin_system/base/base_plugin.py
View File

@@ -306,10 +306,10 @@ class BasePlugin(ABC):
"""备份配置文件""" """备份配置文件"""
import shutil import shutil
import datetime import datetime
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = f"{config_file_path}.backup_{timestamp}" backup_path = f"{config_file_path}.backup_{timestamp}"
try: try:
shutil.copy2(config_file_path, backup_path) shutil.copy2(config_file_path, backup_path)
logger.info(f"{self.log_prefix} 配置文件已备份到: {backup_path}") logger.info(f"{self.log_prefix} 配置文件已备份到: {backup_path}")
@@ -320,26 +320,31 @@ class BasePlugin(ABC):
def _migrate_config_values(self, old_config: Dict[str, Any], new_config: Dict[str, Any]) -> Dict[str, Any]: def _migrate_config_values(self, old_config: Dict[str, Any], new_config: Dict[str, Any]) -> Dict[str, Any]:
"""将旧配置值迁移到新配置结构中 """将旧配置值迁移到新配置结构中
Args: Args:
old_config: 旧配置数据 old_config: 旧配置数据
new_config: 基于新schema生成的默认配置 new_config: 基于新schema生成的默认配置
Returns: Returns:
Dict[str, Any]: 迁移后的配置 Dict[str, Any]: 迁移后的配置
""" """
def migrate_section(old_section: Dict[str, Any], new_section: Dict[str, Any], section_name: str) -> 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() result = new_section.copy()
for key, value in old_section.items(): for key, value in old_section.items():
if key in new_section: if key in new_section:
# 特殊处理：config_version字段总是使用新版本 # 特殊处理：config_version字段总是使用新版本
if section_name == "plugin" and key == "config_version": if section_name == "plugin" and key == "config_version":
# 保持新的版本号，不迁移旧值 # 保持新的版本号，不迁移旧值
logger.debug(f"{self.log_prefix} 更新配置版本: {section_name}.{key} = {result[key]} (旧值: {value})") logger.debug(
f"{self.log_prefix} 更新配置版本: {section_name}.{key} = {result[key]} (旧值: {value})"
)
continue continue
# 键存在于新配置中，复制值 # 键存在于新配置中，复制值
if isinstance(value, dict) and isinstance(new_section[key], dict): if isinstance(value, dict) and isinstance(new_section[key], dict):
# 递归处理嵌套字典 # 递归处理嵌套字典
@@ -350,26 +355,32 @@ class BasePlugin(ABC):
else: else:
# 键在新配置中不存在，记录警告 # 键在新配置中不存在，记录警告
logger.warning(f"{self.log_prefix} 配置项 {section_name}.{key} 在新版本中已被移除") logger.warning(f"{self.log_prefix} 配置项 {section_name}.{key} 在新版本中已被移除")
return result return result
migrated_config = {} migrated_config = {}
# 迁移每个配置节 # 迁移每个配置节
for section_name, new_section_data in new_config.items(): 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): if (
migrated_config[section_name] = migrate_section(old_config[section_name], new_section_data, section_name) 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: else:
# 新增的节或类型不匹配，使用默认值 # 新增的节或类型不匹配，使用默认值
migrated_config[section_name] = new_section_data migrated_config[section_name] = new_section_data
if section_name in old_config: if section_name in old_config:
logger.warning(f"{self.log_prefix} 配置节 {section_name} 结构已改变，使用默认值") logger.warning(f"{self.log_prefix} 配置节 {section_name} 结构已改变，使用默认值")
# 检查旧配置中是否有新配置没有的节 # 检查旧配置中是否有新配置没有的节
for section_name in old_config.keys(): for section_name in old_config.keys():
if section_name not in migrated_config: if section_name not in migrated_config:
logger.warning(f"{self.log_prefix} 配置节 {section_name} 在新版本中已被移除") logger.warning(f"{self.log_prefix} 配置节 {section_name} 在新版本中已被移除")
return migrated_config return migrated_config
def _generate_config_from_schema(self) -> Dict[str, Any]: def _generate_config_from_schema(self) -> Dict[str, Any]:
@@ -378,19 +389,19 @@ class BasePlugin(ABC):
return {} return {}
config_data = {} config_data = {}
# 遍历每个配置节 # 遍历每个配置节
for section, fields in self.config_schema.items(): for section, fields in self.config_schema.items():
if isinstance(fields, dict): if isinstance(fields, dict):
section_data = {} section_data = {}
# 遍历节内的字段 # 遍历节内的字段
for field_name, field in fields.items(): for field_name, field in fields.items():
if isinstance(field, ConfigField): if isinstance(field, ConfigField):
section_data[field_name] = field.default section_data[field_name] = field.default
config_data[section] = section_data config_data[section] = section_data
return config_data return config_data
def _save_config_to_file(self, config_data: Dict[str, Any], config_file_path: str): def _save_config_to_file(self, config_data: Dict[str, Any], config_file_path: str):
@@ -402,7 +413,7 @@ class BasePlugin(ABC):
toml_str = f"# {self.plugin_name} - 配置文件\n" toml_str = f"# {self.plugin_name} - 配置文件\n"
plugin_description = self.get_manifest_info("description", "插件配置文件") plugin_description = self.get_manifest_info("description", "插件配置文件")
toml_str += f"# {plugin_description}\n" toml_str += f"# {plugin_description}\n"
# 获取当前期望的配置版本 # 获取当前期望的配置版本
expected_version = self._get_expected_config_version() expected_version = self._get_expected_config_version()
toml_str += f"# 配置版本: {expected_version}\n\n" toml_str += f"# 配置版本: {expected_version}\n\n"
@@ -418,7 +429,7 @@ class BasePlugin(ABC):
# 遍历节内的字段 # 遍历节内的字段
if isinstance(fields, dict) and section in config_data: if isinstance(fields, dict) and section in config_data:
section_data = config_data[section] section_data = config_data[section]
for field_name, field in fields.items(): for field_name, field in fields.items():
if isinstance(field, ConfigField): if isinstance(field, ConfigField):
# 添加字段描述 # 添加字段描述
@@ -502,31 +513,33 @@ class BasePlugin(ABC):
# 加载现有配置 # 加载现有配置
with open(config_file_path, "r", encoding="utf-8") as f: with open(config_file_path, "r", encoding="utf-8") as f:
existing_config = toml.load(f) or {} existing_config = toml.load(f) or {}
# 检查配置版本 # 检查配置版本
current_version = self._get_current_config_version(existing_config) current_version = self._get_current_config_version(existing_config)
# 如果配置文件没有版本信息，跳过版本检查 # 如果配置文件没有版本信息，跳过版本检查
if current_version == "0.0.0": if current_version == "0.0.0":
logger.debug(f"{self.log_prefix} 配置文件无版本信息，跳过版本检查") logger.debug(f"{self.log_prefix} 配置文件无版本信息，跳过版本检查")
self.config = existing_config self.config = existing_config
else: else:
expected_version = self._get_expected_config_version() expected_version = self._get_expected_config_version()
if current_version != expected_version: if current_version != expected_version:
logger.info(f"{self.log_prefix} 检测到配置版本需要更新: 当前=v{current_version}, 期望=v{expected_version}") logger.info(
f"{self.log_prefix} 检测到配置版本需要更新: 当前=v{current_version}, 期望=v{expected_version}"
)
# 生成新的默认配置结构 # 生成新的默认配置结构
new_config_structure = self._generate_config_from_schema() new_config_structure = self._generate_config_from_schema()
# 迁移旧配置值到新结构 # 迁移旧配置值到新结构
migrated_config = self._migrate_config_values(existing_config, new_config_structure) migrated_config = self._migrate_config_values(existing_config, new_config_structure)
# 保存迁移后的配置 # 保存迁移后的配置
self._save_config_to_file(migrated_config, config_file_path) self._save_config_to_file(migrated_config, config_file_path)
logger.info(f"{self.log_prefix} 配置文件已从 v{current_version} 更新到 v{expected_version}") logger.info(f"{self.log_prefix} 配置文件已从 v{current_version} 更新到 v{expected_version}")
self.config = migrated_config self.config = migrated_config
else: else:
logger.debug(f"{self.log_prefix} 配置版本匹配 (v{current_version})，直接加载") logger.debug(f"{self.log_prefix} 配置版本匹配 (v{current_version})，直接加载")

26
src/plugins/built_in/mute_plugin/plugin.py
View File

@@ -93,31 +93,31 @@ class MuteAction(BaseAction):
def _check_group_permission(self) -> Tuple[bool, Optional[str]]: def _check_group_permission(self) -> Tuple[bool, Optional[str]]:
"""检查当前群是否有禁言动作权限 """检查当前群是否有禁言动作权限
Returns: Returns:
Tuple[bool, Optional[str]]: (是否有权限, 错误信息) Tuple[bool, Optional[str]]: (是否有权限, 错误信息)
""" """
# 如果不是群聊，直接返回False # 如果不是群聊，直接返回False
if not self.is_group: if not self.is_group:
return False, "禁言动作只能在群聊中使用" return False, "禁言动作只能在群聊中使用"
# 获取权限配置 # 获取权限配置
allowed_groups = self.get_config("permissions.allowed_groups", []) allowed_groups = self.get_config("permissions.allowed_groups", [])
# 如果配置为空，表示不启用权限控制 # 如果配置为空，表示不启用权限控制
if not allowed_groups: if not allowed_groups:
logger.info(f"{self.log_prefix} 群组权限未配置，允许所有群使用禁言动作") logger.info(f"{self.log_prefix} 群组权限未配置，允许所有群使用禁言动作")
return True, None return True, None
# 检查当前群是否在允许列表中 # 检查当前群是否在允许列表中
current_group_key = f"{self.platform}:{self.group_id}" current_group_key = f"{self.platform}:{self.group_id}"
for allowed_group in allowed_groups: for allowed_group in allowed_groups:
if allowed_group == current_group_key: if allowed_group == current_group_key:
logger.info(f"{self.log_prefix} 群组 {current_group_key} 有禁言动作权限") logger.info(f"{self.log_prefix} 群组 {current_group_key} 有禁言动作权限")
return True, None return True, None
logger.warning(f"{self.log_prefix} 群组 {current_group_key} 没有禁言动作权限") logger.warning(f"{self.log_prefix} 群组 {current_group_key} 没有禁言动作权限")
return False, f"当前群组没有使用禁言动作的权限" return False, "当前群组没有使用禁言动作的权限"
async def execute(self) -> Tuple[bool, Optional[str]]: async def execute(self) -> Tuple[bool, Optional[str]]:
"""执行智能禁言判定""" """执行智能禁言判定"""
@@ -276,7 +276,7 @@ class MuteCommand(BaseCommand):
def _check_user_permission(self) -> Tuple[bool, Optional[str]]: def _check_user_permission(self) -> Tuple[bool, Optional[str]]:
"""检查当前用户是否有禁言命令权限 """检查当前用户是否有禁言命令权限
Returns: Returns:
Tuple[bool, Optional[str]]: (是否有权限, 错误信息) Tuple[bool, Optional[str]]: (是否有权限, 错误信息)
""" """
@@ -284,27 +284,27 @@ class MuteCommand(BaseCommand):
chat_stream = self.message.chat_stream chat_stream = self.message.chat_stream
if not chat_stream: if not chat_stream:
return False, "无法获取聊天流信息" return False, "无法获取聊天流信息"
current_platform = chat_stream.platform current_platform = chat_stream.platform
current_user_id = str(chat_stream.user_info.user_id) current_user_id = str(chat_stream.user_info.user_id)
# 获取权限配置 # 获取权限配置
allowed_users = self.get_config("permissions.allowed_users", []) allowed_users = self.get_config("permissions.allowed_users", [])
# 如果配置为空，表示不启用权限控制 # 如果配置为空，表示不启用权限控制
if not allowed_users: if not allowed_users:
logger.info(f"{self.log_prefix} 用户权限未配置，允许所有用户使用禁言命令") logger.info(f"{self.log_prefix} 用户权限未配置，允许所有用户使用禁言命令")
return True, None return True, None
# 检查当前用户是否在允许列表中 # 检查当前用户是否在允许列表中
current_user_key = f"{current_platform}:{current_user_id}" current_user_key = f"{current_platform}:{current_user_id}"
for allowed_user in allowed_users: for allowed_user in allowed_users:
if allowed_user == current_user_key: if allowed_user == current_user_key:
logger.info(f"{self.log_prefix} 用户 {current_user_key} 有禁言命令权限") logger.info(f"{self.log_prefix} 用户 {current_user_key} 有禁言命令权限")
return True, None return True, None
logger.warning(f"{self.log_prefix} 用户 {current_user_key} 没有禁言命令权限") logger.warning(f"{self.log_prefix} 用户 {current_user_key} 没有禁言命令权限")
return False, f"你没有使用禁言命令的权限" return False, "你没有使用禁言命令的权限"
async def execute(self) -> Tuple[bool, Optional[str]]: async def execute(self) -> Tuple[bool, Optional[str]]:
"""执行禁言命令""" """执行禁言命令"""