chuanqi/cq2_sdk/scripts/test_refresh.py

"""Live-test: can the 30-day JWT refresh itself via getByRefresh?

From app.asar JS map: user.getByRefresh = /userApi/v1/UserInfo/getByRefresh
The client calls this to re-fetch user info (and possibly re-issue token)
without re-login. If token renewal works here, the slider is only needed
ONCE per ~month, and maybe never again via refresh chaining.
"""
import asyncio
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")

from cq2_sdk.login import Box996Login, ManualGeetestSolver
from cq2_sdk.login.box_996 import Box996Error
from cq2_sdk.store import UserStore, connection


async def main() -> int:
    with connection() as conn:
        u = UserStore(conn).get_by_username("longwu988")
    if not u or not u.token:
        print("FAIL: no token in MySQL")
        return 1
    print(f"token (from saz): {u.token[:30]}...")

    box = Box996Login(geetest=ManualGeetestSolver(), device_id="00:15:5d:0a:04:00")
    box.token = u.token
    # host/prefix rule (from captures): user-api host serves paths WITHOUT
    # the /userApi gateway prefix — get_phone_code worked at
    # user-api.../v1/UserInfo/get_phone_code, so strip the prefix here too.
    candidates = [
        ("https://user-api.996hezi.com", "/v1/UserInfo/getByRefresh"),
        ("https://api.996hezi.com", "/userApi/v1/UserInfo/getByRefresh"),
        ("https://api.996hezi.com", "/v1/UserInfo/getByRefresh"),
    ]
    r = None
    for host, path in candidates:
        try:
            r = await box._request("POST", host, path)
            print(f"[HIT] {host}{path}")
            break
        except Box996Error as e:
            print(f"[{e.code}] {host}{path} — {e.msg[:50]}")
            r = r  # keep last
        except Exception as e:
            msg = str(e)
            tag = "404" if "404" in msg else type(e).__name__
            print(f"[{tag}] {host}{path}")
    if r is None:
        return 1

    print(f"code = {r.get('code')}")
    print(f"msg  = {r.get('msg', '')}")
    d = r.get("data") or {}
    if isinstance(d, dict):
        info = d.get("userinfo") or d
        print(f"nick = {info.get('nick_name')}  id = {info.get('id')}")
        new_tok = d.get("token") or info.get("token")
        if new_tok and new_tok != u.token:
            print(f"*** NEW TOKEN ISSUED (len={len(new_tok)}) — refresh chain works! ***")
        else:
            print("(no new token in response — endpoint re-validates but may not renew)")
    return 0


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