fix:自动更新插件配置文件版本
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user