chuanqi/launch_cq2_mac.py

# -*- coding: utf-8 -*-
"""
launch_cq2_mac.py —— 996传奇盒子 Mac 端自动登录 + 启动 cq2 游戏客户端
=====================================================================
复刻 Mac 盒子(com.mac.BoxMacOS.xq)的原生启动链路（逆向实证，见 docs）：

  1. 账号密码登录  api.996hezi.com /auth/v3/login/safe/byPwd
     (GeekedTest 解极验v4滑块 + RSA(loginInfo) + BOXAPP-SHA256 签名)
  2. 启动区服      pc-api.996hezi.com /v1/homeapi/gameindex/EnterGameClient
     body: {box_game_id, serverid, device_info, appid}
     resp: data.start_info ← 正是盒子写入 App Group 的那份凭证
  3. 写共享凭证    defaults write group.com.XianQu.chuanqibox SHBoxMacOS <json>
     (App Group 域，经 cfprefsd 守护进程代理写入，无需沙盒权限)
  4. 拉起游戏      /usr/bin open 盒子沙盒内已下载的游戏 app
     ~/Library/Containers/com.mac.BoxMacOS.xq/Data/Documents/game/996传奇盒子.app

用法:
    .venv-mac/bin/python launch_cq2_mac.py -u <账号> -p <密码> \
        [--zone 14448581] [--game 206923]

依赖: requests pycryptodome pymysql curl_cffi opencv-python-headless numpy (GeekedTest)
"""
import argparse
import json
import os
import random
import string
import subprocess
import sys
import time

import requests as plain_requests
import urllib3
urllib3.disable_warnings()

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "已经开发的部分代码", "chuanqi-main", "code", "GeekedTest"))

from login_996pc import (  # noqa: E402
    do_login, box_auth, box_headers, APPV, UA, DEVICE_ID,
)

# ----------------------------------------------------------------------------
# 常量（逆向实证）
# ----------------------------------------------------------------------------
PC_API = "https://pc-api.996hezi.com"
ENTER_GAME_PATH = "/v1/homeapi/gameindex/EnterGameClient"
APPID = "GAjWNTqNpOVx"                     # 996box PC/Mac 盒子 sdkAppid

GAME_APP_PATH = os.path.expanduser(
    "~/Library/Containers/com.mac.BoxMacOS.xq/Data/Documents/game/996传奇盒子.app")

APP_GROUP = "group.com.XianQu.chuanqibox"  # 盒子+游戏 共享 App Group
PLIST_KEY = "SHBoxMacOS"                   # 游戏客户端启动时读取的键

DEFAULT_GAME_ID = 206923                   # 玖扬复古176
DEFAULT_ZONE_ID = 14448581                 # 霸者36区


def enter_game(token: str, game_id: int, zone_id: int) -> dict:
    """调 EnterGameClient —— 等价于在盒子里点击【启动】按钮。

    返回 data.start_info（盒子将要传给游戏客户端的全部凭证）。
    """
    device_info = json.dumps({
        "type": "pc", "model": "pc", "os": "Win10",
        "width": 1920, "height": 1032, "lang": "zh-CN",
        "id": "43b2d8df251be893bf57a67560b1e74e",
        "net_type": "unknown", "mac": DEVICE_ID, "ua": UA,
    }, separators=(",", ":"))
    body = json.dumps({
        "box_game_id": game_id,
        "serverid": zone_id,
        "device_info": device_info,
        "appid": APPID,
    }, separators=(",", ":"))

    h = box_headers(token)
    h["X-BOX-AUTHORIZATION"] = box_auth("POST", ENTER_GAME_PATH, body, token)
    r = plain_requests.post(PC_API + ENTER_GAME_PATH, data=body.encode(),
                            headers=h, timeout=25, verify=False)
    j = r.json()
    code = j.get("code")
    if code == 81111:
        raise SystemExit(
            f"[!] code=81111 请实名认证 —— 该账号未实名(auth_status=0)，"
            f"服务端拒绝启动游戏。请在盒子/官方页完成实名后重试。")
    if code != 200:
        raise SystemExit(f"[!] EnterGameClient 失败 code={code} msg={j.get('msg')}")
    data = j.get("data") or {}
    start = data.get("start_info")
    if not start or not start.get("token"):
        raise SystemExit(f"[!] 响应缺少 start_info: {json.dumps(data)[:300]}")
    return start


GROUP_PLIST = os.path.expanduser(
    "~/Library/Group Containers/group.com.XianQu.chuanqibox/"
    "Library/Preferences/group.com.XianQu.chuanqibox.plist")


def write_app_group(start_info: dict, game_id: int, zone_id: int) -> None:
    env = dict(start_info)
    # 与 Mac 盒子写入 SHBoxMacOS 的字段实样对齐。
    # start_info 缺的这些字段由盒子端补：hideServerList/hideBackBox 控制游戏内
    # 是否弹出选区界面（缺失会停在区服列表等人点击）；display 尺寸/osVersion/
    # model 供客户端适配窗口；is_adult/is_realname 为防沉迷标记。
    env.setdefault("platform", "mac")
    env.setdefault("gameId", game_id)
    env.setdefault("serverId", str(zone_id))
    env.setdefault("boxConfig", 1)
    env.setdefault("isBoxLogin", 1)
    env.setdefault("gm", "0")
    env.setdefault("scene", 0)
    env.setdefault("sdkChannel", "996box_mac")
    env.setdefault("macAddress", _box_uuid())
    env.setdefault("hideServerList", 1)
    env.setdefault("hideBackBox", 1)
    env.setdefault("displayWidth", "1230")
    env.setdefault("displayHeight", "760")
    env.setdefault("osVersion", _macos_version())
    env.setdefault("app_ver", "1.0.1")
    env.setdefault("model", _mac_model())
    env.setdefault("is_adult", 1)
    env.setdefault("is_realname", 1)
    payload = json.dumps(env, ensure_ascii=False, separators=(",", ":"))

    # 无沙盒 entitlement 的进程经 cfprefsd 写 group.* 域会落到 ~/Library/Preferences，
    # 而游戏进程（带 App Group entitlement）读的是 Group Containers 那份，
    # 因此直接 plutil 改写目标 plist 文件。
    # 先让 cfprefsd 把它缓存的内存态刷盘，避免我们写入后被回写覆盖。
    subprocess.run(["defaults", "read", APP_GROUP, PLIST_KEY], capture_output=True)
    os.makedirs(os.path.dirname(GROUP_PLIST), exist_ok=True)
    r = subprocess.run(
        ["plutil", "-replace", PLIST_KEY, "-string", payload, GROUP_PLIST],
        capture_output=True, text=True)
    if r.returncode != 0:
        r = subprocess.run(
            ["plutil", "-insert", PLIST_KEY, "-string", payload, GROUP_PLIST],
            capture_output=True, text=True)
    if r.returncode != 0:
        raise SystemExit(f"[!] 写 Group plist 失败: {r.stderr}")
    print(f"[*] 已写入 App Group 凭证 → {GROUP_PLIST}")
    print(f"    uid={env.get('uid')} token={str(env.get('token'))[:20]}...")


def _box_uuid() -> str:
    r = subprocess.run(["defaults", "read", "com.mac.BoxMacOS.xq", "uuidForKeychain"],
                       capture_output=True, text=True)
    if r.returncode == 0 and r.stdout.strip():
        return r.stdout.strip().strip('"')
    return ""


def _macos_version() -> str:
    r = subprocess.run(["sw_vers", "-productVersion"], capture_output=True, text=True)
    return r.stdout.strip() or "14.7.1"


def _mac_model() -> str:
    r = subprocess.run(["sysctl", "-n", "hw.model"], capture_output=True, text=True)
    return r.stdout.strip() or "Mac16,10"


def launch_game_app() -> None:
    """用 /usr/bin open 拉起盒子沙盒内的游戏 app（与盒子 spawn 行为一致）。"""
    if not os.path.isdir(GAME_APP_PATH):
        raise SystemExit(
            f"[!] 游戏客户端未下载: {GAME_APP_PATH}\n"
            f"    请先用 996传奇盒子(Mac) 正常启动一次该游戏完成下载，再跑本脚本。")
    r = subprocess.run(["/usr/bin/open", "-a", GAME_APP_PATH],
                       capture_output=True, text=True)
    if r.returncode != 0:
        raise SystemExit(f"[!] open 失败: {r.stderr}")
    print(f"[*] 已拉起游戏客户端: {GAME_APP_PATH}")


def _game_pid():
    r = subprocess.run(["pgrep", "-f", "AppTranslocation.*996传奇盒子.app/996传奇盒子"],
                       capture_output=True, text=True, errors="ignore")
    if r.returncode == 0 and r.stdout.strip():
        return int(r.stdout.split()[0])
    return None


def _game_window():
    """返回 (window_id, bounds) 或 None。"""
    try:
        import Quartz
    except ImportError:
        return None
    pid = _game_pid()
    if pid is None:
        return None
    for w in Quartz.CGWindowListCopyWindowInfo(Quartz.kCGWindowListOptionAll,
                                               Quartz.kCGNullWindowID):
        if w.get('kCGWindowOwnerPID') == pid:
            b = dict(w.get('kCGWindowBounds') or {})
            if b.get('Width', 0) > 1000 and b.get('Height', 0) > 600:
                return w.get('kCGWindowNumber'), b
    return None


def cg_click(x: float, y: float) -> None:
    """CGEvent 合成点击。需要终端持有「输入监控」权限，否则事件被丢弃。"""
    import Quartz, time
    for evt in (Quartz.kCGEventLeftMouseDown, Quartz.kCGEventLeftMouseUp):
        e = Quartz.CGEventCreateMouseEvent(None, evt, (x, y), Quartz.kCGMouseButtonLeft)
        Quartz.CGEventPost(Quartz.kCGHIDEventTap, e)
        time.sleep(0.08)


def capture_window(path: str) -> bool:
    try:
        import Quartz
        from Foundation import NSMutableData
    except ImportError:
        return False
    win = _game_window()
    if not win:
        return False
    wid, b = win
    rect = Quartz.CGRectMake(b['X'], b['Y'], b['Width'], b['Height'])
    img = Quartz.CGWindowListCreateImage(
        rect,
        Quartz.kCGWindowListOptionOnScreenOnly | Quartz.kCGWindowListExcludeDesktopElements,
        wid, Quartz.kCGWindowImageBoundsIgnoreFraming)
    if not img:
        return False
    ms = NSMutableData.data()
    ti = Quartz.CGImageDestinationCreateWithData(ms, "public.png", 1, None)
    Quartz.CGImageDestinationAddImage(ti, img, None)
    Quartz.CGImageDestinationFinalize(ti)
    with open(path, "wb") as f:
        f.write(bytes(ms))
    return True


def dismiss_popups(deadline_s: float = 90.0) -> bool:
    """关闭游戏启动后的公告弹窗并点击启动面板的【启动】按钮。

    按钮坐标为窗口相对比例（2026-09-18 多次截图实测标定）:
      - 公告弹窗【确定】: 57% 宽 / 70% 高
      - 启动面板【启动】: 52% 宽 / 73% 高（区服/线路已由凭证锁定）
    弹窗可能多条：循环点击确定直至画面稳定，再点启动。
    """
    import time as _t

    def rel_click(b, fx, fy, label):
        x = b['X'] + b['Width'] * fx
        y = b['Y'] + b['Height'] * fy
        cg_click(x, y)
        print(f"[*] 点击{label} @({x:.0f},{y:.0f})")

    def stable_after_click(b, label, wait=2.5):
        """点击后返回画面是否已稳定（无新弹窗弹出）。"""
        import hashlib, subprocess as sp
        def snap():
            if not capture_window("/tmp/_cq2_probe.png"):
                return ""
            try:
                return hashlib.md5(open("/tmp/_cq2_probe.png", "rb").read()).hexdigest()
            except OSError:
                return ""
        before = snap()
        rel_click(b, 0.57, 0.70, label)
        _t.sleep(wait)
        for _ in range(6):
            after = snap()
            if after and after == before:
                return True  # 画面不再变化 = 弹窗序列走完
            before = after
            rel_click(b, 0.57, 0.70, label)
            _t.sleep(wait)
        return False

    print("[*] 自动处理公告弹窗 ...")
    end = _t.time() + deadline_s
    while _t.time() < end:
        win = _game_window()
        if not win:
            _t.sleep(2)
            continue
        _, b = win
        stable_after_click(b, "公告【确定】")
        # 弹窗序列结束后出现启动面板，点【启动】
        rel_click(b, 0.52, 0.73, "【启动】")
        _t.sleep(6)
        # 启动后进入角色选择界面：选1号角色槽并点【开始】（2026-09-18 实测标定）
        win2 = _game_window()
        if win2:
            _, b = win2
            rel_click(b, 0.25, 0.45, "角色槽1")
            _t.sleep(1.5)
            rel_click(b, 0.50, 0.78, "【开始】进游戏")
        return True
    return False


def wait_game_connected(timeout: float = 180.0) -> bool:
    """等游戏进程出现并连上游戏网关(TCP)。"""
    print("[*] 等待游戏进程 + 网关连接 ...")
    deadline = time.time() + timeout
    announced = False
    while time.time() < deadline:
        # 游戏经 App Translocation 启动，进程路径含 AppTranslocation，bundle id 不出现在 cmdline
        r = subprocess.run(["pgrep", "-f", "AppTranslocation.*996传奇盒子.app/996传奇盒子"],
                           capture_output=True, text=True, errors="ignore")
        if r.returncode == 0:
            pid = r.stdout.split()[0]
            if not announced:
                print(f"[*] 游戏进程已启动 pid={pid}")
                announced = True
            net = subprocess.run(["lsof", "-i", "-n", "-P", "-a", "-p", pid],
                                 capture_output=True, text=True, errors="ignore")
            est = [l for l in net.stdout.splitlines()
                   if "ESTABLISHED" in l and ("8083" in l or "12000" in l)]
            if est:
                print(f"[✓] 游戏(pid={pid}) 已连接网关: {est[0].split()[-1]}")
                return True
        time.sleep(3)
    print("[!] 超时：未见游戏网关连接（可能仍在加载资源/校验凭证）")
    return False


def main():
    ap = argparse.ArgumentParser(description="996盒子 Mac 端自动启动 cq2")
    ap.add_argument("-u", "--user", required=True, help="996盒子账号")
    ap.add_argument("-p", "--password", required=True, help="密码")
    ap.add_argument("--game", type=int, default=DEFAULT_GAME_ID,
                    help=f"盒子 gameId (默认 {DEFAULT_GAME_ID} 玖扬复古176)")
    ap.add_argument("--zone", type=int, default=DEFAULT_ZONE_ID,
                    help=f"区服 serverid (默认 {DEFAULT_ZONE_ID} 霸者36区)")
    ap.add_argument("--token-file", help="跳过登录，直接用文件里的 token")
    args = ap.parse_args()

    # ---- 1. 登录 ----
    if args.token_file and os.path.exists(args.token_file):
        token = open(args.token_file).read().strip()
        print(f"[*] 使用已有 token: {token[:40]}...")
    else:
        s = plain_requests.Session()
        print(f"[*] 登录账号: {args.user}")
        j, key = do_login(s, args.user, args.password)
        if j.get("code") != 200 or not (j.get("data") or {}).get("token"):
            raise SystemExit(f"[!] 登录失败: code={j.get('code')} msg={j.get('msg')}")
        token = j["data"]["token"]
        info = (j["data"].get("userInfo") or {})
        print(f"[✓] 登录成功 uid={info.get('id')} 昵称={info.get('nick_name')}")

    # ---- 2. EnterGame（点启动）----
    print(f"[*] 启动游戏 gameId={args.game} zone={args.zone} ...")
    start_info = enter_game(token, args.game, args.zone)
    print(f"[✓] start_info 下发: channel={start_info.get('channel')} "
          f"uid={start_info.get('uid')} gameAccount={start_info.get('gameAccount')}")

    # ---- 3. 写 App Group 凭证 ----
    write_app_group(start_info, args.game, args.zone)

    # ---- 4. 拉起游戏 app ----
    launch_game_app()

    # ---- 5. 自动关闭公告弹窗（需「输入监控」权限）----
    time.sleep(8)  # 等游戏首帧渲染
    dismiss_popups()

    # ---- 6. 等待网关连接 ----
    ok = wait_game_connected()
    sys.exit(0 if ok else 2)


if __name__ == "__main__":
    main()