#!/usr/bin/env python3
"""用 Chrome DevTools Protocol 模拟 Flutter WebView 打开线上 ZGame。

脚本复刻 Flutter 容器最关键的三个边界：先向线上网关登录并签发一次性
launch token；在文档首行注入 launch payload/config；再执行后端下发的远程
bridge 脚本。账号密码只从环境变量读取，避免进入代码、shell history 或截图。
"""

from __future__ import annotations

import argparse
import base64
import json
import os
import pathlib
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any

try:
    import websocket
except ImportError as exc:  # pragma: no cover - 只在本机缺少运行依赖时触发。
    raise SystemExit(
        "缺少 websocket-client，请先执行: python3 -m pip install --user websocket-client"
    ) from exc


DEFAULT_API_BASE = "https://api.global-interaction.com/api/v1"
DEFAULT_CHROME = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
LOCAL_HTTP_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="登录线上账号，并用 Flutter WebView 同等桥接顺序打开 ZGame。"
    )
    parser.add_argument("--api-base", default=DEFAULT_API_BASE)
    parser.add_argument("--app-code", default="lalu")
    parser.add_argument("--account", default="123456")
    parser.add_argument("--game-id", default="zgame_49")
    parser.add_argument("--password-env", default="HYAPP_TEST_PASSWORD")
    parser.add_argument("--chrome", default=DEFAULT_CHROME)
    parser.add_argument("--timeout", type=int, default=60)
    parser.add_argument(
        "--screenshot",
        default=".tmp/zgame-flutter-container-check.png",
        help="验证截图输出位置；.tmp 已被仓库忽略。",
    )
    return parser.parse_args()


def http_json(
    url: str,
    *,
    method: str = "GET",
    body: dict[str, Any] | None = None,
    headers: dict[str, str] | None = None,
    timeout: int = 20,
) -> dict[str, Any]:
    encoded = None if body is None else json.dumps(body).encode("utf-8")
    request_headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        **(headers or {}),
    }
    request = urllib.request.Request(
        url,
        data=encoded,
        headers=request_headers,
        method=method,
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except urllib.error.HTTPError as exc:
        # 错误正文通常包含明确业务码；只输出响应，不输出请求体中的密码/token。
        detail = exc.read().decode("utf-8", errors="replace")[:1000]
        raise RuntimeError(f"HTTP {exc.code} {url}: {detail}") from exc
    if not isinstance(payload, dict):
        raise RuntimeError(f"接口未返回 JSON object: {url}")
    return payload


def require_envelope(payload: dict[str, Any], operation: str) -> dict[str, Any]:
    code = str(payload.get("code", "")).upper()
    if code not in {"OK", "0"}:
        raise RuntimeError(
            f"{operation}失败: code={payload.get('code')} message={payload.get('message')}"
        )
    data = payload.get("data")
    if not isinstance(data, dict):
        raise RuntimeError(f"{operation}响应缺少 data")
    return data


def reserve_port() -> int:
    # 先由内核选择空闲端口，Chrome 随后立即占用；本机验证避免固定调试端口冲突。
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def wait_json(url: str, timeout_seconds: int) -> dict[str, Any]:
    deadline = time.monotonic() + timeout_seconds
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            # 本机 CDP 不能经过系统 HTTP(S)_PROXY，否则 127.0.0.1 会被代理吞掉并超时。
            with LOCAL_HTTP_OPENER.open(url, timeout=1) as response:
                value = json.load(response)
            if isinstance(value, dict):
                return value
        except Exception as exc:  # noqa: BLE001 - Chrome 启动期间连接失败是预期状态。
            last_error = exc
        time.sleep(0.1)
    raise RuntimeError(f"Chrome DevTools 未就绪: {last_error}")


class CDPClient:
    """最小 CDP 客户端，只实现本验证所需的调用、事件收集和超时边界。"""

    def __init__(self, websocket_url: str, timeout_seconds: int) -> None:
        endpoint = urllib.parse.urlsplit(websocket_url)
        # websocket-client 即使显式传空 proxy 也可能继续读取系统代理；预建 TCP
        # socket 可确保 CDP 永远直连本机 Chrome，不把调试流量送出机器。
        raw_socket = socket.create_connection(
            (endpoint.hostname or "127.0.0.1", endpoint.port or 80),
            timeout=timeout_seconds,
        )
        self._socket = websocket.create_connection(
            websocket_url,
            timeout=timeout_seconds,
            origin="http://127.0.0.1",
            socket=raw_socket,
        )
        self._socket.settimeout(1)
        self._next_id = 1
        self.events: list[dict[str, Any]] = []

    def close(self) -> None:
        self._socket.close()

    def call(
        self,
        method: str,
        params: dict[str, Any] | None = None,
        *,
        timeout_seconds: int = 15,
    ) -> dict[str, Any]:
        request_id = self._next_id
        self._next_id += 1
        self._socket.send(
            json.dumps({"id": request_id, "method": method, "params": params or {}})
        )
        deadline = time.monotonic() + timeout_seconds
        while time.monotonic() < deadline:
            try:
                message = json.loads(self._socket.recv())
            except websocket.WebSocketTimeoutException:
                continue
            if message.get("id") == request_id:
                if "error" in message:
                    raise RuntimeError(f"CDP {method}失败: {message['error']}")
                result = message.get("result")
                return result if isinstance(result, dict) else {}
            self.events.append(message)
        raise RuntimeError(f"CDP {method}调用超时")

    def drain(self, duration_seconds: float = 0.1) -> None:
        deadline = time.monotonic() + duration_seconds
        while time.monotonic() < deadline:
            try:
                self.events.append(json.loads(self._socket.recv()))
            except websocket.WebSocketTimeoutException:
                return


def fetch_text(url: str, timeout: int = 20) -> str:
    request = urllib.request.Request(url, headers={"Accept": "application/javascript,*/*"})
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read().decode("utf-8")


def build_flutter_document_start_script(
    *, game_id: str, launch_url: str, session_id: str, bridge_source: str
) -> str:
    query = dict(urllib.parse.parse_qsl(urllib.parse.urlsplit(launch_url).query))
    # Flutter 会把 URL query、后端 launch data 和目录元数据合并；ZGame 取身份字段时
    # 依赖 query 中的 openId/js_code，adapterType 则决定安装 ThirdPlatformBridge。
    config: dict[str, Any] = {
        **query,
        "adapterType": "zgame_v1",
        "adapter_type": "zgame_v1",
        "platformCode": "zgame",
        "platform_code": "zgame",
    }
    payload = {
        "sessionId": session_id,
        "gameSessionId": session_id,
        "gameId": game_id,
        "provider": "zgame",
        "vendorType": "zgame",
        "launchUrl": launch_url,
        "launchConfig": config,
        "entry": {"entryUrl": launch_url},
    }
    # 原生通道在模拟器中只记录动作；用户身份和 login 凭证由远程 bridge 同步返回，
    # 不需要伪造厂商回调，也不会把 token 发送到除即构与线上 API 之外的域名。
    bootstrap = f"""
window.hyappGameLaunchPayload = {json.dumps(payload, ensure_ascii=False)};
window.hyappGameLaunchConfig = {json.dumps(config, ensure_ascii=False)};
window.hyappGameEntry = {json.dumps(payload['entry'], ensure_ascii=False)};
window.addEventListener('unhandledrejection', function(event) {{
  var reason = event && event.reason;
  try {{ console.error('[FlutterSimulatorUnhandledRejection]', JSON.stringify(reason)); }}
  catch (_) {{ console.error('[FlutterSimulatorUnhandledRejection]', String(reason)); }}
}});
window.addEventListener('error', function(event) {{
  console.error('[FlutterSimulatorError]', event && event.message || 'unknown error');
}});
window.GameBridgeChannel = {{
  postMessage: function(message) {{ console.info('[FlutterBridge]', message); }}
}};
window.webkit = window.webkit || {{}};
window.webkit.messageHandlers = window.webkit.messageHandlers || {{}};
window.webkit.messageHandlers.GameBridgeChannel = window.GameBridgeChannel;
"""
    return bootstrap + "\n;\n" + bridge_source


def evaluate_value(cdp: CDPClient, expression: str) -> Any:
    result = cdp.call(
        "Runtime.evaluate",
        {"expression": expression, "returnByValue": True, "awaitPromise": True},
    )
    remote = result.get("result")
    if not isinstance(remote, dict):
        return None
    return remote.get("value")


def browser_state(cdp: CDPClient) -> dict[str, Any]:
    expression = r"""
(() => {
  const visible = (element) => {
    const rect = element.getBoundingClientRect();
    const style = getComputedStyle(element);
    return rect.width > 2 && rect.height > 2 && style.visibility !== 'hidden' && style.display !== 'none';
  };
  const canvases = [...document.querySelectorAll('canvas')];
  return {
    readyState: document.readyState,
    title: document.title,
    bodyText: (document.body && document.body.innerText || '').trim().slice(0, 500),
    bridgeReady: !!(window.ThirdPlatformBridge && window.ThirdPlatformBridge.__hyappZGameBridge),
    canvasCount: canvases.length,
    visibleCanvasCount: canvases.filter(visible).length,
    largestCanvasPixels: canvases.reduce((max, canvas) => Math.max(max, canvas.width * canvas.height), 0),
    imageCount: document.images.length,
    href: location.href
  };
})()
"""
    value = evaluate_value(cdp, expression)
    return value if isinstance(value, dict) else {}


def relevant_browser_errors(events: list[dict[str, Any]]) -> list[str]:
    errors: list[str] = []
    for event in events:
        method = event.get("method")
        params = event.get("params") or {}
        if method == "Runtime.exceptionThrown":
            detail = params.get("exceptionDetails") or {}
            exception = detail.get("exception") or {}
            text = str(
                exception.get("description")
                or detail.get("text")
                or exception.get("value")
                or ""
            )
            if text:
                errors.append(text[:500])
        elif method == "Log.entryAdded":
            entry = params.get("entry") or {}
            if entry.get("level") == "error":
                errors.append(str(entry.get("text") or "")[:500])
        elif method == "Runtime.consoleAPICalled":
            if params.get("type") not in {"error", "warning"}:
                continue
            values = []
            for argument in params.get("args") or []:
                if not isinstance(argument, dict):
                    continue
                values.append(str(argument.get("value") or argument.get("description") or ""))
            if values:
                errors.append("console: " + " ".join(values)[:2000])
        elif method == "Network.loadingFailed":
            errors.append(
                "network: "
                + str(params.get("errorText") or "loading failed")
                + " "
                + str(params.get("blockedReason") or "")
            )
        elif method == "Network.responseReceived":
            response = params.get("response") or {}
            status = int(response.get("status") or 0)
            if status >= 400:
                errors.append(f"http {status}: {response.get('url')}")
    return list(dict.fromkeys(error for error in errors if error))


def browser_network_trace(events: list[dict[str, Any]]) -> list[str]:
    """保留末尾 XHR/fetch，避免把图片/CDN 噪声混进业务登录失败诊断。"""
    trace: list[str] = []
    for event in events:
        if event.get("method") != "Network.responseReceived":
            continue
        params = event.get("params") or {}
        if params.get("type") not in {"XHR", "Fetch"}:
            continue
        response = params.get("response") or {}
        trace.append(f"{int(response.get('status') or 0)} {response.get('url')}")
    return trace[-20:]


def run_browser_check(
    *,
    chrome_path: str,
    launch_url: str,
    game_id: str,
    session_id: str,
    bridge_source: str,
    screenshot_path: pathlib.Path,
    timeout_seconds: int,
) -> tuple[dict[str, Any], list[str]]:
    if not pathlib.Path(chrome_path).is_file():
        raise RuntimeError(f"找不到 Chrome: {chrome_path}")
    port = reserve_port()
    profile_dir = tempfile.mkdtemp(prefix="zgame-flutter-container-")
    process = subprocess.Popen(
        [
            chrome_path,
            "--headless=new",
            f"--remote-debugging-port={port}",
            "--remote-allow-origins=*",
            f"--user-data-dir={profile_dir}",
            "--no-first-run",
            "--no-default-browser-check",
            "--disable-background-networking",
            "about:blank",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    cdp: CDPClient | None = None
    try:
        wait_json(f"http://127.0.0.1:{port}/json/version", 10)
        with LOCAL_HTTP_OPENER.open(
            f"http://127.0.0.1:{port}/json/list", timeout=2
        ) as response:
            targets = json.load(response)
        page = next((item for item in targets if item.get("type") == "page"), None)
        if not isinstance(page, dict) or not page.get("webSocketDebuggerUrl"):
            raise RuntimeError("Chrome 未创建可调试页面")
        cdp = CDPClient(str(page["webSocketDebuggerUrl"]), timeout_seconds)
        for domain in ("Page.enable", "Runtime.enable", "Log.enable", "Network.enable"):
            cdp.call(domain)
        cdp.call(
            "Emulation.setDeviceMetricsOverride",
            {
                "width": 390,
                "height": 844,
                "deviceScaleFactor": 2,
                "mobile": True,
            },
        )
        cdp.call(
            "Network.setUserAgentOverride",
            {
                "userAgent": (
                    "Mozilla/5.0 (Linux; Android 14; FlutterWebView) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
                    "Chrome/136.0.0.0 Mobile Safari/537.36"
                )
            },
        )
        document_start = build_flutter_document_start_script(
            game_id=game_id,
            launch_url=launch_url,
            session_id=session_id,
            bridge_source=bridge_source,
        )
        cdp.call("Page.addScriptToEvaluateOnNewDocument", {"source": document_start})
        cdp.call("Page.navigate", {"url": launch_url})

        deadline = time.monotonic() + timeout_seconds
        state: dict[str, Any] = {}
        while time.monotonic() < deadline:
            time.sleep(1)
            cdp.drain(0.05)
            state = browser_state(cdp)
            # 厂商加载页也有背景素材；必须同时看到桥已安装和可见 canvas，才算
            # 通过授权阶段进入游戏渲染，而不是把 loading 背景误报为成功。
            if state.get("bridgeReady") and int(state.get("visibleCanvasCount") or 0) > 0:
                break

        screenshot_path.parent.mkdir(parents=True, exist_ok=True)
        screenshot = cdp.call(
            "Page.captureScreenshot",
            {"format": "png", "captureBeyondViewport": False},
        )
        screenshot_path.write_bytes(base64.b64decode(str(screenshot.get("data") or "")))
        cdp.drain(0.2)
        errors = relevant_browser_errors(cdp.events)
        network_trace = browser_network_trace(cdp.events)
        if not state.get("bridgeReady"):
            raise RuntimeError(
                "ThirdPlatformBridge 未安装，Flutter 容器模拟失败; "
                + json.dumps({"errors": errors[:10], "xhr": network_trace}, ensure_ascii=False)
            )
        if int(state.get("visibleCanvasCount") or 0) <= 0:
            raise RuntimeError(
                f"游戏未进入 canvas 渲染，最终页面状态: {json.dumps(state, ensure_ascii=False)}; "
                f"诊断: {json.dumps({'errors': errors[:10], 'xhr': network_trace}, ensure_ascii=False)}"
            )
        return state, errors
    finally:
        if cdp is not None:
            cdp.close()
        process.terminate()
        try:
            process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
        shutil.rmtree(profile_dir, ignore_errors=True)


def main() -> int:
    args = parse_args()
    password = os.environ.get(args.password_env, "").strip()
    if not password:
        raise RuntimeError(f"请通过环境变量 {args.password_env} 提供线上测试密码")
    api_base = args.api_base.rstrip("/")
    common_headers = {"X-App-Code": args.app_code}

    login = require_envelope(
        http_json(
            f"{api_base}/auth/account/login",
            method="POST",
            body={
                "account": args.account,
                "password": password,
                "device_id": "zgame-flutter-container-check",
            },
            headers=common_headers,
        ),
        "线上账号登录",
    )
    access_token = str(login.get("access_token") or "").strip()
    if not access_token:
        raise RuntimeError("线上账号登录响应缺少 access_token")
    auth_headers = {**common_headers, "Authorization": f"Bearer {access_token}"}

    games = require_envelope(
        http_json(f"{api_base}/games?scene=voice_room", headers=auth_headers),
        "读取游戏目录",
    )
    catalog = games.get("games")
    if not isinstance(catalog, list) or not any(
        isinstance(item, dict) and item.get("game_id") == args.game_id
        for item in catalog
    ):
        raise RuntimeError(f"线上 voice_room 目录找不到 {args.game_id}")

    launch = require_envelope(
        http_json(
            f"{api_base}/games/{urllib.parse.quote(args.game_id)}/launch",
            method="POST",
            body={},
            headers=auth_headers,
        ),
        "签发游戏启动会话",
    )
    launch_url = str(launch.get("launch_url") or "").strip()
    session_id = str(launch.get("session_id") or "").strip()
    if not launch_url or not session_id:
        raise RuntimeError("启动响应缺少 launch_url/session_id")
    query = dict(urllib.parse.parse_qsl(urllib.parse.urlsplit(launch_url).query))
    bridge_url = str(query.get("bridge_script_url") or query.get("bridgeScriptUrl") or "")
    open_id = str(query.get("open_id") or query.get("openId") or "")
    js_code = str(query.get("js_code") or query.get("jscode") or "")
    callback_base = str(query.get("callback_base") or query.get("callbackBase") or "").rstrip("/")
    if not bridge_url or not open_id or not js_code or not callback_base:
        raise RuntimeError("启动链接缺少 bridge/open_id/js_code/callback_base")

    # 先直接验证服务端 login 回调，再让浏览器走一遍厂商 H5；两层都通过才能区分
    # “桥接显示正常”与“实际会话无法登录”。login 不消耗 token，页面仍可继续使用。
    callback = http_json(
        f"{callback_base}/api/server/login",
        method="POST",
        body={"open_id": open_id, "js_code": js_code},
        headers=common_headers,
    )
    callback_data = require_envelope(callback, "ZGame 服务端登录回调")
    if not str(callback_data.get("session") or "").strip():
        raise RuntimeError("ZGame 服务端登录回调未返回 session")

    screenshot_path = pathlib.Path(args.screenshot).expanduser().resolve()
    # 先输出本次短期链接；即使 H5 最终失败，调用方仍能复现同一厂商会话。
    print(f"expires_at_ms={launch.get('expires_at_ms')}", flush=True)
    print(f"launch_url={launch_url}", flush=True)
    state, browser_errors = run_browser_check(
        chrome_path=args.chrome,
        launch_url=launch_url,
        game_id=args.game_id,
        session_id=session_id,
        bridge_source=fetch_text(bridge_url),
        screenshot_path=screenshot_path,
        timeout_seconds=args.timeout,
    )
    print("ZGame Flutter 容器模拟验证通过")
    print(f"account={args.account} game_id={args.game_id} session_id={session_id}")
    print(
        "browser_state="
        + json.dumps(
            {
                "bridge_ready": state.get("bridgeReady"),
                "visible_canvas_count": state.get("visibleCanvasCount"),
                "largest_canvas_pixels": state.get("largestCanvasPixels"),
                "title": state.get("title"),
            },
            ensure_ascii=False,
        )
    )
    print(f"browser_error_count={len(browser_errors)}")
    for error in browser_errors[:5]:
        print(f"browser_error={error}")
    print(f"screenshot={screenshot_path}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:  # noqa: BLE001 - CLI 统一输出可执行失败原因。
        print(f"验证失败: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc