"""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.
"""

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


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)
    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
        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 == "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,
            )
        )
    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 _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 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 = {skill.skill_name: skill for skill in character_skills}
    if len(runtime) != len(character_skills):
        raise ValueError(f"duplicate skill name: {character_path}")

    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 = runtime.get(generated_skill.skill_name)
        if runtime_skill is None:
            chunks.append(_comment(generated[generated_skill.start:generated_skill.end], generated_skill.skill_name))
            commented += 1
        else:
            block = character[runtime_skill.start:runtime_skill.end]
            generated_block = generated[generated_skill.start:generated_skill.end]
            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"})
            chunks.append(block.rstrip() + "\n")
            matched += 1
        cursor = generated_skill.end
    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)
    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")
    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")
            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'} {generated_path.relative_to(GENERATED_ROOT)}: "
              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())