""" 数据迁移工具 用于处理旧版本配置文件到新版本的迁移 按需调用，不影响正常运行性能 """ import os import json import csv class MigrationTool: """数据迁移工具类""" def __init__(self, plugin_dir, user_base_dir, logger=None): """ 初始化迁移工具 参数: plugin_dir: 插件目录路径 user_base_dir: 用户配置基础目录 logger: 日志函数（可选） """ self.plugin_dir = plugin_dir self.user_base_dir = user_base_dir if logger: self._log_func = logger else: def default_logger(msg): from .common import _ANSI_CLEAR_EOL print(f"\r{_ANSI_CLEAR_EOL}{msg}", flush=True) self._log_func = default_logger # 定义路径 self.legacy_config_dir = os.path.join(plugin_dir, "config") self.config_dir = os.path.join(user_base_dir, "config") self.tags_dir = os.path.join(user_base_dir, "tags") self.rules_dir = os.path.join(user_base_dir, "rules") def _log(self, msg: str): """统一日志调用层""" self._log_func(msg) # ---版本比对工具--- def _compare_versions(self, v1: str, v2: str) -> int: """ 比较两个版本号 返回: 1: v1 > v2 0: v1 == v2 -1: v1 < v2 """ def parse(v): return [int(x) for x in str(v).split('.')] p1, p2 = parse(v1), parse(v2) # 补齐长度 max_len = max(len(p1), len(p2)) p1.extend([0] * (max_len - len(p1))) p2.extend([0] * (max_len - len(p2))) for a, b in zip(p1, p2): if a > b: return 1 if a < b: return -1 return 0 # ---config.json 专用迁移--- def ensure_config_json_exists(self, file_path: str, default_data: dict, legacy_path: str = None) -> bool: """ 确保 config.json 存在（专用迁移逻辑） 逻辑: 1. 文件存在 → 跳过 2. 文件不存在 + 旧文件存在 → 提取 API Key 和模型信息，映射到新服务商 3. 文件不存在 + 无旧文件 → 创建默认配置 """ if os.path.exists(file_path): return False os.makedirs(os.path.dirname(file_path), exist_ok=True) # 检查旧版本文件 if legacy_path and os.path.exists(legacy_path): try: with open(legacy_path, 'r', encoding='utf-8') as f: legacy_data = json.load(f) # 执行专用迁移：提取 API Key 和模型信息 migrated_data = self._migrate_config_api_keys_to_services(legacy_data, default_data) self._save_with_version(file_path, migrated_data, default_data) self._log("[config.json] 从插件目录迁移旧文件完成（已提取 API Key 和模型信息）") return True except Exception as e: self._log(f"[config.json] 迁移旧文件失败: {str(e)}，使用默认配置") # 创建默认配置 self._log("[config.json] 文件不存在，创建默认配置...") self._save_with_version(file_path, default_data, default_data) return True def _migrate_config_api_keys_to_services(self, legacy_data: dict, default_data: dict) -> dict: """ 从旧配置提取 API Key 和模型信息，映射到新配置的服务商 逻辑: 1. 遍历旧配置中有 API Key 的服务 2. 在新配置中查找对应的服务商 ID 3. 如果找到，填入 API Key 和模型信息 4. 如果找不到，创建新服务商 """ import copy result = copy.deepcopy(default_data) # 提取旧配置中的服务信息 legacy_services = legacy_data.get('model_services', []) if not legacy_services: # 兼容更旧的格式（llm/vlm providers） legacy_services = self._extract_legacy_providers(legacy_data) # 构建新配置服务商的 ID 映射 new_services = result.get('model_services', []) service_id_map = {s.get('id'): i for i, s in enumerate(new_services)} for legacy_service in legacy_services: api_key = legacy_service.get('api_key', '').strip() if not api_key: continue service_id = legacy_service.get('id', '') if service_id in service_id_map: # 服务商存在，更新 API Key 和模型信息 idx = service_id_map[service_id] new_services[idx]['api_key'] = api_key # 迁移模型信息 for model_type in ['llm_models', 'vlm_models']: legacy_models = legacy_service.get(model_type, []) if legacy_models: new_services[idx][model_type] = legacy_models # 迁移其他配置 for key in ['base_url', 'auto_unload', 'disable_thinking', 'enable_advanced_params', 'filter_thinking_output']: if key in legacy_service: new_services[idx][key] = legacy_service[key] self._log(f"[config.json] 迁移服务商: {service_id}") else: # 服务商不存在，创建新服务商 new_service = { 'id': service_id, 'type': legacy_service.get('type', 'openai_compatible'), 'name': legacy_service.get('name', service_id), 'description': legacy_service.get('description', f'{service_id}（从旧版迁移）'), 'base_url': legacy_service.get('base_url', ''), 'api_key': api_key, 'disable_thinking': legacy_service.get('disable_thinking', True), 'enable_advanced_params': legacy_service.get('enable_advanced_params', True), 'filter_thinking_output': legacy_service.get('filter_thinking_output', True), 'llm_models': legacy_service.get('llm_models', []), 'vlm_models': legacy_service.get('vlm_models', []) } new_services.append(new_service) self._log(f"[config.json] 创建新服务商: {service_id}") # 迁移 current_services if 'current_services' in legacy_data: result['current_services'] = legacy_data['current_services'] # 迁移百度翻译配置 if 'baidu_translate' in legacy_data: result['baidu_translate'] = legacy_data['baidu_translate'] result['model_services'] = new_services return result def _extract_legacy_providers(self, legacy_data: dict) -> list: """ 从更旧的配置格式（llm/vlm providers）提取服务信息 兼容 v1.0 格式: { "llm": {"providers": {"zhipu": {...}, "custom": {...}}}, "vlm": {"providers": {"zhipu": {...}}} } """ services = [] provider_map = {} # 用于合并同一个 provider 的 llm 和 vlm 配置 for service_type in ['llm', 'vlm']: if service_type not in legacy_data: continue providers = legacy_data[service_type].get('providers', {}) for provider_name, provider_config in providers.items(): api_key = provider_config.get('api_key', '').strip() if not api_key: continue if provider_name not in provider_map: provider_map[provider_name] = { 'id': provider_name, 'type': 'openai_compatible', 'name': provider_config.get('name', provider_name), 'base_url': provider_config.get('base_url', ''), 'api_key': api_key, 'llm_models': [], 'vlm_models': [] } # 添加模型信息 model_name = provider_config.get('model', '') if model_name: models_key = f'{service_type}_models' provider_map[provider_name][models_key].append({ 'name': model_name, 'display_name': '', 'is_default': True, 'temperature': provider_config.get('temperature', 0.7), 'max_tokens': provider_config.get('max_tokens', 1000), 'top_p': provider_config.get('top_p', 0.9) }) services = list(provider_map.values()) return services # ---统一保存方法--- def _save_with_version(self, file_path: str, data: dict, default_data: dict) -> bool: """ 保存配置文件，自动处理版本号 参数: file_path: 目标文件路径 data: 要保存的数据 default_data: 默认配置（用于获取版本号） 返回: bool: 保存成功返回 True """ try: # 确保版本号存在且在开头 version = data.get('__config_version') or default_data.get('__config_version', '2.0') data = {'__config_version': version, **{k: v for k, v in data.items() if k != '__config_version'}} with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) return True except Exception as e: self._log(f"保存文件失败 [{os.path.basename(file_path)}]: {str(e)}") return False def _ensure_simple_config(self, file_path: str, default_data: dict, file_desc: str = "config") -> bool: """ 简单确保配置文件存在（不进行版本管理和迁移） 逻辑: - 文件存在 → 跳过 - 文件不存在 → 创建默认配置（不含版本号） 适用于: active_prompts.json, tags_user.json 等简单配置文件 """ if os.path.exists(file_path): return False os.makedirs(os.path.dirname(file_path), exist_ok=True) try: # 移除版本号（这些文件不需要版本管理） data_to_save = {k: v for k, v in default_data.items() if not k.startswith('__')} with open(file_path, 'w', encoding='utf-8') as f: json.dump(data_to_save, f, ensure_ascii=False, indent=2) self._log(f"[{file_desc}] 文件不存在，创建默认配置...") return True except Exception as e: self._log(f"[{file_desc}] 创建默认配置失败: {str(e)}") return False # ---配置文件确保存在--- def ensure_config_exists(self, file_path: str, default_data: dict, legacy_path: str = None, file_desc: str = "config") -> bool: """ 确保单个配置文件存在 逻辑: 1. 文件存在 → 跳过（增量更新由 migrate_incremental_updates 处理） 2. 文件不存在 + 旧文件存在 → 迁移旧文件，与默认配置合并，添加版本号 3. 文件不存在 + 无旧文件 → 创建默认配置 参数: file_path: 目标文件路径 default_data: 默认配置数据 legacy_path: 旧版本文件路径（可选） file_desc: 文件描述（用于日志） 返回: bool: True 表示创建了新文件，False 表示文件已存在 """ if os.path.exists(file_path): return False # 确保目录存在 os.makedirs(os.path.dirname(file_path), exist_ok=True) # 检查是否有旧版本文件 if legacy_path and os.path.exists(legacy_path): try: with open(legacy_path, 'r', encoding='utf-8') as f: legacy_data = json.load(f) # 与默认配置合并，补全缺失的字段 merged_data = self._merge_with_defaults(legacy_data, default_data, file_desc) self._save_with_version(file_path, merged_data, default_data) self._log(f"[{file_desc}] 从插件目录迁移旧文件完成") return True except Exception as e: self._log(f"[{file_desc}] 迁移旧文件失败: {str(e)}，使用默认配置") # 创建默认配置 self._log(f"[{file_desc}] 文件不存在，创建默认配置...") self._save_with_version(file_path, default_data, default_data) return True def _merge_with_defaults(self, user_data: dict, default_data: dict, file_desc: str = "") -> dict: """ 将用户数据与默认配置合并，补全缺失的字段 参数: user_data: 用户数据（旧版本） default_data: 默认配置 file_desc: 文件描述（用于日志） 返回: 合并后的数据 """ import copy result = copy.deepcopy(default_data) # 递归合并用户数据到结果中（用户数据优先） self._recursive_merge(result, user_data, file_desc) return result def _recursive_merge(self, base: dict, overlay: dict, file_desc: str = "", path: str = ""): """ 递归合并：将 overlay 的值覆盖到 base 中（保留 base 的结构） 策略: - overlay 中存在的键覆盖 base 中的值 - 如果都是 dict，递归合并 - 跳过版本字段（由 _save_with_version 处理） """ for key, value in overlay.items(): # 跳过版本字段 if key.startswith("__"): continue current_path = f"{path}.{key}" if path else key if key in base: if isinstance(base[key], dict) and isinstance(value, dict): # 递归合并嵌套字典 self._recursive_merge(base[key], value, file_desc, current_path) else: # 直接覆盖（用户值优先） base[key] = value else: # overlay 中有但 base 中没有的键，直接添加（用户自定义内容） base[key] = value def ensure_all_configs_exist(self, default_configs: dict, legacy_dir: str): """ 确保所有配置文件存在 参数: default_configs: 默认配置字典 legacy_dir: 旧版本文件目录 """ # config.json（使用专用迁移方法） if 'config' in default_configs: self.ensure_config_json_exists( os.path.join(self.config_dir, "config.json"), default_configs['config'], os.path.join(legacy_dir, "config.json") ) # system_prompts.json if 'system_prompts' in default_configs: self.ensure_config_exists( os.path.join(self.rules_dir, "system_prompts.json"), default_configs['system_prompts'], os.path.join(legacy_dir, "system_prompts.json"), "system_prompts.json" ) # active_prompts.json 和 tags_user.json 不需要版本管理和迁移， # 直接在文件不存在时创建默认配置 self._ensure_simple_config( os.path.join(self.config_dir, "active_prompts.json"), default_configs.get('active_prompts', {}), "active_prompts.json" ) self._ensure_simple_config( os.path.join(self.config_dir, "tags_user.json"), default_configs.get('tags_user', {"favorites": []}), "tags_user.json" ) # kontext_presets.json if 'kontext_presets' in default_configs: self.ensure_config_exists( os.path.join(self.rules_dir, "kontext_presets.json"), default_configs['kontext_presets'], os.path.join(legacy_dir, "kontext_presets.json"), "kontext_presets.json" ) def migrate_incremental_updates(self, default_configs): """ 执行增量更新：将默认配置中的新字段添加到用户配置中 参数: default_configs: 包含各类默认配置的字典 { 'config': ..., 'system_prompts': ..., 'kontext_presets': ... } """ try: results = {} # 1. 更新 config.json if 'config' in default_configs: results['config_update'] = self._update_config_json(default_configs['config']) # 2. 更新 system_prompts.json if 'system_prompts' in default_configs: results['system_prompts_update'] = self._update_json_file( os.path.join(self.rules_dir, "system_prompts.json"), default_configs['system_prompts'], "system_prompts" ) # active_prompts.json 和 tags_user.json 不需要增量更新 # （结构简单，没有版本管理需求） # 5. 更新 kontext_presets.json if 'kontext_presets' in default_configs: results['kontext_presets_update'] = self._update_json_file( os.path.join(self.rules_dir, "kontext_presets.json"), default_configs['kontext_presets'], "kontext_presets" ) return results except Exception as e: self._log(f"增量更新失败: {str(e)}") return {} def _update_config_json(self, default_config): """ 处理 config.json 的增量更新 (带版本检查) 合并策略（与其他配置文件统一）: 1. 版本比对 2. 根级字段: 使用通用的 _deep_merge_defaults 补全 3. model_services: 按 id 匹配，只补全用户已有服务的缺失字段（不追加新服务） 4. 完成后同步版本号 """ user_config_path = os.path.join(self.config_dir, "config.json") if not os.path.exists(user_config_path): return False try: with open(user_config_path, 'r', encoding='utf-8') as f: user_config = json.load(f) # 获取版本号 template_version = default_config.get('__config_version', '2.0') user_version = user_config.get('__config_version') # 不设默认值 # 如果用户文件没有版本号，只补上版本号（静默跳过） if user_version is None: user_config = {'__config_version': template_version, **user_config} with open(user_config_path, 'w', encoding='utf-8') as f: json.dump(user_config, f, ensure_ascii=False, indent=2) return True # 版本比对：模板版本 <= 用户版本时跳过 cmp_result = self._compare_versions(template_version, user_version) if cmp_result <= 0: return False import copy # 1. 根级字段补全（排除 model_services，单独处理） for key, value in default_config.items(): if key == "model_services": continue # model_services 单独处理 if key not in user_config: user_config[key] = copy.deepcopy(value) self._log(f"[config.json] 补全根字段: {key}") elif isinstance(value, dict) and isinstance(user_config[key], dict): # 递归合并嵌套字典（如 baidu_translate、current_services） self._deep_merge_defaults(user_config[key], value) # 2. model_services 按 id 匹配合并 self._merge_model_services(user_config, default_config) # 更新版本号（重构字典确保版本号在开头） user_config = {'__config_version': template_version, **{k: v for k, v in user_config.items() if k != '__config_version'}} with open(user_config_path, 'w', encoding='utf-8') as f: json.dump(user_config, f, ensure_ascii=False, indent=2) self._log(f"[config.json] 增量更新已完成 (v{user_version} -> v{template_version})") return True except Exception as e: self._log(f"[config.json] 更新检查出错: {str(e)}") return False def _merge_model_services(self, user_config, default_config): """ 按 id 匹配合并 model_services（完整策略） 策略: - 补全用户已有服务的缺失字段 - 追加模板中用户不存在的服务商（版本更新时的新服务商） - 不覆盖 llm_models/vlm_models（用户自定义的模型列表） """ if 'model_services' not in default_config: return if 'model_services' not in user_config: user_config['model_services'] = [] # 构建用户服务的 id 集合 user_service_ids = {s.get('id') for s in user_config['model_services'] if s.get('id')} import copy # 1. 补全用户已有服务的缺失字段 template_services_map = { s.get('id'): s for s in default_config['model_services'] if s.get('id') } for user_service in user_config['model_services']: service_id = user_service.get('id') if not service_id or service_id not in template_services_map: continue template_service = template_services_map[service_id] service_name = user_service.get('name', service_id) # 补全服务级别的缺失字段 for key, value in template_service.items(): if key in ['llm_models', 'vlm_models']: # 模型列表不补全（用户自定义） continue if key not in user_service: user_service[key] = copy.deepcopy(value) self._log(f"[config.json] 补全服务 '{service_name}' 字段: {key}") # 2. 追加模板中用户不存在的服务商 for template_service in default_config['model_services']: service_id = template_service.get('id') if not service_id or service_id in user_service_ids: continue # 追加新服务商 new_service = copy.deepcopy(template_service) user_config['model_services'].append(new_service) self._log(f"[config.json] 追加新服务商: {new_service.get('name', service_id)}") def _update_json_file(self, file_path, default_data, file_desc): """ 通用的 JSON 文件增量更新 (带版本检查) 逻辑: 1. 检查用户文件是否有版本号 2. 如果没有版本号，只补上当前模板版本号（跳过增量更新） 3. 如果有版本号，比对模板版本和用户版本 4. 仅当模板版本 > 用户版本时执行增量 5. 增量完成后同步用户版本号 """ if not os.path.exists(file_path): return False try: with open(file_path, 'r', encoding='utf-8') as f: user_data = json.load(f) # 获取版本号 template_version = default_data.get('__config_version', '2.0') user_version = user_data.get('__config_version') # 不设默认值 # 如果用户文件没有版本号，只补上版本号（静默处理） if user_version is None: user_data = {'__config_version': template_version, **user_data} self._save_with_version(file_path, user_data, default_data) return True # 版本比对：模板版本 <= 用户版本时跳过 cmp_result = self._compare_versions(template_version, user_version) if cmp_result <= 0: return False # 执行深度合并 modified = self._deep_merge_defaults(user_data, default_data) # ---特殊处理：system_prompts 规则内容覆盖--- if file_desc == "system_prompts": modified = self._overwrite_prompts_from_template(user_data, default_data) or modified # ---特殊处理：为 system_prompts 中的所有规则补全 category 和 showIn 字段--- if file_desc == "system_prompts": modified = self._ensure_prompts_have_category(user_data) or modified modified = self._ensure_prompts_have_show_in(user_data) or modified # 无论是否有字段变更，都需要更新版本号（重构字典确保版本号在开头） user_data = {'__config_version': template_version, **{k: v for k, v in user_data.items() if k != '__config_version'}} with open(file_path, 'w', encoding='utf-8') as f: json.dump(user_data, f, ensure_ascii=False, indent=2) self._log(f"[{file_desc}.json] 增量更新已完成 (v{user_version} -> v{template_version})") return True except Exception as e: self._log(f"[{file_desc}.json] 更新检查出错: {str(e)}") return False def _ensure_prompts_have_category(self, system_prompts_data): """ 确保 system_prompts 中的所有规则都有 category 字段 为所有规则类型（expand_prompts、vision_prompts、video_prompts）中的每个规则 补全 category 字段（默认值为空字符串） 返回是否发生了修改 """ modified = False # 需要处理的规则类型 prompt_types = ['expand_prompts', 'vision_prompts', 'video_prompts'] for prompt_type in prompt_types: if prompt_type not in system_prompts_data: continue prompts = system_prompts_data[prompt_type] if not isinstance(prompts, dict): continue for prompt_id, prompt_data in prompts.items(): if not isinstance(prompt_data, dict): continue # 为规则补全 category 字段 if 'category' not in prompt_data: prompt_data['category'] = '' modified = True return modified def _ensure_prompts_have_show_in(self, system_prompts_data): """ 确保 system_prompts 中的所有规则都有 showIn 字段 为所有规则类型（expand_prompts、vision_prompts、video_prompts）中的每个规则 补全 showIn 字段（默认值为 ["frontend", "node"]） 返回是否发生了修改 """ modified = False # 需要处理的规则类型 prompt_types = ['expand_prompts', 'vision_prompts', 'video_prompts'] for prompt_type in prompt_types: if prompt_type not in system_prompts_data: continue prompts = system_prompts_data[prompt_type] if not isinstance(prompts, dict): continue for prompt_id, prompt_data in prompts.items(): if not isinstance(prompt_data, dict): continue # 为规则补全 showIn 字段 if 'showIn' not in prompt_data: prompt_data['showIn'] = ["frontend", "node"] modified = True return modified def _overwrite_prompts_from_template(self, user_data, template_data): """ 用模板中的规则内容覆盖用户配置中的同名规则 当版本更新时（模板版本 > 用户版本），将模板中的规则内容 完整覆盖到用户配置中的对应规则，确保内置规则保持最新。 覆盖策略: - 仅覆盖模板和用户配置中都存在的同名规则 - 用户自定义的规则（模板中不存在）保持不变 - 模板中新增的规则由 _deep_merge_defaults 处理 返回是否发生了修改 """ import copy modified = False # 需要处理的规则类型 prompt_types = ['expand_prompts', 'vision_prompts', 'video_prompts', 'translate_prompts'] for prompt_type in prompt_types: # 检查模板中是否有该类型 if prompt_type not in template_data: continue template_prompts = template_data[prompt_type] if not isinstance(template_prompts, dict): continue # 检查用户配置中是否有该类型 if prompt_type not in user_data: continue user_prompts = user_data[prompt_type] if not isinstance(user_prompts, dict): continue # 遍历模板中的每个规则 for prompt_id, template_prompt in template_prompts.items(): if not isinstance(template_prompt, dict): continue # 检查用户配置中是否存在同名规则 if prompt_id in user_prompts: # 用模板内容完整覆盖用户规则 user_prompts[prompt_id] = copy.deepcopy(template_prompt) modified = True self._log(f"[system_prompts] 覆盖规则: {template_prompt.get('name', prompt_id)}") return modified def _deep_merge_defaults(self, user_data, default_data): """ 递归将 default_data 中的缺失字段合并到 user_data 合并策略: - dict: 递归合并，补全缺失的键 - list: 将模板中不存在于用户列表的新元素追加到末尾 返回是否发生了修改 """ modified = False import copy # ---处理字典类型--- if isinstance(user_data, dict) and isinstance(default_data, dict): for key, value in default_data.items(): if key not in user_data: # 字段不存在，直接添加 user_data[key] = copy.deepcopy(value) modified = True else: # 字段存在，递归检查 if self._deep_merge_defaults(user_data[key], value): modified = True # ---处理数组类型--- elif isinstance(user_data, list) and isinstance(default_data, list): # 将模板数组中不存在于用户数组的元素追加到末尾 for item in default_data: if item not in user_data: user_data.append(copy.deepcopy(item)) modified = True return modified def migrate_tags_json_to_csv(self): """ 迁移旧版 JSON 标签到 CSV 格式 迁移逻辑: 1. 检查 tags 目录是否为空 2. 如果为空，读取插件 config 目录下的 tags.json 和 tags_user.json 3. tags.json → 转换为 "默认标签.csv" 4. tags_user.json → 转换为 "用户标签.csv" CSV 格式: 标签名\t标签值\t一级分类\t二级分类\t三级分类\t四级分类 """ try: # 1. 检查是否需要迁移 if not self._should_migrate_tags(): return False # 2. 读取 JSON 文件 tags_data, user_tags_data = self._load_legacy_tags_json() migrated_count = 0 # ---处理 tags.json → 默认标签.csv--- if tags_data: csv_rows = [] self._extract_tags_recursive(tags_data, [], csv_rows) if csv_rows: csv_filename = "默认标签.csv" self._write_tags_csv(csv_rows, csv_filename) self._log(f"[tags.json] ✅ 成功迁移 {len(csv_rows)} 个标签到 {csv_filename}") migrated_count += len(csv_rows) # ---处理 tags_user.json → 用户标签.csv--- if user_tags_data: csv_rows = [] # tags_user.json 是2层结构: {分类: {标签名: 标签值}} for category, tags in user_tags_data.items(): if not isinstance(tags, dict): continue for tag_name, tag_value in tags.items(): # CSV 行: [标签名, 标签值, 一级分类, 二级分类, 三级分类, 四级分类] row = [ tag_name, tag_value, category, "", # 二级分类（空） "", # 三级分类（空） "" # 四级分类（空） ] csv_rows.append(row) if csv_rows: csv_filename = "用户标签.csv" self._write_tags_csv(csv_rows, csv_filename) self._log(f"[tags_user.json] ✅ 成功迁移 {len(csv_rows)} 个标签到 {csv_filename}") migrated_count += len(csv_rows) # ---如果两个文件都不存在，尝试从模板创建默认标签--- if not tags_data and not user_tags_data: template_path = os.path.join(self.plugin_dir, "config", "tags_template.json") if os.path.exists(template_path): try: with open(template_path, 'r', encoding='utf-8') as f: template_data = json.load(f) csv_rows = [] self._extract_tags_recursive(template_data, [], csv_rows) if csv_rows: csv_filename = "默认标签.csv" self._write_tags_csv(csv_rows, csv_filename) self._log(f"✨ 检测到全新环境，已基于模板创建初始标签文件: {csv_filename}") return True except Exception as e: self._log(f"❗ 从模板创建初始标签失败: {str(e)}") # 模板读取失败或不存在时的最后保底 csv_rows = [["欢迎使用提示词小助手", "", "指南", "开始", "", ""]] csv_filename = "默认标签.csv" self._write_tags_csv(csv_rows, csv_filename) self._log(f"✨ 已创建简易初始标签文件: {csv_filename}") return True return migrated_count > 0 except Exception as e: self._log(f"[tags] ❗ 标签迁移失败: {str(e)}") return False def _should_migrate_tags(self): """检查是否需要迁移标签""" # 检查 tags 目录是否存在 if not os.path.exists(self.tags_dir): return True # 检查是否有 CSV 文件 try: csv_files = [f for f in os.listdir(self.tags_dir) if f.endswith('.csv')] return len(csv_files) == 0 except Exception: return True def _load_legacy_tags_json(self): """ 加载旧版 JSON 标签文件 返回: (tags_data, user_tags_data) 元组 """ tags_data = None user_tags_data = None # 读取 tags.json legacy_tags_path = os.path.join(self.legacy_config_dir, "tags.json") if os.path.exists(legacy_tags_path): try: with open(legacy_tags_path, 'r', encoding='utf-8') as f: tags_data = json.load(f) except Exception as e: self._log(f"❗ 读取 tags.json 失败: {str(e)}") # 读取 tags_user.json legacy_user_tags_path = os.path.join(self.legacy_config_dir, "tags_user.json") if os.path.exists(legacy_user_tags_path): try: with open(legacy_user_tags_path, 'r', encoding='utf-8') as f: user_tags_data = json.load(f) except Exception as e: self._log(f"❗ 读取 tags_user.json 失败: {str(e)}") return tags_data, user_tags_data def _extract_tags_recursive(self, data, categories, csv_rows): """ 递归提取标签数据 根据嵌套深度判断是分类还是标签： - 如果值是字符串，则为标签（键=标签名，值=标签值） - 如果值是字典，则为分类，继续递归 参数: data: 当前层级的数据字典 categories: 当前路径上的分类列表（最多4级） csv_rows: 结果列表，用于收集CSV行 """ for key, value in data.items(): if isinstance(value, str): # 值是字符串，说明当前键是标签名，值是标签值 # 构建CSV行：[标签名, 标签值, 一级分类, 二级分类, 三级分类, 四级分类] row = [key, value] # 填充分类（共4级，不足的补空字符串） for i in range(4): if i < len(categories): row.append(categories[i]) else: row.append("") csv_rows.append(row) elif isinstance(value, dict): # 值是字典，说明当前键是分类名，继续递归 # 限制最多4级分类，超过则忽略更深层级 if len(categories) < 4: new_categories = categories + [key] self._extract_tags_recursive(value, new_categories, csv_rows) else: # 超过4级分类，记录警告并跳过 self._log(f"⚠️ 分类层级超过4级，已忽略: {' → '.join(categories)} → {key}") def _write_tags_csv(self, csv_rows, filename): """ 写入 CSV 文件 参数: csv_rows: CSV 行数据 filename: 文件名 """ csv_path = os.path.join(self.tags_dir, filename) # 确保目录存在 os.makedirs(self.tags_dir, exist_ok=True) # 写入 CSV 文件（使用 utf-8-sig 编码，兼容 Excel） with open(csv_path, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f) # 写入表头 writer.writerow(['标签名', '标签值', '一级分类', '二级分类', '三级分类', '四级分类']) # 写入数据 writer.writerows(csv_rows) # --- Config.json 迁移 --- def migrate_config_api_keys(self): """ 迁移旧版 config.json 中的 API Key 到新版配置 迁移逻辑: 1. 检查用户配置目录的 config.json 是否存在 2. 如果不存在，读取插件 config 目录下的 config.json 3. 提取 API Key 并迁移到新版配置（v2.0 model_services 格式） 提取的 API Key: - baidu_translate: app_id, secret_key - llm.providers: zhipu, siliconflow, custom 的 api_key - vlm.providers: zhipu, siliconflow, custom 的 api_key """ try: # 1. 检查是否需要迁移 user_config_path = os.path.join(self.config_dir, "config.json") if os.path.exists(user_config_path): return False # 2. 读取旧版 config.json legacy_config_path = os.path.join(self.legacy_config_dir, "config.json") if not os.path.exists(legacy_config_path): return False # 3. 加载旧版配置 with open(legacy_config_path, 'r', encoding='utf-8') as f: legacy_config = json.load(f) self._log(f"[config.json] 找到旧版配置，准备迁移到 v2.0 格式") # 4. 将完整的旧版配置保存到临时文件，供 config_manager 转换 migration_data_path = os.path.join(self.config_dir, ".migration_legacy_config.json") os.makedirs(self.config_dir, exist_ok=True) with open(migration_data_path, 'w', encoding='utf-8') as f: json.dump(legacy_config, f, ensure_ascii=False, indent=2) return True except Exception as e: self._log(f"❗ config.json 迁移失败: {str(e)}") return False def _extract_api_keys_from_legacy_config(self, legacy_config): """ 从旧版 config.json 提取 API Key 参数: legacy_config: 旧版配置字典 返回: 提取的 API Key 字典 """ api_keys = {} # 提取百度翻译配置 if 'baidu_translate' in legacy_config: baidu = legacy_config['baidu_translate'] if baidu.get('app_id') or baidu.get('secret_key'): api_keys['baidu_translate'] = { 'app_id': baidu.get('app_id', ''), 'secret_key': baidu.get('secret_key', '') } self._log("提取百度翻译配置") # 提取 LLM API Key if 'llm' in legacy_config and 'providers' in legacy_config['llm']: llm_providers = legacy_config['llm']['providers'] api_keys['llm'] = {} for provider_name in ['zhipu', 'siliconflow', 'custom']: if provider_name in llm_providers: api_key = llm_providers[provider_name].get('api_key', '') if api_key: api_keys['llm'][provider_name] = api_key self._log(f"提取 LLM {provider_name} API Key") # 提取 VLM API Key if 'vlm' in legacy_config and 'providers' in legacy_config['vlm']: vlm_providers = legacy_config['vlm']['providers'] api_keys['vlm'] = {} for provider_name in ['zhipu', 'siliconflow', 'custom']: if provider_name in vlm_providers: api_key = vlm_providers[provider_name].get('api_key', '') if api_key: api_keys['vlm'][provider_name] = api_key self._log(f"提取 VLM {provider_name} API Key") return api_keys def run_migrations(plugin_dir, user_base_dir, logger=None, default_configs=None): """ 运行所有迁移任务 执行顺序: 1. 确保所有配置文件存在（缺则创建/迁移） 2. 执行旧版 API Key 迁移 3. 执行增量更新（版本比对后按需执行） 参数: plugin_dir: 插件目录路径 user_base_dir: 用户配置基础目录 logger: 日志函数（可选） default_configs: 默认配置字典（用于创建默认文件和增量更新） 返回: 迁移结果字典 """ tool = MigrationTool(plugin_dir, user_base_dir, logger) legacy_dir = os.path.join(plugin_dir, "config") results = { 'configs_created': False, 'tags_migration': False, 'config_migration': False, 'incremental_updates': {} } # 1. 确保所有配置文件存在 if default_configs: tool.ensure_all_configs_exist(default_configs, legacy_dir) results['configs_created'] = True # 2. 执行旧版 API Key 迁移 results['config_migration'] = tool.migrate_config_api_keys() results['tags_migration'] = tool.migrate_tags_json_to_csv() # 3. 执行增量更新（版本比对后按需执行） if default_configs: results['incremental_updates'] = tool.migrate_incremental_updates(default_configs) return results