From 4eae2b8fe08135872f6081fd212a233ea5bb2b8b Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Fri, 15 Aug 2025 12:51:53 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AE=80=E5=8C=96MaiZone=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=8A=A0=E8=BD=BD=E5=99=A8=EF=BC=8C=E7=A7=BB?= =?UTF-8?q?=E9=99=A4=E5=86=97=E4=BD=99=E7=9A=84=E9=85=8D=E7=BD=AE=E9=AA=8C?= =?UTF-8?q?=E8=AF=81=E5=92=8C=E8=BF=81=E7=A7=BB=E9=80=BB=E8=BE=91=EF=BC=8C?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=BB=98=E8=AE=A4=E9=85=8D=E7=BD=AE=E5=88=9B?= =?UTF-8?q?=E5=BB=BA=E5=92=8C=E4=BF=9D=E5=AD=98=E5=8A=9F=E8=83=BD=EF=BC=8C?= =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E9=85=8D=E7=BD=AE=E9=A1=B9=E7=9A=84=E8=8E=B7?= =?UTF-8?q?=E5=8F=96=E5=92=8C=E8=AE=BE=E7=BD=AE=E6=96=B9=E6=B3=95=EF=BC=8C?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=8E=B7=E5=8F=96=E5=92=8C=E8=AE=BE=E7=BD=AE?= =?UTF-8?q?=E6=95=B4=E4=B8=AA=E9=85=8D=E7=BD=AE=E8=8A=82=E7=9A=84=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/plugins/built_in/maizone/config_loader.py | 448 ++++++------------ 1 file changed, 158 insertions(+), 290 deletions(-) diff --git a/src/plugins/built_in/maizone/config_loader.py b/src/plugins/built_in/maizone/config_loader.py index 37d932b8a..0a9652a80 100644 --- a/src/plugins/built_in/maizone/config_loader.py +++ b/src/plugins/built_in/maizone/config_loader.py @@ -1,15 +1,12 @@ """ -MaiZone插件独立配置文件加载系统 +MaiZone插件配置加载器 -这个模块提供了一个独立的配置文件加载系统，用于替代原本插件中的config加载系统。 -它支持TOML格式的配置文件，具有配置验证、默认值处理、类型转换等功能。 +简化的配置文件加载系统，专注于基本的配置文件读取和写入功能。 +支持TOML格式的配置文件，具有基本的类型转换和默认值处理。 """ import toml -import shutil -import datetime -from typing import Dict, Any, Union, List, Optional, Type -from dataclasses import dataclass, field +from typing import Dict, Any, Optional from pathlib import Path from src.common.logger import get_logger @@ -17,80 +14,8 @@ from src.common.logger import get_logger logger = get_logger("MaiZone.ConfigLoader") -@dataclass -class ConfigFieldSpec: - """配置字段规格定义""" - name: str - type_hint: Type - default: Any - description: str = "" - required: bool = False - choices: Optional[List[Any]] = None - min_value: Optional[Union[int, float]] = None - max_value: Optional[Union[int, float]] = None - - def validate_value(self, value: Any) -> tuple[bool, str]: - """验证配置值是否符合规格""" - # 类型检查 - if not isinstance(value, self.type_hint): - try: - # 尝试类型转换 - if self.type_hint == bool and isinstance(value, str): - value = value.lower() in ('true', '1', 'yes', 'on') - else: - value = self.type_hint(value) - except (ValueError, TypeError): - return False, f"类型错误: 期望 {self.type_hint.__name__}, 得到 {type(value).__name__}" - - # 选择项检查 - if self.choices and value not in self.choices: - return False, f"值不在允许范围内: {self.choices}" - - # 数值范围检查 - if isinstance(value, (int, float)): - if self.min_value is not None and value < self.min_value: - return False, f"值小于最小值 {self.min_value}" - if self.max_value is not None and value > self.max_value: - return False, f"值大于最大值 {self.max_value}" - - return True, "" - - -@dataclass -class ConfigSectionSpec: - """配置节规格定义""" - name: str - description: str = "" - fields: Dict[str, ConfigFieldSpec] = field(default_factory=dict) - - def add_field(self, field_spec: ConfigFieldSpec): - """添加字段规格""" - self.fields[field_spec.name] = field_spec - - def validate_section(self, section_data: Dict[str, Any]) -> tuple[bool, List[str]]: - """验证配置节数据""" - errors = [] - - # 检查必需字段 - for field_name, field_spec in self.fields.items(): - if field_spec.required and field_name not in section_data: - errors.append(f"缺少必需字段: {field_name}") - - # 验证每个字段 - for field_name, value in section_data.items(): - if field_name in self.fields: - field_spec = self.fields[field_name] - is_valid, error_msg = field_spec.validate_value(value) - if not is_valid: - errors.append(f"{field_name}: {error_msg}") - else: - logger.warning(f"未知配置字段: {self.name}.{field_name}") - - return len(errors) == 0, errors - - class MaiZoneConfigLoader: - """MaiZone插件独立配置加载器""" + """MaiZone插件配置加载器 - 简化版""" def __init__(self, plugin_dir: str, config_filename: str = "config.toml"): """ @@ -104,9 +29,9 @@ class MaiZoneConfigLoader: self.config_filename = config_filename self.config_file_path = self.plugin_dir / config_filename self.config_data: Dict[str, Any] = {} - self.config_specs: Dict[str, ConfigSectionSpec] = {} - self.config_version = "2.1.0" + # 确保插件目录存在 + self.plugin_dir.mkdir(parents=True, exist_ok=True) def load_config(self) -> bool: """ @@ -116,194 +41,71 @@ class MaiZoneConfigLoader: bool: 是否成功加载 """ try: - # 如果配置文件不存在，生成默认配置 + # 如果配置文件不存在，创建默认配置 if not self.config_file_path.exists(): - logger.info(f"配置文件不存在，生成默认配置: {self.config_file_path}") - self._generate_default_config() + logger.info(f"配置文件不存在，创建默认配置: {self.config_file_path}") + self._create_default_config() # 加载配置文件 with open(self.config_file_path, 'r', encoding='utf-8') as f: self.config_data = toml.load(f) logger.info(f"成功加载配置文件: {self.config_file_path}") - - # 验证配置 - self._validate_config() - - # 检查版本并迁移 - self._check_and_migrate_config() - return True except Exception as e: logger.error(f"加载配置文件失败: {e}") + # 如果加载失败，使用默认配置 + self.config_data = self._get_default_config() return False - def _generate_default_config(self): - """生成默认配置文件""" - try: - # 确保插件目录存在 - self.plugin_dir.mkdir(parents=True, exist_ok=True) - - # 生成默认配置数据 - default_config = {} - for section_name, section_spec in self.config_specs.items(): - section_data = {} - for field_name, field_spec in section_spec.fields.items(): - section_data[field_name] = field_spec.default - default_config[section_name] = section_data - - # 保存到文件 - self._save_config_to_file(default_config) - self.config_data = default_config - - logger.info(f"默认配置文件已生成: {self.config_file_path}") - - except Exception as e: - logger.error(f"生成默认配置文件失败: {e}") + def _create_default_config(self): + """创建默认配置文件""" + default_config = self._get_default_config() + self._save_config_to_file(default_config) + self.config_data = default_config + + def _get_default_config(self) -> Dict[str, Any]: + """获取默认配置""" + return { + "plugin": { + "enabled": True, + "name": "MaiZone", + "version": "2.1.0" + }, + "qzone": { + "qq": "", + "auto_login": True, + "check_interval": 300, + "max_retries": 3 + }, + "ai": { + "enabled": False, + "model": "gpt-3.5-turbo", + "max_tokens": 150, + "temperature": 0.7 + }, + "monitor": { + "enabled": False, + "keywords": [], + "check_friends": True, + "check_groups": False + }, + "scheduler": { + "enabled": False, + "schedules": [] + } + } def _save_config_to_file(self, config_data: Dict[str, Any]): - """保存配置到文件（带注释）""" - toml_content = "# MaiZone插件配置文件\n" - toml_content += "# 让你的麦麦发QQ空间说说、评论、点赞，支持AI配图、定时发送和自动监控功能\n" - toml_content += f"# 配置版本: {self.config_version}\n\n" - - for section_name, section_spec in self.config_specs.items(): - if section_name not in config_data: - continue - - # 添加节描述 - toml_content += f"# {section_spec.description}\n" - toml_content += f"[{section_name}]\n\n" - - section_data = config_data[section_name] - for field_name, field_spec in section_spec.fields.items(): - if field_name not in section_data: - continue - - # 添加字段描述 - toml_content += f"# {field_spec.description}\n" - if field_spec.choices: - toml_content += f"# 可选值: {', '.join(map(str, field_spec.choices))}\n" - if field_spec.min_value is not None or field_spec.max_value is not None: - range_str = "# 范围: " - if field_spec.min_value is not None: - range_str += f"最小值 {field_spec.min_value}" - if field_spec.max_value is not None: - if field_spec.min_value is not None: - range_str += f", 最大值 {field_spec.max_value}" - else: - range_str += f"最大值 {field_spec.max_value}" - toml_content += range_str + "\n" - - # 添加字段值 - value = section_data[field_name] - if isinstance(value, str): - toml_content += f'{field_name} = "{value}"\n' - elif isinstance(value, bool): - toml_content += 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) + "]" - elif all(isinstance(item, dict) for item in value): - # 处理字典列表（如schedules） - # 使用 TOML 内联表格式 - formatted_items = [] - for item in value: - # TOML 内联表中的字符串需要转义 - item_str = ", ".join([f'{k} = "{str(v)}"' for k, v in item.items()]) - formatted_items.append(f"{{ {item_str} }}") - formatted_list = "[\n " + ",\n ".join(formatted_items) + "\n]" - else: - formatted_list = str(value) - toml_content += f"{field_name} = {formatted_list}\n" - else: - toml_content += f"{field_name} = {value}\n" - - toml_content += "\n" - - toml_content += "\n" - - # 写入文件 - with open(self.config_file_path, 'w', encoding='utf-8') as f: - f.write(toml_content) - - def _validate_config(self) -> bool: - """验证配置数据""" - all_valid = True - - for section_name, section_spec in self.config_specs.items(): - if section_name not in self.config_data: - logger.warning(f"配置文件缺少节: {section_name}") - continue - - section_data = self.config_data[section_name] - is_valid, errors = section_spec.validate_section(section_data) - - if not is_valid: - logger.error(f"配置节 {section_name} 验证失败:") - for error in errors: - logger.error(f" - {error}") - all_valid = False - - return all_valid - - def _check_and_migrate_config(self): - """检查配置版本并进行迁移""" - current_version = self.get_config("plugin.config_version", "1.0.0") - - if current_version != self.config_version: - logger.info(f"检测到配置版本变更: {current_version} -> {self.config_version}") - - # 备份旧配置 - self._backup_config() - - # 迁移配置 - self._migrate_config(current_version, self.config_version) - - # 更新版本号 - self.config_data["plugin"]["config_version"] = self.config_version - - # 保存迁移后的配置 - self._save_config_to_file(self.config_data) - - logger.info(f"配置已迁移到版本 {self.config_version}") - - def _backup_config(self): - """备份当前配置文件""" - if self.config_file_path.exists(): - timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") - backup_path = self.config_file_path.with_suffix(f".backup_{timestamp}.toml") - shutil.copy2(self.config_file_path, backup_path) - logger.info(f"配置文件已备份到: {backup_path}") - - def _migrate_config(self, from_version: str, to_version: str): - """迁移配置数据""" - # 创建新的配置结构 - new_config = {} - - for section_name, section_spec in self.config_specs.items(): - new_section = {} - - # 复制现有配置值 - if section_name in self.config_data: - old_section = self.config_data[section_name] - for field_name, field_spec in section_spec.fields.items(): - if field_name in old_section: - new_section[field_name] = old_section[field_name] - else: - new_section[field_name] = field_spec.default - logger.info(f"添加新配置项: {section_name}.{field_name} = {field_spec.default}") - else: - # 新增节，使用默认值 - for field_name, field_spec in section_spec.fields.items(): - new_section[field_name] = field_spec.default - logger.info(f"添加新配置节: {section_name}") - - new_config[section_name] = new_section - - self.config_data = new_config + """保存配置到文件""" + try: + with open(self.config_file_path, 'w', encoding='utf-8') as f: + toml.dump(config_data, f) + logger.debug(f"配置已保存到: {self.config_file_path}") + except Exception as e: + logger.error(f"保存配置文件失败: {e}") + raise def get_config(self, key: str, default: Any = None) -> Any: """ @@ -316,23 +118,30 @@ class MaiZoneConfigLoader: Returns: Any: 配置值或默认值 """ + if not self.config_data: + logger.warning("配置数据为空，返回默认值") + return default + keys = key.split('.') current = self.config_data - for k in keys: - if isinstance(current, dict) and k in current: - current = current[k] - else: - return default - - return current + try: + for k in keys: + if isinstance(current, dict) and k in current: + current = current[k] + else: + return default + return current + except Exception as e: + logger.warning(f"获取配置失败 {key}: {e}") + return default def set_config(self, key: str, value: Any) -> bool: """ 设置配置值 Args: - key: 配置键名 + key: 配置键名，格式为 "section.field" value: 配置值 Returns: @@ -340,36 +149,23 @@ class MaiZoneConfigLoader: """ try: keys = key.split('.') - if len(keys) != 2: + if len(keys) < 2: logger.error(f"配置键格式错误: {key}，应为 'section.field' 格式") return False - section_name, field_name = keys + # 获取或创建嵌套字典结构 + current = self.config_data + for k in keys[:-1]: + if k not in current: + current[k] = {} + elif not isinstance(current[k], dict): + logger.error(f"配置路径冲突: {k} 不是字典类型") + return False + current = current[k] - # 检查节是否存在 - if section_name not in self.config_specs: - logger.error(f"未知配置节: {section_name}") - return False - - # 检查字段是否存在 - if field_name not in self.config_specs[section_name].fields: - logger.error(f"未知配置字段: {key}") - return False - - # 验证值 - field_spec = self.config_specs[section_name].fields[field_name] - is_valid, error_msg = field_spec.validate_value(value) - if not is_valid: - logger.error(f"配置值验证失败 {key}: {error_msg}") - return False - - # 设置值 - if section_name not in self.config_data: - self.config_data[section_name] = {} - - self.config_data[section_name][field_name] = value + # 设置最终值 + current[keys[-1]] = value logger.debug(f"设置配置: {key} = {value}") - return True except Exception as e: @@ -400,6 +196,64 @@ class MaiZoneConfigLoader: """ return self.load_config() + def get_section(self, section_name: str) -> Optional[Dict[str, Any]]: + """ + 获取整个配置节 + + Args: + section_name: 配置节名称 + + Returns: + Optional[Dict[str, Any]]: 配置节数据或None + """ + return self.config_data.get(section_name) + + def set_section(self, section_name: str, section_data: Dict[str, Any]) -> bool: + """ + 设置整个配置节 + + Args: + section_name: 配置节名称 + section_data: 配置节数据 + + Returns: + bool: 是否设置成功 + """ + try: + if not isinstance(section_data, dict): + logger.error(f"配置节数据必须为字典类型: {section_name}") + return False + + self.config_data[section_name] = section_data + logger.debug(f"设置配置节: {section_name}") + return True + except Exception as e: + logger.error(f"设置配置节失败 {section_name}: {e}") + return False + + def has_config(self, key: str) -> bool: + """ + 检查配置项是否存在 + + Args: + key: 配置键名 + + Returns: + bool: 配置项是否存在 + """ + keys = key.split('.') + current = self.config_data + + try: + for k in keys: + if isinstance(current, dict) and k in current: + current = current[k] + else: + return False + return True + except Exception: + return False + def get_config_info(self) -> Dict[str, Any]: """ 获取配置信息 @@ -409,7 +263,21 @@ class MaiZoneConfigLoader: """ return { "config_file": str(self.config_file_path), - "config_version": self.config_version, - "sections": list(self.config_specs.keys()), + "config_exists": self.config_file_path.exists(), + "sections": list(self.config_data.keys()) if self.config_data else [], "loaded": bool(self.config_data) } + + def reset_to_default(self) -> bool: + """ + 重置为默认配置 + + Returns: + bool: 是否重置成功 + """ + try: + self.config_data = self._get_default_config() + return self.save_config() + except Exception as e: + logger.error(f"重置配置失败: {e}") + return False