675 lines
30 KiB
Python
675 lines
30 KiB
Python
"""将生成数据合并到已有职业实现。
|
|
|
|
优先按职业 ref、其次 UUID 匹配；缺少稳定身份时才兼容旧中文名称。
|
|
保留运行时类编号、名称、关联规则和手写方法；旧技能缺少 ref 时补齐绑定及导入。
|
|
生成器独有技能保持注释，运行时独有技能原样保留；职业变体按 prefix 查找来源。
|
|
|
|
先执行 backend/openapi/scripts/generate.py 同步 refs 和生成数据，再运行本脚本。
|
|
--check 只检查合并及语法，不写职业文件；--generated-root/--character-root 可指定验证目录。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
from dataclasses import dataclass
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
GENERATED_ROOT = ROOT / "backend" / "openapi" / "generate"
|
|
CHARACTER_ROOT = ROOT / "backend" / "core" / "character" / "GF"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SkillBlock:
|
|
class_name: str
|
|
skill_name: str
|
|
start: int
|
|
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]:
|
|
result = [0]
|
|
for line in source.splitlines(keepends=True):
|
|
result.append(result[-1] + len(line))
|
|
return result
|
|
|
|
|
|
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):
|
|
continue
|
|
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":
|
|
position = ast.get_source_segment(source, statement.value)
|
|
elif targets[0].id == "cellPos":
|
|
cell_pos = ast.get_source_segment(source, statement.value)
|
|
if not isinstance(name, str) or node.end_lineno is None:
|
|
raise ValueError(f"{path}:{node.lineno}: invalid {node.name}")
|
|
result.append(
|
|
SkillBlock(
|
|
node.name,
|
|
name,
|
|
offsets[node.lineno - 1],
|
|
offsets[node.end_lineno],
|
|
position,
|
|
cell_pos,
|
|
ref, ref_key, uuid,
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def _module_skill_bounds(source: str, path: Path) -> tuple[int, int]:
|
|
"""Return the code-only header end and profession-footer start offsets."""
|
|
tree = ast.parse(source, filename=str(path))
|
|
offsets = _offsets(source)
|
|
indexes = [
|
|
index
|
|
for index, node in enumerate(tree.body)
|
|
if isinstance(node, ast.ClassDef) and re.fullmatch(r"Skill\d+", node.name)
|
|
]
|
|
if not indexes:
|
|
raise ValueError(f"missing skills: {path}")
|
|
first_index, last_index = indexes[0], indexes[-1]
|
|
header_end = offsets[tree.body[first_index - 1].end_lineno] if first_index else 0
|
|
footer_start = (
|
|
offsets[tree.body[last_index + 1].lineno - 1]
|
|
if last_index + 1 < len(tree.body)
|
|
else len(source)
|
|
)
|
|
return header_end, footer_start
|
|
|
|
|
|
def _set_generated_field(block: str, field: str, value: str | None) -> str:
|
|
if value is None:
|
|
return block
|
|
assignment = re.compile(rf"(?m)^(\s*{re.escape(field)}\s*(?::[^=\n]+)?=\s*).*$")
|
|
if assignment.search(block):
|
|
return assignment.sub(rf"\g<1>{value}", block, count=1)
|
|
return re.sub(r"(?m)^(\s*name\s*=.*)$", rf"\g<1>\n {field} = {value}", block, count=1)
|
|
|
|
|
|
def _statement_key(statement: ast.stmt) -> str | None:
|
|
if isinstance(statement, (ast.Assign, ast.AnnAssign)):
|
|
targets = statement.targets if isinstance(statement, ast.Assign) else [statement.target]
|
|
if len(targets) == 1 and isinstance(targets[0], ast.Name):
|
|
return targets[0].id
|
|
return None
|
|
|
|
|
|
def _is_data_key(key: str | None) -> bool:
|
|
return bool(key and re.fullmatch(r"data\d+", key))
|
|
|
|
|
|
def _sync_generated_assignment(
|
|
runtime_block: str, generated_block: str, field: str
|
|
) -> str:
|
|
"""Replace all runtime definitions of one field with the generated one."""
|
|
runtime_class = ast.parse(runtime_block).body[0]
|
|
generated_class = ast.parse(generated_block).body[0]
|
|
if not isinstance(runtime_class, ast.ClassDef) or not isinstance(generated_class, ast.ClassDef):
|
|
return runtime_block
|
|
generated_statement = next(
|
|
(statement for statement in generated_class.body if _statement_key(statement) == field),
|
|
None,
|
|
)
|
|
if generated_statement is None:
|
|
return runtime_block
|
|
|
|
runtime_statements = [
|
|
statement for statement in runtime_class.body if _statement_key(statement) == field
|
|
]
|
|
runtime_lines = runtime_block.splitlines(keepends=True)
|
|
generated_lines = generated_block.splitlines(keepends=True)
|
|
generated_source = generated_lines[
|
|
generated_statement.lineno - 1:generated_statement.end_lineno
|
|
]
|
|
if runtime_statements:
|
|
insertion_line = runtime_statements[0].lineno - 1
|
|
for statement in sorted(runtime_statements, key=lambda item: item.lineno, reverse=True):
|
|
runtime_lines[statement.lineno - 1:statement.end_lineno] = []
|
|
else:
|
|
anchor = next(
|
|
(
|
|
statement
|
|
for key in ("custom", "hasUP", "hasVP", "uuid")
|
|
for statement in reversed(runtime_class.body)
|
|
if _statement_key(statement) == key
|
|
),
|
|
None,
|
|
)
|
|
insertion_line = anchor.end_lineno if anchor is not None else 1
|
|
generated_source = ["\n", *generated_source]
|
|
runtime_lines[insertion_line:insertion_line] = generated_source
|
|
return "".join(runtime_lines)
|
|
|
|
|
|
def _sync_generated_data(runtime_block: str, generated_block: str) -> str:
|
|
"""Rebuild dataN/hitN in numeric order using generated data definitions."""
|
|
runtime_class = ast.parse(runtime_block).body[0]
|
|
generated_class = ast.parse(generated_block).body[0]
|
|
if not isinstance(runtime_class, ast.ClassDef) or not isinstance(generated_class, ast.ClassDef):
|
|
return runtime_block
|
|
runtime_lines = runtime_block.splitlines(keepends=True)
|
|
generated_lines = generated_block.splitlines(keepends=True)
|
|
runtime_items = {
|
|
_statement_key(statement): (index, statement)
|
|
for index, statement in enumerate(runtime_class.body)
|
|
if _statement_key(statement)
|
|
}
|
|
generated_items = {
|
|
_statement_key(statement): (index, statement)
|
|
for index, statement in enumerate(generated_class.body)
|
|
if _statement_key(statement)
|
|
}
|
|
selected_statements = [
|
|
(index, statement, _statement_key(statement))
|
|
for index, statement in enumerate(runtime_class.body)
|
|
if _is_data_key(_statement_key(statement))
|
|
or re.fullmatch(r"hit\d+", _statement_key(statement) or "")
|
|
]
|
|
generated_data = {key: value for key, value in generated_items.items() if _is_data_key(key)}
|
|
if not selected_statements and not generated_data:
|
|
return runtime_block
|
|
|
|
def numeric_key(key: str) -> int:
|
|
return int(re.search(r"\d+$", key).group())
|
|
|
|
def segment(
|
|
lines: list[str], class_node: ast.ClassDef, index: int, statement: ast.stmt
|
|
) -> list[str]:
|
|
previous_end = class_node.body[index - 1].end_lineno if index else class_node.lineno
|
|
return lines[previous_end:statement.end_lineno]
|
|
|
|
# Remove every old dataN/hitN statement together with its leading comment
|
|
# trivia. They are re-emitted below in deterministic numeric order.
|
|
removals: list[tuple[int, int]] = []
|
|
for index, statement, _ in selected_statements:
|
|
previous_end = runtime_class.body[index - 1].end_lineno if index else runtime_class.lineno
|
|
removals.append((previous_end, statement.end_lineno))
|
|
if removals:
|
|
insertion_line = min(start for start, _ in removals)
|
|
else:
|
|
insertion_anchor = next(
|
|
(
|
|
statement
|
|
for statement in runtime_class.body
|
|
if _statement_key(statement) == "associate"
|
|
or isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
),
|
|
None,
|
|
)
|
|
insertion_line = (
|
|
insertion_anchor.lineno - 1
|
|
if insertion_anchor is not None
|
|
else runtime_class.end_lineno
|
|
)
|
|
for start, end in sorted(removals, reverse=True):
|
|
runtime_lines[start:end] = []
|
|
|
|
data_keys = sorted(
|
|
set(generated_data) | {key for key in runtime_items if _is_data_key(key)},
|
|
key=numeric_key,
|
|
)
|
|
rebuilt: list[str] = []
|
|
emitted_hits: set[str] = set()
|
|
for data_key in data_keys:
|
|
source_items = generated_items if data_key in generated_data else runtime_items
|
|
source_lines = generated_lines if data_key in generated_data else runtime_block.splitlines(keepends=True)
|
|
source_class = generated_class if data_key in generated_data else runtime_class
|
|
index, statement = source_items[data_key]
|
|
rebuilt.extend(segment(source_lines, source_class, index, statement))
|
|
|
|
hit_key = f"hit{numeric_key(data_key)}"
|
|
if hit_key in runtime_items:
|
|
_, hit_statement = runtime_items[hit_key]
|
|
rebuilt.extend(runtime_block.splitlines(keepends=True)[hit_statement.lineno - 1:hit_statement.end_lineno])
|
|
emitted_hits.add(hit_key)
|
|
|
|
# Retain unusual hitN fields even when no matching dataN exists.
|
|
for hit_key in sorted(
|
|
(
|
|
key
|
|
for key in runtime_items
|
|
if re.fullmatch(r"hit\d+", key) and key not in emitted_hits
|
|
),
|
|
key=numeric_key,
|
|
):
|
|
_, hit_statement = runtime_items[hit_key]
|
|
rebuilt.extend(runtime_block.splitlines(keepends=True)[hit_statement.lineno - 1:hit_statement.end_lineno])
|
|
|
|
runtime_lines[insertion_line:insertion_line] = rebuilt
|
|
return "".join(runtime_lines)
|
|
|
|
|
|
def _sync_generated_comments(runtime_block: str, generated_block: str) -> str:
|
|
"""Sync generated docstrings and leading field comments into runtime code."""
|
|
runtime_class = ast.parse(runtime_block).body[0]
|
|
generated_class = ast.parse(generated_block).body[0]
|
|
if not isinstance(runtime_class, ast.ClassDef) or not isinstance(generated_class, ast.ClassDef):
|
|
return runtime_block
|
|
|
|
runtime_lines = runtime_block.splitlines(keepends=True)
|
|
generated_lines = generated_block.splitlines(keepends=True)
|
|
replacements: list[tuple[int, int, list[str]]] = []
|
|
|
|
# The generated API description is authoritative, but executable methods
|
|
# and assignments continue to come from the runtime implementation.
|
|
runtime_doc = runtime_class.body[0] if runtime_class.body and isinstance(runtime_class.body[0], ast.Expr) and isinstance(runtime_class.body[0].value, ast.Constant) and isinstance(runtime_class.body[0].value.value, str) else None
|
|
generated_doc = generated_class.body[0] if generated_class.body and isinstance(generated_class.body[0], ast.Expr) and isinstance(generated_class.body[0].value, ast.Constant) and isinstance(generated_class.body[0].value.value, str) else None
|
|
if runtime_doc is not None and generated_doc is not None:
|
|
replacements.append(
|
|
(runtime_doc.lineno - 1, runtime_doc.end_lineno, generated_lines[generated_doc.lineno - 1:generated_doc.end_lineno])
|
|
)
|
|
elif runtime_doc is None and generated_doc is not None:
|
|
# Insert the generated skill description immediately below the class
|
|
# declaration when the old runtime class had no docstring.
|
|
replacements.append(
|
|
(
|
|
runtime_class.lineno,
|
|
runtime_class.lineno,
|
|
generated_lines[generated_doc.lineno - 1:generated_doc.end_lineno],
|
|
)
|
|
)
|
|
|
|
generated_by_key = {_statement_key(statement): (index, statement) for index, statement in enumerate(generated_class.body) if _statement_key(statement)}
|
|
for runtime_index, runtime_statement in enumerate(runtime_class.body):
|
|
key = _statement_key(runtime_statement)
|
|
if key not in generated_by_key:
|
|
continue
|
|
generated_index, generated_statement = generated_by_key[key]
|
|
runtime_previous_end = runtime_class.body[runtime_index - 1].end_lineno if runtime_index else runtime_class.lineno
|
|
generated_previous_end = generated_class.body[generated_index - 1].end_lineno if generated_index else generated_class.lineno
|
|
generated_trivia = generated_lines[generated_previous_end:generated_statement.lineno - 1]
|
|
# Only replace trivia when it carries actual generated comments. Blank
|
|
# spacing alone should not disturb hand-formatted runtime code.
|
|
if any(line.lstrip().startswith("#") for line in generated_trivia):
|
|
replacements.append(
|
|
(runtime_previous_end, runtime_statement.lineno - 1, generated_trivia)
|
|
)
|
|
|
|
for start, end, replacement in sorted(replacements, reverse=True):
|
|
runtime_lines[start:end] = replacement
|
|
return "".join(runtime_lines)
|
|
|
|
|
|
def _append_missing_methods(
|
|
runtime_block: str, generated_block: str, method_names: set[str]
|
|
) -> str:
|
|
"""Append generated class methods that the runtime skill does not define."""
|
|
runtime_class = ast.parse(runtime_block).body[0]
|
|
generated_class = ast.parse(generated_block).body[0]
|
|
if not isinstance(runtime_class, ast.ClassDef) or not isinstance(generated_class, ast.ClassDef):
|
|
return runtime_block
|
|
existing = {
|
|
statement.name
|
|
for statement in runtime_class.body
|
|
if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
}
|
|
generated_lines = generated_block.splitlines(keepends=True)
|
|
additions: list[str] = []
|
|
for statement in generated_class.body:
|
|
if not isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
if statement.name not in method_names or statement.name in existing:
|
|
continue
|
|
additions.extend(generated_lines[statement.lineno - 1:statement.end_lineno])
|
|
existing.add(statement.name)
|
|
if not additions:
|
|
return runtime_block
|
|
return runtime_block.rstrip() + "\n\n" + "".join(additions).rstrip() + "\n"
|
|
|
|
|
|
def _sync_generated_method_docstrings(
|
|
runtime_block: str, generated_block: str, method_names: set[str]
|
|
) -> str:
|
|
"""Use generated method docstrings while preserving runtime method bodies."""
|
|
runtime_class = ast.parse(runtime_block).body[0]
|
|
generated_class = ast.parse(generated_block).body[0]
|
|
if not isinstance(runtime_class, ast.ClassDef) or not isinstance(generated_class, ast.ClassDef):
|
|
return runtime_block
|
|
|
|
runtime_methods = {
|
|
statement.name: statement
|
|
for statement in runtime_class.body
|
|
if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
}
|
|
generated_methods = {
|
|
statement.name: statement
|
|
for statement in generated_class.body
|
|
if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
}
|
|
runtime_lines = runtime_block.splitlines(keepends=True)
|
|
generated_lines = generated_block.splitlines(keepends=True)
|
|
replacements: list[tuple[int, int, list[str]]] = []
|
|
|
|
def docstring_statement(method: ast.FunctionDef | ast.AsyncFunctionDef) -> ast.Expr | None:
|
|
if not method.body:
|
|
return None
|
|
statement = method.body[0]
|
|
if (
|
|
isinstance(statement, ast.Expr)
|
|
and isinstance(statement.value, ast.Constant)
|
|
and isinstance(statement.value.value, str)
|
|
):
|
|
return statement
|
|
return None
|
|
|
|
for name in method_names:
|
|
runtime_method = runtime_methods.get(name)
|
|
generated_method = generated_methods.get(name)
|
|
if runtime_method is None or generated_method is None:
|
|
continue
|
|
generated_doc = docstring_statement(generated_method)
|
|
if generated_doc is None:
|
|
continue
|
|
generated_source = generated_lines[
|
|
generated_doc.lineno - 1:generated_doc.end_lineno
|
|
]
|
|
runtime_doc = docstring_statement(runtime_method)
|
|
if runtime_doc is not None:
|
|
replacements.append(
|
|
(runtime_doc.lineno - 1, runtime_doc.end_lineno, generated_source)
|
|
)
|
|
else:
|
|
# A method without a docstring starts directly with executable
|
|
# code. Insert the generated VP description before that code.
|
|
replacements.append(
|
|
(runtime_method.body[0].lineno - 1, runtime_method.body[0].lineno - 1, generated_source)
|
|
)
|
|
|
|
for start, end, replacement in sorted(replacements, reverse=True):
|
|
runtime_lines[start:end] = replacement
|
|
return "".join(runtime_lines)
|
|
|
|
|
|
def _ensure_get_data_import(source: str) -> str:
|
|
if not re.search(r"(?m)^\s*data\d+\s*=\s*get_data\(", source):
|
|
return source
|
|
pattern = re.compile(r"(?m)^(from\s+core\.basic\.skill\s+import\s+)([^\n]+)$")
|
|
match = pattern.search(source)
|
|
if match is None:
|
|
raise ValueError("generated data requires a core.basic.skill import")
|
|
imports = [item.strip() for item in match.group(2).split(",")]
|
|
if "get_data" in imports:
|
|
return source
|
|
imports.append("get_data")
|
|
return source[:match.start()] + match.group(1) + ", ".join(imports) + source[match.end():]
|
|
|
|
|
|
def _sync_module_assignment(
|
|
runtime_source: str, generated_source: str, field: str
|
|
) -> str:
|
|
"""Make a module-level assignment match the generated module."""
|
|
runtime_tree = ast.parse(runtime_source)
|
|
generated_tree = ast.parse(generated_source)
|
|
generated_statement = next(
|
|
(
|
|
statement
|
|
for statement in generated_tree.body
|
|
if _statement_key(statement) == field
|
|
),
|
|
None,
|
|
)
|
|
if generated_statement is None:
|
|
return runtime_source
|
|
runtime_statements = [
|
|
statement
|
|
for statement in runtime_tree.body
|
|
if _statement_key(statement) == field
|
|
]
|
|
runtime_lines = runtime_source.splitlines(keepends=True)
|
|
generated_lines = generated_source.splitlines(keepends=True)
|
|
assignment = generated_lines[
|
|
generated_statement.lineno - 1:generated_statement.end_lineno
|
|
]
|
|
if runtime_statements:
|
|
insertion_line = runtime_statements[0].lineno - 1
|
|
for statement in sorted(runtime_statements, key=lambda item: item.lineno, reverse=True):
|
|
runtime_lines[statement.lineno - 1:statement.end_lineno] = []
|
|
replacement = assignment
|
|
else:
|
|
imports = [
|
|
statement
|
|
for statement in runtime_tree.body
|
|
if isinstance(statement, (ast.Import, ast.ImportFrom))
|
|
]
|
|
insertion_line = imports[-1].end_lineno if imports else 0
|
|
replacement = [*assignment, "\n"]
|
|
runtime_lines[insertion_line:insertion_line] = replacement
|
|
return "".join(runtime_lines)
|
|
|
|
|
|
def _comment(block: str, name: str) -> str:
|
|
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")
|
|
generated_skills = _skills(generated, generated_path)
|
|
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 = _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)
|
|
chunks = [character[:character_header_end].rstrip() + "\n\n"]
|
|
chunks.append(generated[generated_header_end:generated_skills[0].start])
|
|
cursor = generated_skills[0].start
|
|
matched = commented = 0
|
|
for index, generated_skill in enumerate(generated_skills):
|
|
if index:
|
|
chunks.append(generated[cursor:generated_skill.start])
|
|
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)
|
|
block = _set_generated_field(block, "position", generated_skill.position)
|
|
block = _set_generated_field(block, "cellPos", generated_skill.cell_pos)
|
|
block = _append_missing_methods(block, generated_block, {"vp_1", "vp_2"})
|
|
block = _sync_generated_method_docstrings(
|
|
block, generated_block, {"vp_1", "vp_2"}
|
|
)
|
|
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,
|
|
"runtime_only": len(character_skills) - matched,
|
|
}
|
|
|
|
|
|
def _atomic_write(path: Path, content: str) -> None:
|
|
"""Write beside the target and atomically replace it on Windows.
|
|
|
|
Writing an already-open Python source file directly can intermittently
|
|
produce ``OSError: [Errno 22]`` on Windows. A sibling temporary file also
|
|
prevents a failed write from leaving a partially truncated module.
|
|
"""
|
|
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
try:
|
|
temporary.write_bytes(content.encode("utf-8"))
|
|
for attempt in range(3):
|
|
try:
|
|
os.replace(temporary, path)
|
|
return
|
|
except OSError:
|
|
if attempt == 2:
|
|
raise
|
|
time.sleep(0.1 * (attempt + 1))
|
|
except OSError as error:
|
|
raise OSError(
|
|
error.errno,
|
|
f"failed to replace {path} using temporary file {temporary}: {error.strerror}",
|
|
str(path),
|
|
) from error
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
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 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))
|
|
if not args.check:
|
|
_atomic_write(character_path, output)
|
|
totals["files"] += 1
|
|
for key, value in stats.items():
|
|
totals[key] += value
|
|
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
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|