chuanqi/cq2_sdk/scripts/business/c1_launch_game.py

"""C1: 调启动接口 — 拿 host/port/game_token, 创建 runs 记录, 更新 zone 的 host/port。

数据流:
    MySQL user (token from A1) + server (B1) + zone (B2/B3, zone_id=996box serverid)
        ↓
    Box996Login.enter_game_client(game_id, serverid)
        ↓
    响应: {data: {host, port, game_token, env}}
        ↓
    RunStore.create(user_id, server_db_id, zone_db_id, status="running")
        ↓
    ZoneStore 跟新 host/port  (下次 B3 能看到)
        ↓
    DebugLogger 留痕

跑法:
    uv run python scripts/business/c1_launch_game.py
    uv run python scripts/business/c1_launch_game.py --user longwu988
    uv run python scripts/business/c1_launch_game.py --zone 14414742
    uv run python scripts/business/c1_launch_game.py --check

前置依赖:
    1. A1 跑过 (user.token 在 MySQL)
    2. B1 跑过 (server 在 MySQL)
    3. B2 跑过 (zones 在 MySQL)
"""
from __future__ import annotations
import argparse
import asyncio
import os
import sys
from pathlib import Path
from typing import Any

_HERE = Path(__file__).resolve().parent
_PROJECT = _HERE.parent.parent
sys.path.insert(0, str(_PROJECT / "src"))

from cq2_sdk.env_loader import load_env  # noqa: E402
load_env(_PROJECT / ".env")

from cq2_sdk.debug_log import DebugLogger  # noqa: E402
from cq2_sdk.login import Box996Login, ManualGeetestSolver  # noqa: E402
from cq2_sdk.store import (  # noqa: E402
    RunStore,
    ServerStore,
    UserStore,
    ZoneStore,
    connection,
)

DEFAULT_USERNAME = "longwu988"
DEFAULT_SERVER_NAME = "玖扬复古176"
DEFAULT_DEVICE_ID = "00:15:5d:0a:04:00"


async def launch_and_persist(
    username: str,
    server_name: str,
    zone_id_996box: str | None,  # 显式传某个 zone, None 走 B3 自动选
    log: DebugLogger,
) -> dict[str, Any]:
    """Run C1: 启动游戏 + 落 runs 记录 + 跟新 zone host/port。

    Returns: {
        run_id, user_id, server_db_id, zone_db_id, zone_id_996box,
        host, port, game_token
    }
    """
    log.step("c1_launch_game_start",
             username=username, server_name=server_name,
             zone_id_996box=zone_id_996box)

    # 1. 从 MySQL 拿 user / server / zone
    with connection() as conn:
        user = UserStore(conn).get_by_username(username)
        if user is None or not user.token:
            raise RuntimeError(
                f"user {username!r} 在 MySQL 没 token。先跑 A1。"
            )
        server = ServerStore(conn).get_by_name(server_name)
        if server is None:
            raise RuntimeError(
                f"server {server_name!r} 不在 MySQL。先跑 B1。"
            )
        zones = ZoneStore(conn)
        if zone_id_996box is None:
            # 走 B3: 选最新 openable
            zone = zones.get_latest_openable(server.id)
            if zone is None:
                raise RuntimeError(
                    f"没可用的 openable zone (server={server_name})。先跑 B2 拉新区列表。"
                )
            zone_id_996box = zone.zone_id
            log.info("zone_auto_selected_via_b3",
                     zone_db_id=zone.id, zone_id_996box=zone_id_996box,
                     zone_name=zone.zone_name)
        else:
            # 用户显式指定了 zone_id (e.g. "14414742"), 按这个去查
            zone = zones.get_latest_openable(server.id)  # 用 latest_openable 拿 zone object
            # 简化: 不管 zone_id 怎么传, 都从 latest 拿 zone db_id
            # 实际应该是 get_by_zone_id(zone_id_996box), 但 ZoneStore 没这个方法, 临时用 latest
            # TODO: 加 ZoneStore.get_by_996box_id
            if zone is None or zone.zone_id != zone_id_996box:
                raise RuntimeError(
                    f"zone_id_996box={zone_id_996box!r} 不在 MySQL (或不是 openable)。"
                    f" 实际 latest openable: {zone.zone_id if zone else 'none'}"
                )
            log.info("zone_explicitly_specified",
                     zone_db_id=zone.id, zone_id_996box=zone_id_996box,
                     zone_name=zone.zone_name)
        zone_db_id = zone.id
        game_id = server.game_id
        server_db_id = server.id
        user_id = user.id
    log.info("loaded_from_mysql",
             user_id=user_id, server_db_id=server_db_id, zone_db_id=zone_db_id,
             game_id=game_id, zone_id_996box=zone_id_996box)

    # 2. 调 996box 启动接口
    log.step("c1_call_enter_game_client",
             game_id=game_id, serverid=int(zone_id_996box))
    box = Box996Login(geetest=ManualGeetestSolver(), device_id=DEFAULT_DEVICE_ID)
    box.token = user.token
    resp = await box.enter_game_client(game_id=game_id, serverid=int(zone_id_996box))
    log.info("enter_game_client_response", resp_keys=list(resp.keys()) if isinstance(resp, dict) else type(resp).__name__)

    # 3. 解析响应 — 真实结构 (2026-09-16 实测):
    #    data = { download_info, download_info_client, start_info, config_name }
    #    start_info = 写给 cq.exe 的 env.json 内容 (signkey/token/uid/...)
    #    host/port 不在响应里 — 游戏网关走 .env CQ2_GATE_HOST/PORT
    data = resp if isinstance(resp, dict) else {}
    start_info = data.get("start_info") or {}
    if not start_info:
        log.error("launch_response_missing_start_info", resp_keys=list(data.keys()))
        raise RuntimeError(
            f"启动接口返回没有 start_info: keys={list(data.keys())}"
        )
    game_token = str(start_info.get("token", ""))
    # host/port 不在响应里 — 游戏网关走 .env CQ2_GATE_HOST/PORT。
    # 默认值 2026-09-16 实测: gfgame.dhsf.xqhuyu.com 上只有 12000 开着
    # (8001 直连端口全关, 逐区后端端口藏在 12000 入口代理后面, 见
    # scripts/probe_gateway.py / port_matrix.py 的老逆向结论)。
    host = os.environ.get("CQ2_GATE_HOST", "gfgame.dhsf.xqhuyu.com")
    port = int(os.environ.get("CQ2_GATE_PORT", "12000"))
    log.info("launch_response_parsed",
             host=host, port=port,
             game_account=start_info.get("gameAccount"),
             uid=start_info.get("uid"),
             signkey_prefix=str(start_info.get("signkey", ""))[:12] + "...",
             game_token_prefix=(game_token[:12] + "...") if game_token else "")

    # 4. 创建 run 记录 + 存 start_info (env.json 原料) + 更新 zone host/port
    log.step("c1_persist_run_record")
    import json as _json
    env_dir = _PROJECT / "env_dumps"
    env_dir.mkdir(exist_ok=True)
    with connection() as conn:
        run = RunStore(conn).create(
            user_id=user_id,
            server_id=server_db_id,
            zone_id=zone_db_id,
            status="running",
        )
        log.info("run_created", run_id=run.id, status=run.status)

        # start_info 落盘为 env.json 样式的文件 (真客户端就是这么给 cq.exe 的)
        env_path = env_dir / f"run_{run.id}_env.json"
        env_path.write_text(
            _json.dumps(start_info, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE runs SET snapshot_path = %s WHERE id = %s",
                (str(env_path), run.id),
            )

        # 5. 更新 zone 的 host/port (C2 直接拿来 TCP poll)
        with conn.cursor() as cur:
            cur.execute(
                "UPDATE zones SET host = %s, port = %s WHERE id = %s",
                (host, int(port), zone_db_id),
            )
        log.info("zone_host_port_updated", host=host, port=port,
                 env_file=str(env_path))

    return {
        "run_id": run.id,
        "user_id": user_id,
        "server_db_id": server_db_id,
        "zone_db_id": zone_db_id,
        "zone_id_996box": zone_id_996box,
        "host": host,
        "port": int(port),
        "game_token_prefix": game_token[:16] + "..." if game_token else "",
        "start_info_keys": sorted(start_info.keys())[:12],
        "env_file": str(env_path),
    }


def check_config(username: str | None, server_name: str | None,
                 zone_id: str | None) -> bool:
    print("[check] 配置检查:")
    print(f"  username    = {username!r}  [OK]" if username
          else "  username    = (未设置)  [FAIL]")
    print(f"  server_name = {server_name!r}  [OK]" if server_name
          else "  server_name = (未设置)  [FAIL]")
    print(f"  zone_id     = {zone_id!r}  (可选, 不填走 B3 自动选)"
          if zone_id or True  # always show, just note it's optional
          else "")
    print("  (实际跑还需要 A1 + B1 + B2 都跑过, MySQL 有 user.token + server + zones)")
    return bool(username and server_name)


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="C1: 调 996box 启动接口, 拿 host/port, 创 runs 记录",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument(
        "--user",
        default=os.environ.get("CQ2_ACCOUNT", DEFAULT_USERNAME),
        help="用哪个 user 的 token (默认 .env CQ2_ACCOUNT)",
    )
    p.add_argument(
        "--server",
        default=os.environ.get("CQ2_DEFAULT_SERVER", DEFAULT_SERVER_NAME),
        help=f"区服名 (默认 {DEFAULT_SERVER_NAME})",
    )
    p.add_argument(
        "--zone",
        default=None,
        help="996box zone serverid (e.g. 14414742)。不填走 B3 自动选 latest openable。",
    )
    p.add_argument(
        "--check",
        action="store_true",
        help="只检查 .env / CLI 配置, 不实际调 API",
    )
    return p.parse_args()


def main() -> int:
    args = parse_args()

    if args.check:
        ok = check_config(args.user, args.server, args.zone)
        print(f"\n  result: {'READY' if ok else 'NOT READY'}")
        return 0 if ok else 1

    if not args.user or not args.server:
        check_config(args.user, args.server, args.zone)
        return 1

    with DebugLogger(
        extra={"task": "c1", "username": args.user,
               "server_name": args.server, "zone": args.zone}
    ) as log:
        log.info("c1_start", username=args.user,
                 server_name=args.server, zone=args.zone)
        try:
            result = asyncio.run(launch_and_persist(
                args.user, args.server, args.zone, log,
            ))
        except Exception as e:
            log.error(e, step="c1_unexpected")
            print(f"\n[FAIL] {type(e).__name__}: {e}")
            print(f"  log file: {log.path}")
            return 1

    print("\n[OK] C1 启动 + 落库成功:")
    for k, v in result.items():
        print(f"  {k:<16} = {v}")
    print(f"  log file        = {log.path}")
    print("  下一步: 跑 C2 等游戏真起来, 然后可以开始 D1-D13")
    return 0


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