spider_work/deca_spider/get_token.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/08/12
"""得卡 DECA App token 抓取脚本（root 直读本地存储，替代 mitmproxy 代理抓包）。

设备已 root（Magisk）且 得卡 App 已登录时，直接从 App 私有目录的
Jetpack DataStore 文件读出当前登录态的 accessToken / refreshToken，
算好过期时间，覆盖写入项目根目录 token.json，供 deca_sold_core.py 等复用。

与千岛版差异（都是 root 直读，只是存储格式不同）：
    千岛 token 明文存 SharedPreferences xml，一条 grep 即可；
    得卡 token 存 DataStore 的 protobuf 文件 deka_settings.preferences_pb 里，
    藏在顶层 key `switch_accounts` 的 JSON 数组内（每个账号一条 userInfo），
    需按 protobuf 解出该 string，再解 JSON 取 accessToken 等字段。

App 登录 / 切号后该 pb 会实时更新，App 后台也会用 refreshToken 自动续期并回写，
所以本脚本读到的必然是「此刻有效」的 token。token.json 只需种一次，之后
deca_sold_core.ensure_token 会靠 refreshToken 自续期，refreshToken 失效时再跑本脚本。

用法：
    python get_token.py            # 读设备 → 覆盖写 token.json → 打印脱敏摘要
"""
import os
import sys
import json
import time
import base64
import subprocess

# 得卡 App 包名（pm list packages 实测：有好卡 · 得卡）
PACKAGE_NAME = "com.youhaoka.deka"

# 登录态所在的 DataStore（Preferences，protobuf 二进制）文件
PB_PATH = f"/data/data/{PACKAGE_NAME}/files/datastore/deka_settings.preferences_pb"

# token 藏身的顶层 DataStore key，value 是一段 JSON 数组（多账号切换列表）
ACCOUNTS_KEY = "switch_accounts"

# token.json 与本脚本同目录（deca_sold_core 从运行目录读它）
TOKEN_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "token.json")


def _read_varint(buf: bytes, i: int) -> tuple[int, int]:
    """从字节流指定位置解一个 protobuf varint。

    Args:
        buf (bytes): 原始字节流。
        i (int): 起始下标。

    Returns:
        tuple[int, int]: (解出的整数值, 解析后的新下标)。
    """
    shift = 0
    result = 0
    while True:
        b = buf[i]
        i += 1
        result |= (b & 0x7F) << shift
        if not b & 0x80:                       # 最高位为 0 表示 varint 结束
            return result, i
        shift += 7


def _get_pref_string(raw: bytes, key: str) -> str | None:
    """从 DataStore Preferences protobuf 里取指定 key 的 string value。

    DataStore 结构：顶层 map<string, Value>（field 1，每条是一个 entry message），
    entry 内 field 1=key(string)、field 2=Value message；Value 里 field 5=string 值。

    Args:
        raw (bytes): deka_settings.preferences_pb 的完整字节。
        key (str): 要取的顶层 key 名，如 "switch_accounts"。

    Returns:
        str | None: 该 key 的字符串值；未找到返回 None。
    """
    i = 0
    while i < len(raw):
        tag, i = _read_varint(raw, i)
        field, wire = tag >> 3, tag & 7
        if field == 1 and wire == 2:           # 顶层 map entry（length-delimited）
            ln, i = _read_varint(raw, i)
            entry = raw[i:i + ln]
            i += ln
            j = 0
            k = v = None
            while j < len(entry):
                t2, j = _read_varint(entry, j)
                f2, w2 = t2 >> 3, t2 & 7
                if f2 == 1 and w2 == 2:         # entry.key
                    kl, j = _read_varint(entry, j)
                    k = entry[j:j + kl].decode("utf-8", "replace")
                    j += kl
                elif f2 == 2 and w2 == 2:       # entry.value（Value message）
                    vl, j = _read_varint(entry, j)
                    vb = entry[j:j + vl]
                    j += vl
                    m = 0
                    while m < len(vb):          # 在 Value 里找 field 5=string
                        t3, m = _read_varint(vb, m)
                        f3, w3 = t3 >> 3, t3 & 7
                        if f3 == 5 and w3 == 2:
                            sl, m = _read_varint(vb, m)
                            v = vb[m:m + sl].decode("utf-8", "replace")
                            m += sl
                        elif w3 == 0:           # 跳过其它类型字段（bool/int 等）
                            _, m = _read_varint(vb, m)
                        elif w3 == 2:
                            xl, m = _read_varint(vb, m)
                            m += xl
                        else:
                            break
                else:
                    break
            if k == key:
                return v
        elif wire == 0:                         # 跳过非 map 的 varint 字段
            _, i = _read_varint(raw, i)
        elif wire == 2:                         # 跳过非 map 的 length-delimited 字段
            ln, i = _read_varint(raw, i)
            i += ln
        else:
            break
    return None


def _decode_jwt_exp(access: str) -> int | None:
    """从 JWT accessToken 的 payload 里解出 exp（过期 unix 秒）。

    Args:
        access (str): JWT 字符串（header.payload.signature）。

    Returns:
        int | None: exp 时间戳；解析失败返回 None。
    """
    try:
        payload = access.split(".")[1]
        payload += "=" * (-len(payload) % 4)                       # 补齐 base64url 填充
        data = json.loads(base64.urlsafe_b64decode(payload))
        return int(data["exp"]) if "exp" in data else None
    except Exception:
        return None


def read_device_token() -> dict:
    """从已登录的 得卡 App 读取当前账号的 token 三元组。

    通过 `adb exec-out su -c cat` 以 root 读 DataStore pb 文件（exec-out 保留二进制、
    不做 CRLF 转换），解出 switch_accounts；多账号时取 tokenObtainedAtMs 最大者
    （最近获取 = 当前登录账号）；exp 优先用 JWT 解，兜底 tokenObtainedAtMs+expiresIn。

    Returns:
        dict: {"access": str, "refresh": str, "exp": int, "userId": ..., "phone": str}。

    Raises:
        RuntimeError: adb 缺失 / 超时 / 未读到 pb / 未登录（无账号）时抛出。
    """
    cmd = ["adb", "exec-out", "su", "-c", f"cat {PB_PATH}"]
    try:
        result = subprocess.run(cmd, capture_output=True, timeout=20)
    except FileNotFoundError as e:
        raise RuntimeError(f"[{sys._getframe().f_code.co_name}] 未找到 adb，请确认已配置环境变量") from e
    except subprocess.TimeoutExpired as e:
        raise RuntimeError(f"[{sys._getframe().f_code.co_name}] adb 命令超时，请检查设备连接") from e

    raw = result.stdout or b""
    if not raw:
        err = (result.stderr or b"").decode("utf-8", "replace").strip() or "未知错误"
        raise RuntimeError(
            f"[{sys._getframe().f_code.co_name}] 读取 pb 失败：{err}\n"
            f"排查：1) adb devices 是否有设备 2) 设备是否已 root 3) 得卡 App 是否已登录"
        )

    accounts_json = _get_pref_string(raw, ACCOUNTS_KEY)
    if not accounts_json:
        raise RuntimeError(
            f"[{sys._getframe().f_code.co_name}] pb 里没有 {ACCOUNTS_KEY}，请确认 得卡 App 已登录"
        )
    try:
        accounts = json.loads(accounts_json)
    except Exception as e:
        raise RuntimeError(f"[{sys._getframe().f_code.co_name}] {ACCOUNTS_KEY} 不是合法 JSON：{e}") from e
    if not accounts:
        raise RuntimeError(f"[{sys._getframe().f_code.co_name}] 账号列表为空，请在 得卡 App 登录")

    # 多账号时取 tokenObtainedAtMs 最大（最近登录/续期）的那个作为当前账号
    def _obtained(acc: dict) -> int:
        """取账号 userInfo 里的 tokenObtainedAtMs，缺失按 0。"""
        return int((acc.get("userInfo") or {}).get("tokenObtainedAtMs") or 0)

    cur = max(accounts, key=_obtained)
    info = cur.get("userInfo") or {}
    access = info.get("accessToken")
    refresh = info.get("refreshToken")
    if not access or not refresh:
        raise RuntimeError(f"[{sys._getframe().f_code.co_name}] 当前账号缺 accessToken/refreshToken，请重登")

    exp = _decode_jwt_exp(access)
    if exp is None:                                                # JWT 解不出 → 用获取时刻 + expiresIn
        obtained_s = _obtained(cur) // 1000 or int(time.time())
        exp = obtained_s + int(info.get("expiresIn") or 900)

    return {"access": access, "refresh": refresh, "exp": exp,
            "userId": info.get("userId"), "phone": info.get("phone")}


def write_token_json(tk: dict) -> None:
    """把 token 三元组覆盖写入 token.json（格式与旧抓包脚本一致，消费方零改动）。

    Args:
        tk (dict): read_device_token 的返回值，至少含 access/refresh/exp。
    """
    payload = {"access": tk["access"], "refresh": tk["refresh"], "exp": tk["exp"]}
    with open(TOKEN_FILE, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False)


def main() -> None:
    """读设备 token → 覆盖写 token.json → 打印脱敏摘要。"""
    tk = read_device_token()
    write_token_json(tk)
    acc, exp, now = tk["access"], tk["exp"], int(time.time())
    left = exp - now
    print(f"[OK] token.json 已更新 -> {TOKEN_FILE}")
    print(f"   userId : {tk.get('userId')}   phone : {tk.get('phone')}")
    print(f"   access : JWT 长{len(acc)} 头[{acc[:10]}] 尾[{acc[-6:]}]")
    print(f"   exp    : {exp} ({time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(exp))})"
          f"  剩余 {left} 秒 {'有效' if left > 0 else '已过期(App 侧会自动续期，稍后重跑)'}")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as exc:
        print(exc)
        sys.exit(1)