chore: v5 到 v6 配置迁移脚本
This commit is contained in:
parent
68b0cbda73
commit
ef725b4e6f
|
@ -8,6 +8,7 @@ from . import _v1_to_v2
|
|||
from . import _v2_to_v3
|
||||
from . import _v3_to_v4
|
||||
from . import _v4_to_v5
|
||||
from . import _v5_to_v6
|
||||
|
||||
# 注册表:键为旧版本号,值为迁移函数
|
||||
MIGRATION_REGISTRY: Dict[int, Migration] = {
|
||||
|
@ -15,10 +16,11 @@ MIGRATION_REGISTRY: Dict[int, Migration] = {
|
|||
2: _v2_to_v3.migrate,
|
||||
3: _v3_to_v4.migrate,
|
||||
4: _v4_to_v5.migrate,
|
||||
5: _v5_to_v6.migrate,
|
||||
}
|
||||
|
||||
# 当前最新配置版本
|
||||
LATEST_VERSION: int = 5
|
||||
LATEST_VERSION: int = 6
|
||||
|
||||
__all__ = [
|
||||
"MIGRATION_REGISTRY",
|
||||
|
|
|
@ -0,0 +1,134 @@
|
|||
"""v5 -> v6 迁移脚本
|
||||
|
||||
重构培育配置:将原有的 ProduceConfig 中的培育参数迁移到新的 ProduceSolution 结构中。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import json
|
||||
import uuid
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _sanitize_filename(name: str) -> str:
|
||||
"""
|
||||
清理文件名中的非法字符
|
||||
|
||||
:param name: 原始名称
|
||||
:return: 清理后的文件名
|
||||
"""
|
||||
# 替换 \/:*?"<>| 为下划线
|
||||
return re.sub(r'[\\/:*?"<>|]', '_', name)
|
||||
|
||||
|
||||
def _create_default_solution(old_produce_config: dict[str, Any]) -> tuple[dict[str, Any], str]:
|
||||
"""
|
||||
根据旧的培育配置创建默认的培育方案
|
||||
|
||||
:param old_produce_config: 旧的培育配置
|
||||
:return: (新的培育方案数据, 方案ID)
|
||||
"""
|
||||
# 生成唯一ID
|
||||
solution_id = uuid.uuid4().hex
|
||||
|
||||
# 构建培育数据
|
||||
produce_data = {
|
||||
"mode": old_produce_config.get("mode", "regular"),
|
||||
"idol": old_produce_config.get("idols", [None])[0] if old_produce_config.get("idols") else None,
|
||||
"memory_set": old_produce_config.get("memory_sets", [None])[0] if old_produce_config.get("memory_sets") else None,
|
||||
"support_card_set": old_produce_config.get("support_card_sets", [None])[0] if old_produce_config.get("support_card_sets") else None,
|
||||
"auto_set_memory": old_produce_config.get("auto_set_memory", False),
|
||||
"auto_set_support_card": old_produce_config.get("auto_set_support_card", False),
|
||||
"use_pt_boost": old_produce_config.get("use_pt_boost", False),
|
||||
"use_note_boost": old_produce_config.get("use_note_boost", False),
|
||||
"follow_producer": old_produce_config.get("follow_producer", False),
|
||||
"self_study_lesson": old_produce_config.get("self_study_lesson", "dance"),
|
||||
"prefer_lesson_ap": old_produce_config.get("prefer_lesson_ap", False),
|
||||
"actions_order": old_produce_config.get("actions_order", [
|
||||
"recommended", "visual", "vocal", "dance",
|
||||
"allowance", "outing", "study", "consult", "rest"
|
||||
]),
|
||||
"recommend_card_detection_mode": old_produce_config.get("recommend_card_detection_mode", "normal"),
|
||||
"use_ap_drink": old_produce_config.get("use_ap_drink", False),
|
||||
"skip_commu": old_produce_config.get("skip_commu", True)
|
||||
}
|
||||
|
||||
# 构建方案对象
|
||||
solution = {
|
||||
"type": "produce_solution",
|
||||
"id": solution_id,
|
||||
"name": "默认方案",
|
||||
"description": "从旧配置迁移的默认培育方案",
|
||||
"data": produce_data
|
||||
}
|
||||
|
||||
return solution, solution_id
|
||||
|
||||
|
||||
def _save_solution_to_file(solution: dict[str, Any]) -> None:
|
||||
"""
|
||||
将培育方案保存到文件
|
||||
|
||||
:param solution: 培育方案数据
|
||||
"""
|
||||
solutions_dir = "conf/produce"
|
||||
os.makedirs(solutions_dir, exist_ok=True)
|
||||
|
||||
safe_name = _sanitize_filename(solution["name"])
|
||||
file_path = os.path.join(solutions_dir, f"{safe_name}.json")
|
||||
|
||||
with open(file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(solution, f, ensure_ascii=False, indent=4)
|
||||
|
||||
|
||||
def migrate(user_config: dict[str, Any]) -> str | None: # noqa: D401
|
||||
"""执行 v5→v6 迁移:重构培育配置结构。
|
||||
|
||||
将原有的 ProduceConfig 中的培育参数迁移到新的 ProduceSolution 结构中。
|
||||
"""
|
||||
options = user_config.get("options")
|
||||
if options is None:
|
||||
logger.debug("No 'options' in user_config, skip v5→v6 migration.")
|
||||
return None
|
||||
|
||||
produce_conf = options.get("produce", {})
|
||||
if not produce_conf:
|
||||
logger.debug("No 'produce' config found, skip v5→v6 migration.")
|
||||
return None
|
||||
|
||||
# 检查是否已经是新格式(有 selected_solution_id 字段)
|
||||
if "selected_solution_id" in produce_conf:
|
||||
logger.debug("Produce config already in v6 format, skip migration.")
|
||||
return None
|
||||
|
||||
msg = ""
|
||||
|
||||
try:
|
||||
# 创建默认培育方案
|
||||
solution, solution_id = _create_default_solution(produce_conf)
|
||||
|
||||
# 保存方案到文件
|
||||
_save_solution_to_file(solution)
|
||||
|
||||
# 更新配置为新格式
|
||||
new_produce_conf = {
|
||||
"enabled": produce_conf.get("enabled", False),
|
||||
"selected_solution_id": solution_id,
|
||||
"produce_count": produce_conf.get("produce_count", 1)
|
||||
}
|
||||
|
||||
options["produce"] = new_produce_conf
|
||||
user_config["options"] = options
|
||||
|
||||
msg = f"已将培育配置迁移到新的方案系统。默认方案已创建并保存为 '{solution['name']}'。"
|
||||
logger.info("Successfully migrated produce config to v6 format with solution ID: %s", solution_id)
|
||||
|
||||
except Exception as e:
|
||||
logger.error("Failed to migrate produce config: %s", e)
|
||||
msg = f"培育配置迁移失败:{e}"
|
||||
|
||||
return msg or None
|
|
@ -0,0 +1,200 @@
|
|||
"""测试 v5 到 v6 的配置迁移脚本"""
|
||||
import unittest
|
||||
import tempfile
|
||||
import os
|
||||
import json
|
||||
import shutil
|
||||
from typing import Any
|
||||
|
||||
from kotonebot.kaa.config.migrations._v5_to_v6 import migrate
|
||||
|
||||
|
||||
class TestMigrationV5ToV6(unittest.TestCase):
|
||||
"""测试 v5 到 v6 的配置迁移"""
|
||||
|
||||
def setUp(self):
|
||||
"""设置测试环境"""
|
||||
# 创建临时目录
|
||||
self.temp_dir = tempfile.mkdtemp()
|
||||
self.original_cwd = os.getcwd()
|
||||
os.chdir(self.temp_dir)
|
||||
|
||||
def tearDown(self):
|
||||
"""清理测试环境"""
|
||||
os.chdir(self.original_cwd)
|
||||
shutil.rmtree(self.temp_dir)
|
||||
|
||||
def test_migrate_empty_config(self):
|
||||
"""测试空配置的迁移"""
|
||||
user_config = {}
|
||||
result = migrate(user_config)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_migrate_no_options(self):
|
||||
"""测试没有 options 的配置"""
|
||||
user_config = {"backend": {"type": "mumu12"}}
|
||||
result = migrate(user_config)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_migrate_no_produce_config(self):
|
||||
"""测试没有 produce 配置的情况"""
|
||||
user_config = {"options": {"purchase": {"enabled": False}}}
|
||||
result = migrate(user_config)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_migrate_already_v6_format(self):
|
||||
"""测试已经是 v6 格式的配置"""
|
||||
user_config = {
|
||||
"options": {
|
||||
"produce": {
|
||||
"enabled": True,
|
||||
"selected_solution_id": "test-id",
|
||||
"produce_count": 1
|
||||
}
|
||||
}
|
||||
}
|
||||
result = migrate(user_config)
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_migrate_v5_to_v6_basic(self):
|
||||
"""测试基本的 v5 到 v6 迁移"""
|
||||
# 创建 v5 格式的配置
|
||||
old_produce_config = {
|
||||
"enabled": True,
|
||||
"mode": "pro",
|
||||
"produce_count": 3,
|
||||
"idols": ["i_card-skin-fktn-3-000"],
|
||||
"memory_sets": [1],
|
||||
"support_card_sets": [2],
|
||||
"auto_set_memory": False,
|
||||
"auto_set_support_card": True,
|
||||
"use_pt_boost": True,
|
||||
"use_note_boost": False,
|
||||
"follow_producer": True,
|
||||
"self_study_lesson": "vocal",
|
||||
"prefer_lesson_ap": True,
|
||||
"actions_order": ["recommended", "visual", "vocal"],
|
||||
"recommend_card_detection_mode": "strict",
|
||||
"use_ap_drink": True,
|
||||
"skip_commu": False
|
||||
}
|
||||
|
||||
user_config = {"options": {"produce": old_produce_config}}
|
||||
|
||||
# 执行迁移
|
||||
result = migrate(user_config)
|
||||
|
||||
# 验证结果
|
||||
self.assertIsNotNone(result)
|
||||
assert result is not None # make pylance happy
|
||||
self.assertIn("已将培育配置迁移到新的方案系统", result)
|
||||
|
||||
# 验证新配置格式
|
||||
new_produce_config = user_config["options"]["produce"]
|
||||
self.assertEqual(new_produce_config["enabled"], True)
|
||||
self.assertEqual(new_produce_config["produce_count"], 3)
|
||||
self.assertIsNotNone(new_produce_config["selected_solution_id"])
|
||||
|
||||
# 验证方案文件是否创建
|
||||
solutions_dir = "conf/produce"
|
||||
self.assertTrue(os.path.exists(solutions_dir))
|
||||
|
||||
# 查找创建的方案文件
|
||||
solution_files = [f for f in os.listdir(solutions_dir) if f.endswith('.json')]
|
||||
self.assertEqual(len(solution_files), 1)
|
||||
|
||||
# 验证方案文件内容
|
||||
solution_file = os.path.join(solutions_dir, solution_files[0])
|
||||
with open(solution_file, 'r', encoding='utf-8') as f:
|
||||
solution_data = json.load(f)
|
||||
|
||||
self.assertEqual(solution_data["type"], "produce_solution")
|
||||
self.assertEqual(solution_data["name"], "默认方案")
|
||||
self.assertEqual(solution_data["description"], "从旧配置迁移的默认培育方案")
|
||||
|
||||
# 验证培育数据
|
||||
produce_data = solution_data["data"]
|
||||
self.assertEqual(produce_data["mode"], "pro")
|
||||
self.assertEqual(produce_data["idol"], "i_card-skin-fktn-3-000")
|
||||
self.assertEqual(produce_data["memory_set"], 1)
|
||||
self.assertEqual(produce_data["support_card_set"], 2)
|
||||
self.assertEqual(produce_data["auto_set_memory"], False)
|
||||
self.assertEqual(produce_data["auto_set_support_card"], True)
|
||||
self.assertEqual(produce_data["use_pt_boost"], True)
|
||||
self.assertEqual(produce_data["use_note_boost"], False)
|
||||
self.assertEqual(produce_data["follow_producer"], True)
|
||||
self.assertEqual(produce_data["self_study_lesson"], "vocal")
|
||||
self.assertEqual(produce_data["prefer_lesson_ap"], True)
|
||||
self.assertEqual(produce_data["actions_order"], ["recommended", "visual", "vocal"])
|
||||
self.assertEqual(produce_data["recommend_card_detection_mode"], "strict")
|
||||
self.assertEqual(produce_data["use_ap_drink"], True)
|
||||
self.assertEqual(produce_data["skip_commu"], False)
|
||||
|
||||
def test_migrate_v5_to_v6_with_defaults(self):
|
||||
"""测试使用默认值的 v5 到 v6 迁移"""
|
||||
# 创建最小的 v5 格式配置
|
||||
old_produce_config = {"enabled": False}
|
||||
user_config = {"options": {"produce": old_produce_config}}
|
||||
|
||||
# 执行迁移
|
||||
result = migrate(user_config)
|
||||
|
||||
# 验证结果
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
# 验证新配置格式
|
||||
new_produce_config = user_config["options"]["produce"]
|
||||
self.assertEqual(new_produce_config["enabled"], False)
|
||||
self.assertEqual(new_produce_config["produce_count"], 1)
|
||||
self.assertIsNotNone(new_produce_config["selected_solution_id"])
|
||||
|
||||
# 验证方案文件内容使用了默认值
|
||||
solutions_dir = "conf/produce"
|
||||
solution_files = [f for f in os.listdir(solutions_dir) if f.endswith('.json')]
|
||||
solution_file = os.path.join(solutions_dir, solution_files[0])
|
||||
|
||||
with open(solution_file, 'r', encoding='utf-8') as f:
|
||||
solution_data = json.load(f)
|
||||
|
||||
produce_data = solution_data["data"]
|
||||
self.assertEqual(produce_data["mode"], "regular")
|
||||
self.assertIsNone(produce_data["idol"])
|
||||
self.assertIsNone(produce_data["memory_set"])
|
||||
self.assertIsNone(produce_data["support_card_set"])
|
||||
self.assertEqual(produce_data["auto_set_memory"], False)
|
||||
self.assertEqual(produce_data["auto_set_support_card"], False)
|
||||
self.assertEqual(produce_data["self_study_lesson"], "dance")
|
||||
self.assertEqual(produce_data["skip_commu"], True)
|
||||
|
||||
def test_migrate_v5_to_v6_multiple_idols_memory_support(self):
|
||||
"""测试多个偶像、回忆、支援卡的迁移(只取第一个)"""
|
||||
old_produce_config = {
|
||||
"enabled": True,
|
||||
"idols": ["idol1", "idol2", "idol3"],
|
||||
"memory_sets": [1, 2, 3],
|
||||
"support_card_sets": [4, 5, 6]
|
||||
}
|
||||
user_config = {"options": {"produce": old_produce_config}}
|
||||
|
||||
# 执行迁移
|
||||
result = migrate(user_config)
|
||||
|
||||
# 验证结果
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
# 验证方案文件内容只使用了第一个值
|
||||
solutions_dir = "conf/produce"
|
||||
solution_files = [f for f in os.listdir(solutions_dir) if f.endswith('.json')]
|
||||
solution_file = os.path.join(solutions_dir, solution_files[0])
|
||||
|
||||
with open(solution_file, 'r', encoding='utf-8') as f:
|
||||
solution_data = json.load(f)
|
||||
|
||||
produce_data = solution_data["data"]
|
||||
self.assertEqual(produce_data["idol"], "idol1")
|
||||
self.assertEqual(produce_data["memory_set"], 1)
|
||||
self.assertEqual(produce_data["support_card_set"], 4)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
Loading…
Reference in New Issue