gpt/plan_check_session.py

"""通过 access_token 实时调用 ChatGPT 后端 accounts/check 查询真实套餐级别。

用 curl_cffi(firefox144 指纹)绕过 Cloudflare，读取 plan_type / has_active_subscription。
作为独立脚本，由主程序用 .venv 的 python 以子进程方式调用（stdin 传 JSON，stdout 回 JSON），
与 card_bind_session.py 的调用方式一致（主程序 C:\\Python311 的 curl_cffi 版本过旧，不支持 firefox144）。

stdin  : {"access_token": "...", "account_id": "(可选)", "proxy": "(可选, DIRECT 或 http/socks URL)"}
stdout : {"ok": bool, "httpStatus": int, "plan": "free|plus|pro|...", "active": bool,
          "subPlan": "", "accountId": "", "error": ""}
"""
import sys
import json
import uuid
import base64

from curl_cffi.requests import Session

CHECK_URL = "https://chatgpt.com/backend-api/accounts/check/v4-2023-04-27"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:144.0) Gecko/20100101 Firefox/144.0"


def _decode_jwt(token: str) -> dict:
    try:
        seg = token.split(".")[1].replace("-", "+").replace("_", "/")
        seg += "=" * ((4 - len(seg) % 4) % 4)
        data = json.loads(base64.b64decode(seg).decode("utf-8", "replace"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _account_id_from_token(token: str) -> str:
    auth = (_decode_jwt(token) or {}).get("https://api.openai.com/auth") or {}
    return str(auth.get("chatgpt_account_id") or "")


def _walk_plan(data):
    """深扫响应，取 plan_type / has_active_subscription / subscription_plan（对齐 probe.js）。"""
    plan = ""
    active = False
    sub = ""
    stack = [data]
    while stack:
        cur = stack.pop()
        if isinstance(cur, dict):
            for k, v in cur.items():
                kl = str(k).lower()
                if kl == "plan_type" and isinstance(v, str) and not plan:
                    plan = v
                if kl == "has_active_subscription" and v is True:
                    active = True
                if kl == "subscription_plan" and isinstance(v, str) and not sub:
                    sub = v
                stack.append(v)
        elif isinstance(cur, list):
            stack.extend(cur)
    return plan, active, sub


def _account_id_from_payload(data) -> str:
    if not isinstance(data, dict):
        return ""
    ordering = data.get("account_ordering")
    if isinstance(ordering, list):
        for x in ordering:
            if x and x != "default":
                return str(x)
    accounts = data.get("accounts")
    if isinstance(accounts, dict):
        for k in accounts.keys():
            if k and k != "default":
                return str(k)
    return ""


def main():
    try:
        payload = json.loads(sys.stdin.read() or "{}")
    except Exception:
        payload = {}
    token = str(payload.get("access_token") or "").strip()
    account_id = str(payload.get("account_id") or "").strip() or _account_id_from_token(token)
    proxy = str(payload.get("proxy") or "").strip()
    if not token:
        print(json.dumps({"ok": False, "httpStatus": 0, "error": "missing_access_token"}))
        return

    http = Session(impersonate="firefox144")
    http.trust_env = False
    if proxy and proxy.upper() != "DIRECT":
        http.proxies = {"http": proxy, "https": proxy}

    device_id = str(uuid.uuid4())
    headers = {
        "Authorization": "Bearer " + token,
        "Accept": "application/json",
        "User-Agent": UA,
        "Origin": "https://chatgpt.com",
        "Referer": "https://chatgpt.com/",
        "OAI-Device-Id": device_id,
        "oai-device-id": device_id,
    }
    if account_id:
        headers["ChatGPT-Account-Id"] = account_id
        headers["ChatGPT-Account-ID"] = account_id

    try:
        # 暖身，拿 Cloudflare cookie
        try:
            http.get(
                "https://chatgpt.com/",
                headers={"User-Agent": UA, "Accept-Language": "en-US,en;q=0.9"},
                timeout=25,
            )
        except Exception:
            pass
        r = http.get(CHECK_URL, headers=headers, timeout=30)
        status = r.status_code
        if status == 200:
            data = r.json()
            plan, active, sub = _walk_plan(data)
            print(json.dumps({
                "ok": True,
                "httpStatus": status,
                "plan": plan or "",
                "active": bool(active),
                "subPlan": sub or "",
                "accountId": account_id or _account_id_from_payload(data),
            }))
        else:
            body = (r.text or "")[:300]
            print(json.dumps({"ok": False, "httpStatus": status, "error": body or ("HTTP " + str(status))}))
    except Exception as exc:  # noqa: BLE001
        print(json.dumps({"ok": False, "httpStatus": 0, "error": str(exc)}))


if __name__ == "__main__":
    main()