chuanqi/cq2_sdk/scripts/cdp_full_chain.py

"""Extract the _captcha_ token from 996box localStorage via CDP, then use our
REAL SDK (Box996Login) to complete the full chain:
  login_with_password(captcha=reused_token)
  -> getTodayOpen
  -> EnterGameClient(serverid)
  -> 996box spawns cq.exe -> cq2.exe -> real TCP login traffic
  -> (capture running in background records everything)
"""
import asyncio
import base64
import json
import sys
from pathlib import Path

import httpx
import websockets

sys.path.insert(0, "src")
from cq2_sdk.env_loader import load_env

load_env()


async def get_targets():
    async with httpx.AsyncClient(timeout=5) as c:
        r = await c.get("http://127.0.0.1:9222/json/list")
        return r.json()


async def cdp_eval(ws, expr: str, msg_id: int) -> dict:
    await ws.send(json.dumps({"id": msg_id, "method": "Runtime.evaluate",
                              "params": {"expression": expr, "returnByValue": True}}))
    while True:
        msg = json.loads(await ws.recv())
        if msg.get("id") == msg_id:
            return msg.get("result", {})


async def get_captcha_from_localstorage() -> dict | None:
    targets = await get_targets()
    login = next((t for t in targets if "hzwlan" in t.get("url", "") or "996box" in t.get("url", "")), None)
    if not login:
        print("no hzwlan/996box target")
        return None
    async with websockets.connect(login["webSocketDebuggerUrl"], max_size=10**7) as ws:
        res = await cdp_eval(ws, "localStorage.getItem('_captcha_')", 1)
        val = res.get("result", {}).get("value")
        if not val:
            print("_captcha_ empty")
            return None
        # decode base64 -> JSON
        try:
            padded = val + "=" * (-len(val) % 4)
            decoded = base64.urlsafe_b64decode(padded).decode("utf-8")
            cap = json.loads(decoded)
            print(f"captcha fields: {list(cap.keys())}")
            print(f"  captchaId: {cap.get('captchaId','?')[:20]}...")
            print(f"  genTime:   {cap.get('genTime','?')}")
            return cap
        except Exception as e:
            print(f"decode err: {e}")
            return None


async def main() -> int:
    cap = await get_captcha_from_localstorage()
    if not cap:
        return 1

    # Use our REAL SDK
    from cq2_sdk.login.box_996 import Box996Login

    async def run():
        box = Box996Login()
        try:
            print()
            print("=== SDK login_with_password (reused captcha token) ===")
            try:
                session = await box.login_with_password("longwu988", "Longwu1988", captcha=cap)
                print(f"*** LOGIN OK: uid={session.uid} token={session.token[:25]}... ***")
            except Exception as e:
                print(f"login failed: {type(e).__name__}: {e}")
                print("(captcha likely single-use/expired — expected fallback)")

            print()
            print("=== getTodayOpen (uses stored JWT if login above worked) ===")
            try:
                servers = await box.get_today_open(206923)
                print(f"today's servers: {len(servers)}")
                for s in servers[:5]:
                    print(f"  {s.server_name}  sid={s.serverid}  opens={s.section.strip()}")
                if servers:
                    print()
                    print("=== EnterGameClient on the LATEST server ===")
                    target = servers[-1]
                    data = await box.enter_game_client(206923, target.serverid)
                    print(f"launch OK -> {str(data)[:100]}")
                    print()
                    print("*** 996box Electron should now spawn cq.exe -> cq2.exe ***")
                    print("*** background capture (if running) records the TCP login ***")
                    return 0
            except Exception as e:
                print(f"game API failed: {type(e).__name__}: {e}")
        finally:
            await box.close()
        return 1

    return await run()


sys.exit(asyncio.run(main()))