#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""修复 edge-tts 的 403 问题（微软 2026-10 更换 Edge Read Aloud 端点）。

【现象】
    合成语音时报：
        aiohttp.client_exceptions.WSServerHandshakeError: 403, message='Invalid response status'
    同时伴随这些特征（可用来确认不是自己的环境问题）：
        - `edge_tts --list-voices`（普通 HTTPS）仍然正常返回
        - 系统时钟已 NTP 同步，偏差在几秒内
        - edge-tts 已经是最新版（PyPI 上的 7.2.8）
        - 换网络 / 换机器同样 403

【原因】
    微软把 Edge Read Aloud 的「合成」端点从旧地址
        speech.platform.bing.com/consumer/speech/synthesize/readaloud/edge/v1
    迁到了新地址
        api.msedgeservices.com/tts/cognitiveservices/websocket/v1
    旧端点在 2026-10-10 起于多数地区开始返回 403，2026-10-12 起全球（含美国）
    全线 403。而 edge-tts 7.2.8（2025-12 发布）尚未跟进。
    上游讨论：https://github.com/rany2/edge-tts/issues/490
    上游修复：https://github.com/rany2/edge-tts/pull/491 （未合并）

【做法】
    把 edge_tts/constants.py 中的 WSS_URL 指向新端点。新端点同样接受旧 token，
    参数名改为 Ocp-Apim-Subscription-Key；Sec-MS-GEC / Sec-MS-GEC-Version
    继续沿用原来的 v1 形式即可，无需 v2 token。

【用法】
    python3 patch_edge_tts_403.py            # 打补丁
    python3 patch_edge_tts_403.py --check    # 只看当前状态，不改动
    python3 patch_edge_tts_403.py --verify   # 打补丁后实际合成一段音频来验证
    python3 patch_edge_tts_403.py --revert   # 还原成 edge-tts 的原始端点

【注意】
    重新安装或升级 edge-tts 会覆盖 constants.py，补丁随之丢失，重跑本脚本即可。
    新端点对同一 IP 有限流（429），脚本验证时会自动重试几次。
"""

import argparse
import importlib.util
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time

# 补丁块的起始标记。用它来界定「补丁插入的内容」，
# 这样 --revert 能把注释和赋值语句一起清干净，逐字节还原成 edge-tts 原文件。
MARKER = "# edge-tts-403-fix (PA_horn_speaker)"

ORIGINAL_WSS_URL = 'WSS_URL = f"wss://{BASE_URL}/edge/v1?TrustedClientToken={TRUSTED_CLIENT_TOKEN}"\n'

PATCHED_WSS_URL = "\n".join([
    MARKER,
    "# 2026-10 起微软把 Edge Read Aloud 合成从旧端点 speech.platform.bing.com",
    "# 迁到了 api.msedgeservices.com（旧端点在多数地区返回 403）。",
    "# 新端点同样接受旧 token，参数名为 Ocp-Apim-Subscription-Key；",
    "# Sec-MS-GEC / Sec-MS-GEC-Version 仍沿用旧的 v1 形式即可。",
    "# 语音列表接口仍由旧端点提供，故 VOICE_LIST 保持不变。",
    "WSS_URL = (",
    '    "wss://api.msedgeservices.com/tts/cognitiveservices/websocket/v1"',
    '    f"?Ocp-Apim-Subscription-Key={TRUSTED_CLIENT_TOKEN}"',
    ")",
    "",
])

# 匹配「可选的补丁块」+「WSS_URL 赋值语句」（赋值可能是单行或跨行括号形式），
# 直到下一行的 VOICE_LIST 定义为止。
WSS_BLOCK = re.compile(
    r"(?:^" + re.escape(MARKER) + r"\n(?:^#[^\n]*\n)*)?"
    r"^WSS_URL = .*?\n(?=^VOICE_LIST = )",
    re.S | re.M,
)


def find_constants_path():
    """定位已安装的 edge_tts/constants.py。"""
    spec = importlib.util.find_spec("edge_tts")
    if spec is None or not spec.origin:
        raise SystemExit("找不到 edge_tts，请先安装：pip install edge-tts")
    return os.path.join(os.path.dirname(spec.origin), "constants.py")


def read(path):
    with open(path, "r", encoding="utf-8") as f:
        return f.read()


def is_patched(text):
    return "api.msedgeservices.com" in text


def do_check(path):
    text = read(path)
    if not WSS_BLOCK.search(text):
        raise SystemExit("未能在 %s 中定位 WSS_URL，edge-tts 结构可能已变化" % path)
    print("edge-tts constants.py：%s" % path)
    print("状态：%s" % ("已打补丁（指向新端点）" if is_patched(text) else "未打补丁（仍是旧端点，会 403）"))
    return 0


def do_patch(path, revert=False):
    text = read(path)
    if not WSS_BLOCK.search(text):
        raise SystemExit("未能在 %s 中定位 WSS_URL，edge-tts 结构可能已变化" % path)

    patched = is_patched(text)
    if revert:
        if not patched:
            print("当前已是 edge-tts 的原始端点，无需还原。")
            return 0
        replacement = ORIGINAL_WSS_URL
    else:
        if patched:
            print("当前已指向新端点，无需重复打补丁。")
            return 0
        replacement = PATCHED_WSS_URL

    backup = path + ".bak"
    if not os.path.exists(backup):
        shutil.copy2(path, backup)
        print("已备份原文件到 %s" % backup)

    new_text = WSS_BLOCK.sub(lambda _: replacement, text, count=1)
    with open(path, "w", encoding="utf-8", newline="") as f:
        f.write(new_text)

    print("已%s：%s" % ("还原为原始端点" if revert else "指向新端点", path))
    if not revert:
        print("下一步可用 `python3 %s --verify` 实际合成一段音频确认。" % os.path.basename(__file__))
    return 0


def do_verify(attempts=3):
    """实际合成一段音频，确认真能拿到数据（而不只是改了个变量）。"""
    out = os.path.join(tempfile.gettempdir(), "edge_tts_verify.mp3")
    for i in range(1, attempts + 1):
        if os.path.exists(out):
            os.remove(out)
        proc = subprocess.run(
            [sys.executable, "-m", "edge_tts",
             "--voice", "zh-CN-XiaoxiaoNeural",
             "--text", "测试广播，今天水果特价。",
             "--write-media", out],
            capture_output=True, text=True,
        )
        if proc.returncode == 0 and os.path.exists(out) and os.path.getsize(out) > 0:
            print("验证通过：成功合成 %d 字节音频 -> %s" % (os.path.getsize(out), out))
            return 0
        tail = (proc.stderr or "").strip().splitlines()
        print("第 %d/%d 次失败：%s" % (i, attempts, tail[-1] if tail else "无输出"))
        if i < attempts:
            time.sleep(2 * i)
    print("验证失败。若错误仍是 403，说明端点又变了，请查看 "
          "https://github.com/rany2/edge-tts/issues 上的最新讨论。")
    print("若错误是 429，说明只是被限流，稍后重试即可。")
    return 1


def main():
    ap = argparse.ArgumentParser(
        description="修复 edge-tts 的 403（微软 2026-10 更换 Edge Read Aloud 端点）")
    ap.add_argument("--check", action="store_true", help="只检查当前状态，不改动文件")
    ap.add_argument("--revert", action="store_true", help="还原为 edge-tts 的原始端点")
    ap.add_argument("--verify", action="store_true", help="实际合成一段音频来验证")
    ap.add_argument("--constants", metavar="PATH",
                    help="手动指定 constants.py 路径（默认自动定位已安装的 edge_tts）")
    args = ap.parse_args()

    if args.check:
        return do_check(args.constants or find_constants_path())

    if args.verify:
        return do_verify()

    path = args.constants or find_constants_path()
    return do_patch(path, revert=args.revert)


if __name__ == "__main__":
    sys.exit(main())