"""Generate and integrate all SakuraMemory heroine lines with edge-tts."""

from __future__ import annotations

import asyncio
import json
from pathlib import Path
import re
import sys

import edge_tts


EXPECTED_LINE_COUNT = 56
VOICE = "zh-CN-XiaoxiaoNeural"
RATE = "-8%"
PITCH = "+0Hz"
VOLUME = "+0%"

TOOLS_DIR = Path(__file__).resolve().parent
PROJECT_DIR = TOOLS_DIR.parent
SCRIPT_PATH = PROJECT_DIR / "game" / "script.rpy"
VOICE_DIR = PROJECT_DIR / "game" / "audio" / "voice" / "mio"
MANIFEST_PATH = VOICE_DIR / "voice_manifest.json"
BACKUP_PATH = SCRIPT_PATH.with_suffix(".rpy.before_voice")

MIO_LINE = re.compile(r'^(?P<indent>\s*)mio\s+"(?P<text>(?:[^"\\]|\\.)*)"\s*$')
GENERATED_VOICE = re.compile(
    r'^\s*voice\s+"audio/voice/mio/mio_\d{3}\.mp3"\s*$'
)


def decode_renpy_text(value: str) -> str:
    return value.replace(r'\"', '"').replace(r"\\", "\\")


def is_valid_mp3(path: Path) -> bool:
    if not path.is_file() or path.stat().st_size < 1024:
        return False
    with path.open("rb") as audio:
        header = audio.read(3)
    return header == b"ID3" or header[:1] == b"\xff"


def load_dialogue_lines() -> tuple[list[str], list[dict[str, object]]]:
    original_lines = SCRIPT_PATH.read_text(encoding="utf-8").splitlines()
    clean_lines = [line for line in original_lines if not GENERATED_VOICE.match(line)]
    entries: list[dict[str, object]] = []

    for line_number, line in enumerate(clean_lines, start=1):
        match = MIO_LINE.match(line)
        if not match:
            continue
        number = len(entries) + 1
        entries.append(
            {
                "number": number,
                "line_number": line_number,
                "indent": match.group("indent"),
                "text": decode_renpy_text(match.group("text")),
                "file": f"mio_{number:03d}.mp3",
            }
        )

    if len(entries) != EXPECTED_LINE_COUNT:
        raise RuntimeError(
            f"预计找到 {EXPECTED_LINE_COUNT} 句樱井澪台词，实际找到 {len(entries)} 句。"
            "为避免错配，已停止生成。"
        )
    return clean_lines, entries


async def generate_one(entry: dict[str, object], semaphore: asyncio.Semaphore) -> None:
    number = int(entry["number"])
    output = VOICE_DIR / str(entry["file"])
    if is_valid_mp3(output):
        print(f"[{number:02d}/{EXPECTED_LINE_COUNT}] 已存在，跳过：{output.name}")
        return

    async with semaphore:
        last_error: Exception | None = None
        for attempt in range(1, 4):
            try:
                temporary = output.with_suffix(".mp3.part")
                if temporary.exists():
                    temporary.unlink()
                communicator = edge_tts.Communicate(
                    text=str(entry["text"]),
                    voice=VOICE,
                    rate=RATE,
                    pitch=PITCH,
                    volume=VOLUME,
                )
                await communicator.save(str(temporary))
                if not is_valid_mp3(temporary):
                    raise RuntimeError("服务返回的音频无效或为空")
                temporary.replace(output)
                print(f"[{number:02d}/{EXPECTED_LINE_COUNT}] 完成：{output.name}")
                return
            except Exception as error:  # retry transient network/service failures
                last_error = error
                print(f"[{number:02d}/{EXPECTED_LINE_COUNT}] 第 {attempt} 次失败：{error}")
                if attempt < 3:
                    await asyncio.sleep(attempt * 2)
        raise RuntimeError(f"{output.name} 生成失败：{last_error}")


async def generate_all(entries: list[dict[str, object]]) -> None:
    semaphore = asyncio.Semaphore(3)
    tasks = [asyncio.create_task(generate_one(entry, semaphore)) for entry in entries]
    await asyncio.gather(*tasks)


def integrate_script(clean_lines: list[str], entries: list[dict[str, object]]) -> None:
    entry_iter = iter(entries)
    current = next(entry_iter, None)
    integrated: list[str] = []

    for line in clean_lines:
        if MIO_LINE.match(line):
            if current is None:
                raise RuntimeError("配音接入时台词数量发生变化。")
            integrated.append(
                f'{current["indent"]}voice "audio/voice/mio/{current["file"]}"'
            )
            current = next(entry_iter, None)
        integrated.append(line)

    if current is not None:
        raise RuntimeError("配音接入未覆盖全部台词。")

    if not BACKUP_PATH.exists():
        BACKUP_PATH.write_text("\n".join(clean_lines) + "\n", encoding="utf-8")

    temporary = SCRIPT_PATH.with_suffix(".rpy.voice_tmp")
    temporary.write_text("\n".join(integrated) + "\n", encoding="utf-8")
    temporary.replace(SCRIPT_PATH)


def write_manifest(entries: list[dict[str, object]]) -> None:
    manifest = {
        "character": "樱井澪",
        "voice": VOICE,
        "rate": RATE,
        "pitch": PITCH,
        "volume": VOLUME,
        "line_count": len(entries),
        "lines": [
            {"number": entry["number"], "file": entry["file"], "text": entry["text"]}
            for entry in entries
        ],
    }
    MANIFEST_PATH.write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
    )


def main() -> int:
    try:
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stderr.reconfigure(encoding="utf-8")
        VOICE_DIR.mkdir(parents=True, exist_ok=True)
        clean_lines, entries = load_dialogue_lines()
        print(f"找到 {len(entries)} 句樱井澪台词，开始在线生成。")
        asyncio.run(generate_all(entries))

        invalid = [entry["file"] for entry in entries if not is_valid_mp3(VOICE_DIR / str(entry["file"]))]
        if invalid:
            raise RuntimeError("以下音频未通过检查：" + ", ".join(map(str, invalid)))

        write_manifest(entries)
        integrate_script(clean_lines, entries)
        print("全部 56 句配音已生成、验证并接入 game/script.rpy。")
        return 0
    except Exception as error:
        print(f"生成失败：{error}", file=sys.stderr)
        print("script.rpy 未接入不完整音频，可以稍后重新运行本工具。", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

