chuanqi/cq2_sdk/scripts/record_session.py

"""Record a full game session to JSONL for offline replay.

USAGE:
    # Password login (recommended)
    uv run python scripts/record_session.py --phone <account> --password <pwd>

    # SMS login
    uv run python scripts/record_session.py --phone <phone> --sms-code <code>

    # From .env
    uv run python scripts/record_session.py

Runs the full pre-game + 13-task flow and logs every send/recv to
tests/recorded/session_<timestamp>.jsonl. Use the recorded file with
`MockTcpClient` to write tests that exercise task decision logic
without a live game server.

NOTE: This script requires a live 996box/cq2 server and valid
account credentials. Use it after verifying the SDK against
production at least once with `smoke.py`.
"""
import argparse
import asyncio
import os
import sys
from datetime import datetime
from pathlib import Path

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


async def main() -> int:
    parser = argparse.ArgumentParser(description="cq2_sdk session recorder")
    parser.add_argument("--phone", help="Phone number or account name (CQ2_PHONE or CQ2_ACCOUNT)")
    parser.add_argument("--password", help="Account password (CQ2_PASSWORD)")
    parser.add_argument("--sms-code", help="SMS verification code (CQ2_SMS_CODE)")
    parser.add_argument("--lua-path", help="Path to decrypted client Lua (CQ2_LUA_PATH)")
    args = parser.parse_args()

    from cq2_sdk.env_loader import load_env
    load_env()

    phone = args.phone or os.environ.get("CQ2_PHONE") or os.environ.get("CQ2_ACCOUNT")
    password = args.password or os.environ.get("CQ2_PASSWORD")
    sms_code = args.sms_code or os.environ.get("CQ2_SMS_CODE")
    lua_path = args.lua_path or os.environ.get("CQ2_LUA_PATH")

    if not phone:
        print("error: --phone (or CQ2_PHONE/CQ2_ACCOUNT env) required", file=sys.stderr)
        return 1
    if not password and not sms_code:
        print("error: need --password OR --sms-code", file=sys.stderr)
        return 1

    from cq2_sdk.client import Cq2Client
    from cq2_sdk.testing import Recorder

    out_dir = Path("tests/recorded")
    out_dir.mkdir(parents=True, exist_ok=True)
    out = out_dir / f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jsonl"
    print(f"[record] writing to {out}")

    async with Cq2Client(
        phone=phone,
        password=password,
        sms_code=sms_code,
        lua_path=lua_path,
        http_secret=os.environ.get("CQ2_SIGNKEY", ""),
        sign_version=2,
    ) as cli:
        # run_until_enter_game() first — it sets up TCP + task context
        await cli.launcher.run_until_enter_game(server_name="玖扬复古176", zone_index=0)
        # Then create the recorder (now cli.task.ctx.tcp exists)
        assert cli.task is not None
        recorder = Recorder(cli.task.ctx.tcp, out)
        recorder.install()
        try:
            result = await cli.run_newbie_flow()
        finally:
            recorder.close()
    print(f"[record] done: {len(result.completed_tasks)} tasks in {result.elapsed:.1f}s")
    return 0


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