feat:优化职业技能关联实现

scripts/migrate_generated_characters.py

@@ -1,9 +1,11 @@
-"""Merge OpenAPI-generated layouts into runtime character modules.
+"""将生成数据合并到已有职业实现。
 
-Skills are matched strictly by their constant ``name``. Runtime Skill class
-identifiers and implementations are preserved; ``position`` and ``cellPos``
-come from the generated class. Generated-only skills are retained as commented blocks.
-The runtime header and profession-specific footer are preserved unchanged.
+优先按职业 ref、其次 UUID 匹配；缺少稳定身份时才兼容旧中文名称。
+保留运行时类编号、名称、关联规则和手写方法；旧技能缺少 ref 时补齐绑定及导入。
+生成器独有技能保持注释，运行时独有技能原样保留；职业变体按 prefix 查找来源。
+
+先执行 backend/openapi/scripts/generate.py 同步 refs 和生成数据，再运行本脚本。
+--check 只检查合并及语法，不写职业文件；--generated-root/--character-root 可指定验证目录。
 """
 
 from __future__ import annotations
@@ -30,6 +32,9 @@ class SkillBlock:
     end: int
     position: str | None
     cell_pos: str | None
+    ref: str | None
+    ref_key: tuple[str, str, str] | None
+    uuid: str | None
 
 
 def _offsets(source: str) -> list[int]:
@@ -42,6 +47,12 @@ def _offsets(source: str) -> list[int]:
 def _skills(source: str, path: Path) -> list[SkillBlock]:
     tree = ast.parse(source, filename=str(path))
     offsets = _offsets(source)
+    imports = {
+        item.asname or item.name: (node.module, item.name)
+        for node in tree.body if isinstance(node, ast.ImportFrom)
+        and (node.module or '').startswith('core.character.refs.')
+        for item in node.names
+    }
     result: list[SkillBlock] = []
     for node in tree.body:
         if not isinstance(node, ast.ClassDef) or not re.fullmatch(r"Skill\d+", node.name):
@@ -49,12 +60,21 @@ def _skills(source: str, path: Path) -> list[SkillBlock]:
         name = None
         position = None
         cell_pos = None
+        ref = ref_key = uuid = None
         for statement in node.body:
             if not isinstance(statement, (ast.Assign, ast.AnnAssign)):
                 continue
             targets = statement.targets if isinstance(statement, ast.Assign) else [statement.target]
             if len(targets) != 1 or not isinstance(targets[0], ast.Name):
                 continue
+            if targets[0].id == 'uuid' and isinstance(statement.value, ast.Constant):
+                uuid = statement.value.value
+            if targets[0].id == 'ref':
+                value = statement.value
+                if not isinstance(value, ast.Attribute) or not isinstance(value.value, ast.Name) or value.value.id not in imports:
+                    raise ValueError(f'{path}: 技能 ref 必须引用已导入的职业 Ref 成员')
+                ref = ast.unparse(value)
+                ref_key = (*imports[value.value.id], value.attr)
             if targets[0].id == "name" and isinstance(statement.value, ast.Constant):
                 name = statement.value.value
             elif targets[0].id == "position":
@@ -71,6 +91,7 @@ def _skills(source: str, path: Path) -> list[SkillBlock]:
                 offsets[node.end_lineno],
                 position,
                 cell_pos,
+                ref, ref_key, uuid,
             )
         )
     return result
@@ -458,11 +479,72 @@ def _sync_module_assignment(
 
 
 def _comment(block: str, name: str) -> str:
-    lines = [f"# MIGRATION: character 中没有同名技能“{name}”，暂不启用\n"]
+    lines = [f"# MIGRATION: character 中没有匹配身份的技能“{name}”，暂不启用\n"]
     lines.extend("# " + line.rstrip() + "\n" if line.strip() else "#\n" for line in block.splitlines())
     return "".join(lines)
 
 
+def _identity_indexes(skills, path):
+    """校验稳定身份唯一性，禁止重复引用或 UUID 被字典静默覆盖。"""
+    indexes = {}
+    for field in ('ref_key', 'uuid', 'skill_name'):
+        index = {}
+        for skill in skills:
+            value = getattr(skill, field)
+            if value is None:
+                continue
+            if value in index:
+                raise ValueError(f'{path}: duplicate {field}: {value}')
+            index[value] = skill
+        indexes[field] = index
+    return indexes
+
+
+def _match_skill(skill, indexes):
+    """优先 ref、其次 UUID；仅在缺少可比较身份时兼容旧中文名称。"""
+    by_ref = indexes['ref_key'].get(skill.ref_key)
+    by_uuid = indexes['uuid'].get(skill.uuid)
+    if by_ref is not None and by_uuid is not None and by_ref is not by_uuid:
+        raise ValueError(f'技能 ref 与 UUID 指向不同目标: {skill.skill_name}')
+    candidate = by_ref or by_uuid or indexes['skill_name'].get(skill.skill_name)
+    if candidate is None:
+        return None
+    if skill.ref_key and candidate.ref_key and skill.ref_key != candidate.ref_key:
+        raise ValueError(f'技能引用冲突: {skill.skill_name}: {candidate.ref} -> {skill.ref}')
+    if by_ref is None and skill.uuid and candidate.uuid and skill.uuid != candidate.uuid:
+        raise ValueError(f'同名技能 UUID 冲突: {skill.skill_name}')
+    return candidate
+
+
+def _ensure_ref_imports(source, generated):
+    """补齐生成代码的职业 Ref 导入，保留原有别名，拒绝别名指向不同对象。"""
+    tree = ast.parse(source)
+    imports = {
+        item.asname or item.name: (node.module, item.name)
+        for node in tree.body if isinstance(node, ast.ImportFrom) for item in node.names
+    }
+    additions = []
+    for node in ast.parse(generated).body:
+        if not isinstance(node, ast.ImportFrom) or not (node.module or '').startswith('core.character.refs.'):
+            continue
+        for item in node.names:
+            alias = item.asname or item.name
+            if alias in imports:
+                if imports[alias] != (node.module, item.name):
+                    raise ValueError(f'职业引用导入别名冲突: {alias}')
+            else:
+                additions.append(f'from {node.module} import {item.name}' + (f' as {item.asname}' if item.asname else '') + '\n')
+                imports[alias] = (node.module, item.name)
+    if not additions:
+        return source
+    position = max((node.end_lineno for node in tree.body if isinstance(node, (ast.Import, ast.ImportFrom))), default=0)
+    if position == 0:
+        return ''.join(additions) + '\n' + source.lstrip('\r\n')
+    lines = source.splitlines(keepends=True)
+    lines[position:position] = additions
+    return ''.join(lines)
+
+
 def merge(generated_path: Path, character_path: Path) -> tuple[str, dict[str, int]]:
     generated = generated_path.read_text(encoding="utf-8")
     character = character_path.read_text(encoding="utf-8")
@@ -470,9 +552,9 @@ def merge(generated_path: Path, character_path: Path) -> tuple[str, dict[str, in
     character_skills = _skills(character, character_path)
     if not generated_skills or not character_skills:
         raise ValueError(f"missing skills: {generated_path} or {character_path}")
-    runtime = {skill.skill_name: skill for skill in character_skills}
-    if len(runtime) != len(character_skills):
-        raise ValueError(f"duplicate skill name: {character_path}")
+    runtime = _identity_indexes(character_skills, character_path)
+    _identity_indexes(generated_skills, generated_path)
+    used = set()
 
     character_header_end, character_footer_start = _module_skill_bounds(character, character_path)
     generated_header_end, _ = _module_skill_bounds(generated, generated_path)
@@ -483,13 +565,21 @@ def merge(generated_path: Path, character_path: Path) -> tuple[str, dict[str, in
     for index, generated_skill in enumerate(generated_skills):
         if index:
             chunks.append(generated[cursor:generated_skill.start])
-        runtime_skill = runtime.get(generated_skill.skill_name)
+        runtime_skill = _match_skill(generated_skill, runtime)
         if runtime_skill is None:
             chunks.append(_comment(generated[generated_skill.start:generated_skill.end], generated_skill.skill_name))
             commented += 1
         else:
+            if runtime_skill.class_name in used:
+                raise ValueError(f'多个生成技能匹配同一运行时技能: {runtime_skill.class_name}')
+            used.add(runtime_skill.class_name)
             block = character[runtime_skill.start:runtime_skill.end]
             generated_block = generated[generated_skill.start:generated_skill.end]
+            if generated_skill.ref is not None and runtime_skill.ref is None:
+                block = _set_generated_field(block, 'ref', generated_skill.ref)
+            # 名称保留运行时值，防止破坏未迁移的名称关联；稳定 ref 匹配不要求名称相同。
+            if generated_skill.uuid is not None and generated_skill.ref_key == runtime_skill.ref_key and generated_skill.ref_key is not None:
+                block = _set_generated_field(block, 'uuid', repr(generated_skill.uuid))
             block = _sync_generated_assignment(block, generated_block, "vps")
             block = _sync_generated_data(block, generated_block)
             block = _sync_generated_comments(block, generated_block)
@@ -502,11 +592,16 @@ def merge(generated_path: Path, character_path: Path) -> tuple[str, dict[str, in
             chunks.append(block.rstrip() + "\n")
             matched += 1
         cursor = generated_skill.end
+    # 未出现在新生成数据中的手写技能必须保留，不能随布局重建被删除。
+    for skill in character_skills:
+        if skill.class_name not in used:
+            chunks.append('\n' + character[skill.start:skill.end].rstrip() + '\n')
     chunks.append("\n" + character[character_footer_start:].lstrip("\r\n"))
     merged = "".join(chunks)
     merged = "\n".join(line.rstrip() for line in merged.splitlines()) + "\n"
     merged = _sync_module_assignment(merged, generated, "prefix")
     merged = _ensure_get_data_import(merged)
+    merged = _ensure_ref_imports(merged, generated)
     return merged, {
         "generated": len(generated_skills), "runtime": len(character_skills),
         "matched": matched, "commented": commented,
@@ -545,12 +640,22 @@ def _atomic_write(path: Path, content: str) -> None:
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--check", action="store_true", help="validate without writing")
+    parser.add_argument("--generated-root", type=Path, default=GENERATED_ROOT)
+    parser.add_argument("--character-root", type=Path, default=CHARACTER_ROOT)
     args = parser.parse_args()
     totals = {key: 0 for key in ("files", "generated", "runtime", "matched", "commented", "runtime_only")}
-    for generated_path in sorted(GENERATED_ROOT.rglob("*.py")):
-        character_path = CHARACTER_ROOT / generated_path.relative_to(GENERATED_ROOT)
-        if not character_path.exists():
-            print(f"SKIP {generated_path.relative_to(GENERATED_ROOT)}: no runtime profession")
+    for character_path in sorted(args.character_root.rglob("*.py")):
+        if character_path.name == '__init__.py':
+            continue
+        # 使用数据 prefix 找生成来源，输出圣骑士等变体也能复用来源职业。
+        module = ast.parse(character_path.read_text(encoding='utf-8'))
+        prefix = next((node.value.value for node in module.body
+                       if _statement_key(node) == 'prefix' and isinstance(node.value, ast.Constant)), None)
+        relative = character_path.relative_to(args.character_root)
+        generated_relative = Path(*prefix.split('/')[:2]).with_suffix('.py') if isinstance(prefix, str) else relative
+        generated_path = args.generated_root / generated_relative
+        if not generated_path.exists():
+            print(f"SKIP {relative}: no generated profession")
             continue
         output, stats = merge(generated_path, character_path)
         ast.parse(output, filename=str(character_path))
@@ -559,7 +664,7 @@ def main() -> int:
         totals["files"] += 1
         for key, value in stats.items():
             totals[key] += value
-        print(f"{'CHECK' if args.check else 'WRITE'} {generated_path.relative_to(GENERATED_ROOT)}: "
+        print(f"{'CHECK' if args.check else 'WRITE'} {relative}: "
               f"matched={stats['matched']} commented={stats['commented']} runtime_only={stats['runtime_only']}")
     print("TOTAL " + " ".join(f"{key}={value}" for key, value in totals.items()))
     return 0