chuanqi/cq2_sdk/scripts/smoke.py

"""End-to-end smoke test — REAL 996box protocol (HTTP part).

USAGE:
    # Password login + list today's servers + launch (all HTTP)
    uv run python scripts/smoke.py

    Requires .env with CQ2_ACCOUNT / CQ2_PASSWORD.
    Geetest captcha: server may demand behavior check (code 101000);
    ManualGeetestSolver will prompt on stdin when that happens.
"""
import argparse
import asyncio
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir, "src"))


async def main() -> int:
    parser = argparse.ArgumentParser(description="cq2_sdk HTTP smoke test")
    parser.add_argument("--game-id", type=int, default=206923)
    args = parser.parse_args()

    from cq2_sdk.env_loader import load_env
    load_env()

    account = os.environ.get("CQ2_ACCOUNT") or os.environ.get("CQ2_PHONE")
    password = os.environ.get("CQ2_PASSWORD")
    if not account or not password:
        print("error: CQ2_ACCOUNT / CQ2_PASSWORD required in .env", file=sys.stderr)
        return 1

    from cq2_sdk.client import Cq2Client

    print(f"[smoke] login as {account!r} (real protocol: api.jiyuhz.cn)")
    async with Cq2Client(account=account, password=password) as cli:
        assert cli.box_session is not None
        print(f"[smoke] login OK: uid={cli.box_session.uid}")

        servers = await cli.launcher.get_today_servers(args.game_id)
        print(f"[smoke] today's servers: {len(servers)}")
        for s in servers[:5]:
            print(f"    {s.server_name}  sid={s.serverid}  opens={s.section.strip()}")

        if servers:
            target = servers[-1]
            print(f"[smoke] EnterGameClient -> {target.server_name} (sid={target.serverid})")
            data = await cli.launcher.enter_game(target.serverid, game_id=args.game_id)
            print(f"[smoke] launch OK: {str(data)[:120]}")

    print("[smoke] HTTP chain complete. TCP part awaits Step 2 (Frida).")
    return 0


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