chuanqi/cq2_sdk/scripts/test_full_auto_login.py

"""End-to-end automated 996box password login via jfbym + wulu-geetest-bypass.

Full chain (cost per run: ~0.008 yuan for one jfbym type-22222 call):
  1. wulu Geetest(risk_type='slide').load()  → captcha params (bg/slice image paths, lot_number, process_token, pow_detail)
  2. download bg image
  3. jfbym.solve_distance(bg)                  → gap distance in px
  4. wulu Geetest generates w (RSA+AES), builds verify query
  5. POST /verify  → seccode {captcha_id, lot_number, pass_token, gen_time, captcha_output}
  6. Box996Login.login_with_password(un, pw, captcha={**seccode, event:1, type:"login"})
  7. Save new token to MySQL

Configuration via .env + CLI overrides.

We use the real 996box production captcha_id (captured from saz):
  26e753da4a8206359763bcb9bd327eb4
If it has rotated, the /load will return a 1024 (captchaId expired) and we
can fall back to the official geetest demo captcha.
"""
from __future__ import annotations

import argparse
import asyncio
import base64
import json
import os
import sys
from pathlib import Path

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

from cq2_sdk.env_loader import load_env
load_env(_HERE.parent / ".env")

# 996box production captcha_id (from 手机号短信验证登录过程.saz)
CAPTCHA_ID_996BOX = "26e753da4a8206359763bcb9bd327eb4"


def b64_json(d: dict) -> str:
    """Match what 996box client sends in the _captcha_ field: base64(JSON)."""
    return base64.b64encode(json.dumps(d, separators=(",", ":")).encode()).decode()


def run_e2e(
    username: str,
    password: str,
    captcha_id: str,
    jfbym_token: str,
    use_jfbym: bool = False,
) -> int:
    import httpx

    from wulu_geetest_bypass import Geetest
    from wreq import Emulation
    from cq2_sdk.login.box_996 import Box996Login, Box996Error
    from cq2_sdk.login.geetest import ManualGeetestSolver  # not used for 2nd call; satisfies Box996Login ctor
    from cq2_sdk.store import UserStore, connection
    from cq2_sdk.debug_log import DebugLogger

    if use_jfbym:
        from cq2_sdk.login.geetest_jfbym import make_geetest_jfbym_solver
        from cq2_sdk.login.jfbym import JfbymSliderSolver
        solver = JfbymSliderSolver(token=jfbym_token)
        geetest_solver = make_geetest_jfbym_solver(solver)
        Geetest.register_solver("slide", geetest_solver)
    else:
        # Use wulu's BUILT-IN OpenCV slide solver (no jfbym override).
        # Reason: jfbym gap_only override (191 → setLeft=191) produced w that
        # geetest server rejected with result: fail. The gap position from
        # jfbym != the setLeft value wulu's trajectory math expects. Avoid
        # this calibration hazard by delegating to wulu's tested opencv
        # path. Cloud jfbym is kept as an OPT-IN (--use-jfbym) for envs
        # where opencv is unavailable.
        from wulu_geetest_bypass.solver import solve_slide as _open_cv_slide
        Geetest.register_solver("slide", _open_cv_slide)

    # 2) Set up Geetest with 996box-like referer
    log = DebugLogger(extra={"task": "e2e_login", "user": username})
    log.info("e2e_start", captcha_id=captcha_id)

    headers_996box = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 "
                      "(KHTML, like Gecko) 996Box/5.1.1.9007 Chrome/108.0.5359.215 Electron/",
        "Referer": "https://pc.hzwlan.com/",
        "Origin": "https://pc.hzwlan.com",
        "X-Requested-With": "XMLHttpRequest",
    }

    g = Geetest(
        captcha_id=captcha_id,
        risk_type="slide",
        client_type="web",
        client_options={
            "headers": headers_996box,
            "emulation": Emulation.Chrome147,
            "verify": False,  # wulu uses wreq with tls emulation; .996hezi.com chain ok
        },
    )

    async def solve_then_login() -> dict:
        # 3) load + verify (one jfbym call inside)
        log.info("geetest_resolve_start")
        seccode = await g.resolve(retry=2)
        log.info("geetest_resolve_ok", seccode_keys=list(seccode.keys()))

        # 4) Inject into 996box password login
        captcha_payload = {
            "loginInfo": "",  # unused for password, but 996box client includes it
            "channel_id": "",
            "game_id": "",
            "inviter_uid": "",
            "needConfirmLogin": True,
            "phone": "",
            "code": "",
            "_captcha_": b64_json({
                "captchaId": seccode["captcha_id"],
                "captchaOutput": seccode["captcha_output"],
                "genTime": seccode["gen_time"],
                "lotNumber": seccode["lot_number"],
                "passToken": seccode["pass_token"],
            }),
        }
        log.info("calling_box996_login", has_captcha=bool(captcha_payload["_captcha_"]),
                 captcha_payload_keys=list(captcha_payload.keys()))

        box = Box996Login(
            geetest=ManualGeetestSolver(),  # not used on 2nd call; harmless
            device_id="00:15:5d:0a:04:00",
        )
        try:
            session = await box.login_with_password(
                username, password, captcha=captcha_payload,
            )
        except Box996Error as e:
            # Dump raw response so we can see what 996box actually rejected
            log.error("box996_login_failed", code=e.code, box_msg=e.msg[:120],
                     full_resp=str(e).__class__.__name__)
            return {"ok": False, "stage": "login", "error": f"{e.code}: {e.msg}"}
        except CaptchaRequiredError as e:
            log.error("box996_captcha_rejected", code=e.code, box_msg=e.msg[:120])
            return {"ok": False, "stage": "login", "error": f"captcha: {e.code}: {e.msg}"}

        log.info("box996_login_ok", uid=session.uid, token_len=len(session.token))
        return {"ok": True, "session": session, "seccode": seccode}

    result = asyncio.run(solve_then_login())
    if not result["ok"]:
        print(f"\n[FAIL] stage={result['stage']}: {result['error']}")
        return 1

    session = result["session"]

    # 5) Persist new token to MySQL
    with connection() as conn:
        users = UserStore(conn)
        existing = users.get_by_username(username)
        if existing is None:
            print(f"[FAIL] {username} not in MySQL; run scaffold first")
            return 1
        updated = users.update_token(user_id=existing.id, token=session.token)
        log.info("token_saved", user_id=updated.id, token_prefix=session.token[:24] + "...")

    print(f"\n[OK] full auto-login SUCCESS")
    print(f"  uid        = {session.uid}")
    print(f"  new token  = {session.token[:30]}...")
    print(f"  saved to   = users.id={updated.id} (longwu988)")
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--user", default=os.environ.get("CQ2_ACCOUNT"))
    ap.add_argument("--password", default=os.environ.get("CQ2_PASSWORD"))
    ap.add_argument("--captcha-id", default=CAPTCHA_ID_996BOX)
    ap.add_argument("--jfbym-token", default=os.environ.get("CQ2_JFBYM_TOKEN"))
    ap.add_argument("--use-jfbym", action="store_true",
                    help="Override wulu's opencv with jfbym cloud recognition (may need calibration)")
    args = ap.parse_args()

    if not all([args.user, args.password, args.jfbym_token]):
        print("ERROR: missing --user / --password / --jfbym-token (or .env CQ2_*)")
        return 1
    return run_e2e(
        args.user, args.password, args.captcha_id, args.jfbym_token, args.use_jfbym,
    )


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