"""对历史失败账号做补救:用 access_token 调 backend-api 检查 plan,若已 plus 则上传 CPA。 注意:/api/auth/session 走 NextAuth 只认 cookie,不能用 Bearer 直连。 真正的鉴权接口是 /backend-api/me(也叫 accounts/check),它接受 Authorization: Bearer。 """ from __future__ import annotations import copy import json import time import urllib.error import urllib.request from typing import Callable from cpa_uploader import ( get_session_plan_type, is_plus_session, upload_session_to_cpa, ) from storage import add_event, get_account, upsert_account ME_URL = "https://chatgpt.com/backend-api/me" ACCOUNTS_CHECK_URL = "https://chatgpt.com/backend-api/accounts/check/v4-2023-04-27" def _http_get_json(url: str, *, headers: dict, timeout: int = 20, log: Callable[[str], None] = print) -> tuple[int, str, dict]: started = time.time() text = "" status = 0 try: from curl_cffi import requests as curl_requests # type: ignore r = curl_requests.get(url, headers=headers, impersonate="chrome136", timeout=timeout) text = r.text status = r.status_code except Exception as exc: log(f"[recheck] curl_cffi 不可用: {exc!r},回退 urllib") req = urllib.request.Request(url, method="GET") for k, v in headers.items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=timeout) as resp: text = resp.read().decode("utf-8", errors="replace") status = resp.status except urllib.error.HTTPError as exc2: text = exc2.read().decode("utf-8", errors="replace") if hasattr(exc2, "read") else "" status = exc2.code elapsed_ms = int((time.time() - started) * 1000) log(f"[recheck] GET {url} HTTP {status} 耗时 {elapsed_ms}ms 长度={len(text)}") parsed: dict = {} try: parsed = json.loads(text or "{}") except Exception: parsed = {} return status, text, parsed def _extract_plan_from_me(payload: dict) -> tuple[str, str, str]: """从 /backend-api/me 或 /accounts/check 响应里抽 (plan_type, account_id, email)。""" if not isinstance(payload, dict): return "", "", "" # /me 里直接给 email 和 chat_id;plan 通常在 accounts..account.plan_type email = "" if isinstance(payload.get("email"), str): email = payload["email"] accounts = payload.get("accounts") if isinstance(accounts, dict): # 优先找 plan_type='plus' best_id = "" best_plan = "" for acc_id, item in accounts.items(): if not isinstance(item, dict): continue account_obj = item.get("account") if isinstance(item.get("account"), dict) else item plan = (account_obj or {}).get("plan_type") or (account_obj or {}).get("planType") or "" plan = str(plan or "").strip() if not plan: continue if plan.lower() == "plus" and not best_plan: best_plan = plan best_id = acc_id break if not best_plan: best_plan = plan best_id = acc_id if best_plan: return best_plan, best_id, email # 兜底:顶层 plan_type plan = str(payload.get("plan_type") or payload.get("planType") or "").strip() return plan, "", email def _merge_plus_into_session(base_session: dict, plan: str, account_id: str, email: str) -> dict: """把 backend-api 拉到的 plan/account_id 合并进 DB 里的 session,构造给 CPA 用的 plus session。""" sess = copy.deepcopy(base_session) if isinstance(base_session, dict) else {} if "account" not in sess or not isinstance(sess.get("account"), dict): sess["account"] = {} if plan: sess["account"]["planType"] = plan sess["planType"] = plan if account_id and not sess["account"].get("id"): sess["account"]["id"] = account_id if email: if "user" not in sess or not isinstance(sess.get("user"), dict): sess["user"] = {} if not sess["user"].get("email"): sess["user"]["email"] = email return sess def recheck_account( email: str, *, cpa_url: str, cpa_management_key: str, log: Callable[[str], None] = print, ) -> dict: """对单个账号做补救。返回 { ok, planType, action, cpa, error }。""" acc = get_account(email) if not acc: return {"ok": False, "error": "账号不存在"} plus_sess = acc.get("plus_session") or {} init_sess = acc.get("initial_session") or {} access_token = (plus_sess.get("accessToken") if isinstance(plus_sess, dict) else None) \ or (init_sess.get("accessToken") if isinstance(init_sess, dict) else None) if not access_token: return {"ok": False, "error": "数据库里找不到 accessToken"} log(f"[recheck] {email} 开始补救(访问 backend-api/me)") add_event(email, "recheck", "info", "begin") headers = { "Authorization": f"Bearer {access_token}", "Accept": "application/json", "Origin": "https://chatgpt.com", "Referer": "https://chatgpt.com/", } plan = "" account_id = "" me_email = "" last_payload: dict = {} # 先试 /backend-api/me(轻量),失败回退 /accounts/check for url in (ME_URL, ACCOUNTS_CHECK_URL): try: status, text, payload = _http_get_json(url, headers=headers, log=log) except Exception as exc: log(f"[recheck] {url} 异常: {exc!r}") continue if status >= 400: log(f"[recheck] {url} 返回 {status},预览={text[:200]}") continue last_payload = payload plan, account_id, me_email = _extract_plan_from_me(payload) log(f"[recheck] {url} 提取 plan={plan!r} account_id={account_id!r} email={me_email!r}") if plan: break if not plan: msg = "重拉 session 失败:所有 backend-api 接口都未返回 plan_type" log(f"[recheck] {email} {msg}") upsert_account(email, acc.get("password") or "", fields={"last_error": msg}) add_event(email, "recheck", "error", msg) return {"ok": False, "error": msg} # 用 base session(plus_session 优先,否则 init_session)合并新 plan base_for_merge = plus_sess if isinstance(plus_sess, dict) and plus_sess else init_sess merged = _merge_plus_into_session(base_for_merge or {}, plan, account_id, me_email or email) # accessToken 用我们手头那个(merged 不一定包含) if not merged.get("accessToken"): merged["accessToken"] = access_token upsert_account(email, acc.get("password") or "", fields={ "plan_type": plan, "plus_session": merged, }) if not is_plus_session(merged): upsert_account(email, acc.get("password") or "", fields={ "final_status": "plus_check_failed", "last_error": f"补救后 planType 仍为 {plan!r}", }) add_event(email, "recheck", "warn", f"plan={plan}") return {"ok": False, "planType": plan, "action": "still_not_plus"} upsert_account(email, acc.get("password") or "", fields={ "final_status": "plus", "last_error": "" }) add_event(email, "plus_check", "ok", f"plan={plan} (recheck)") if not (cpa_url and cpa_management_key): upsert_account(email, acc.get("password") or "", fields={ "final_status": "cpa_skipped" }) add_event(email, "cpa", "warn", "未配置 CPA") return {"ok": True, "planType": plan, "action": "plus_no_cpa_config"} try: cpa_result = upload_session_to_cpa( merged, cpa_url=cpa_url, management_key=cpa_management_key, email_hint=email, log=log, ) except Exception as exc: msg = f"CPA 上传失败: {exc}" log(f"[recheck] {email} {msg}") upsert_account(email, acc.get("password") or "", fields={ "final_status": "cpa_failed", "last_error": msg, }) add_event(email, "cpa", "error", msg) return {"ok": False, "planType": plan, "action": "cpa_failed", "error": msg} upsert_account(email, acc.get("password") or "", fields={ "final_status": "cpa_uploaded", "cpa_file_name": cpa_result.get("fileName"), "cpa_uploaded_at": int(time.time() * 1000), "last_error": "", }) add_event(email, "cpa", "ok", cpa_result.get("fileName"), payload=cpa_result) return { "ok": True, "planType": plan, "action": "cpa_uploaded", "cpa": cpa_result, }