"""Qoder 国际版 用量读取/签到脚本
凭证链（两种方式，按优先级自动选择）：
方式 A（推荐，最稳定）: PAT 换 job token
1. 在 https://qoder.com → Settings → Personal Access Tokens 创建 PAT（pt-...）
2. POST /api/v1/jobToken/exchange {"personal_token": "<pt-...>"}
-> 返回 {accessToken(jt-...), refreshToken(jrt-...), ...}
3. 后续请求用 Bearer jt-...
4. jt 过期时 POST /api/v1/jobToken/refresh 用 jrt 换新的 jt
方式 B（同 CN 脚本链路）: 客户端本地凭证仓
Local State -> os_crypt.encrypted_key(base64 "DPAPI"…) -> CryptUnprotectData
-> AES-256 key -> 解 auth.v1.dat（"v10" + 12B nonce + ct + 16B tag）
（国际版客户端目录名需实测，默认尝试 com.qoder.app.stable）
接口域名：
OpenAPI: https://openapi.qoder.sh
聊天网关: https://api3.qoder.sh/algo/api/v2/ （本脚本用不到）
实测注意：
- PAT 不能直接放 Authorization，必须先 exchange 成 jt
- campaigns 接口同样需要 Cosy-ClientType 头，否则返回空数组
- 若 401/TOKEN_INVALID，优先检查 jt 是否过期，尝试 refresh
"""
import argparse
import base64
import ctypes
import ctypes.wintypes as wt
import json
import os
import sys
import requests
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
# ====================== 国际版差异配置（按需修改）======================
API = "https://openapi.qoder.sh" # 国际版 OpenAPI 域名
# 客户端凭证目录：国际版目录名需实测，常见候选如下，脚本会依次尝试
PROFILE_CANDIDATES = [
"com.qoder.app.stable",
"com.qoder.app",
"com.qoder.stable",
"Qoder",
]
# ======================================================================
# Windows 控制台默认 GBK，中文输出会花屏
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
CLIENT_TYPE = "10" # 客户端日志里的 clientType
VERSION = "0.4.2" # 客户端版本号，可按实际安装版本调整
AUTH_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "auths.json")
# ---------------------- Windows DPAPI 解密 ----------------------
class _Blob(ctypes.Structure):
_fields_ = [("cbData", wt.DWORD), ("pbData", ctypes.POINTER(ctypes.c_char))]
def _dpapi_unprotect(data: bytes) -> bytes:
kernel32 = ctypes.windll.kernel32
inp, out = _Blob(), _Blob()
inp.cbData = len(data)
inp.pbData = ctypes.create_string_buffer(data)
if not ctypes.windll.crypt32.CryptUnprotectData(
ctypes.byref(inp), None, None, None, None, 0, ctypes.byref(out)
):
raise OSError(f"CryptUnprotectData failed, GetLastError={kernel32.GetLastError()}")
try:
return ctypes.string_at(out.pbData, out.cbData)
finally:
kernel32.LocalFree(out.pbData)
def _find_profile() -> str:
"""依次尝试候选目录名，找到包含 Local State 的那个。"""
appdata = os.environ["APPDATA"]
for name in PROFILE_CANDIDATES:
path = os.path.join(appdata, name)
if os.path.isfile(os.path.join(path, "Local State")):
return path
raise FileNotFoundError(
f"未找到 Qoder 国际版凭证目录，尝试过: {PROFILE_CANDIDATES}\n"
f"请打开 %APPDATA% 查看实际目录名后修改 PROFILE_CANDIDATES"
)
def load_auth_from_client() -> dict:
"""解密国际版客户端登录仓，返回 {token, refreshToken, expiresAt, user, ...}。"""
profile = _find_profile()
with open(os.path.join(profile, "Local State"), encoding="utf-8") as fh:
raw = base64.b64decode(json.load(fh)["os_crypt"]["encrypted_key"])
if raw[:5] != b"DPAPI":
raise ValueError("unexpected encrypted_key prefix")
key = _dpapi_unprotect(raw[5:])
with open(os.path.join(profile, "auth.v1.dat"), "rb") as fh:
blob = fh.read()
if blob[:3] != b"v10":
raise ValueError("auth.v1.dat is not a v10 envelope")
return json.loads(AESGCM(key).decrypt(blob[3:15], blob[15:], None))
# ---------------------- PAT 换 job token ----------------------
def exchange_pat(pat: str) -> dict:
"""用 PAT (pt-...) 换短时 job token (jt-...)。
返回示例（字段名以实测为准）:
{ "accessToken": "jt-...", "refreshToken": "jrt-...", "expiresIn": 3600, ... }
"""
r = requests.post(
f"{API}/api/v1/jobToken/exchange",
json={"personal_token": pat},
headers={"Accept": "application/json", "User-Agent": "Qoder"},
timeout=20,
)
if r.status_code != 200:
raise RuntimeError(f"jobToken/exchange 失败 HTTP {r.status_code}: {r.text[:300]}")
return r.json()
def refresh_job_token(refresh_token: str) -> dict:
"""用 jrt-... 刷新 jt-...，返回同 exchange 结构。"""
r = requests.post(
f"{API}/api/v1/jobToken/refresh",
json={"refreshToken": refresh_token},
headers={"Accept": "application/json", "User-Agent": "Qoder"},
timeout=20,
)
if r.status_code != 200:
raise RuntimeError(f"jobToken/refresh 失败 HTTP {r.status_code}: {r.text[:300]}")
return r.json()
# ---------------------- Session 构造 ----------------------
def make_session(token: str) -> requests.Session:
"""公共头挂到 session 上；国际版 Authorization 同样必须是 Bearer <jt-...>。"""
s = requests.Session()
s.headers.update({
"Accept": "application/json",
"User-Agent": "Qoder",
"Authorization": f"Bearer {token}",
"Cosy-ClientType": CLIENT_TYPE,
"Cosy-Version": VERSION,
})
return s
def get(session: requests.Session, path: str, params: dict | None = None):
r = session.get(API + path, params=params, timeout=20)
if r.status_code != 200:
return {"_error": r.status_code, "_body": r.text[:400]}
return r.json()
# ---------------------- 接口清单 ----------------------
# 路径与 CN 版一致（同属 sash 服务），若 404 请用 --dump-asar 自查
ENDPOINTS = {
"credits-summary": ("/sash/api/v1/ai-conversations/credits-summary", None),
"seat-activity": ("/sash/api/v1/ai-conversations/seat-activity", None),
"credits-heatmap": ("/sash/api/v1/ai-conversations/credits-heatmap", {"days": 30, "organization_id": ""}),
"campaigns": ("/sash/api/v1/me/campaigns", None),
"plan": ("/api/v2/user/plan", None),
"quota": ("/api/v2/quota/usage", None),
}
# ---------------------- 签到逻辑 ----------------------
def claim_one(session: requests.Session, campaign_id: str) -> dict:
"""POST /sash/api/v1/me/campaigns/{campaignId}/claim（服务端幂等）"""
r = session.post(f"{API}/sash/api/v1/me/campaigns/{campaign_id}/claim", timeout=20)
try:
body = r.json()
except ValueError:
body = {"_raw": r.text[:300]}
return {"_http": r.status_code, **body} if isinstance(body, dict) else {"_http": r.status_code, "_raw": body}
def print_quota(session: requests.Session, prefix: str = "剩余额度") -> None:
"""拉取 quota/usage 并打印可用额度。"""
quota = get(session, ENDPOINTS["quota"][0])
if "_error" in quota:
print(f"{prefix} : 请求失败 {quota['_error']}")
return
parts = []
for label, key in (("套餐", "userQuota"), ("签到/加油包", "addOnQuota")):
q = quota.get(key) or {}
if q.get("total"):
parts.append(f"{label} 剩余 {q.get('remaining'):g}/{q.get('total'):g}")
print(f"{prefix} : {'，'.join(parts) if parts else '无可用额度'}"
f"{'（已超额）' if quota.get('isQuotaExceeded') else ''}")
def do_claim(session: requests.Session, dry_run: bool) -> int:
doc = get(session, "/sash/api/v1/me/campaigns")
if "_error" in doc:
print(f"活动列表请求失败: {doc['_error']} {doc['_body'][:120]}")
return 1
campaigns = doc.get("campaigns") or []
if not campaigns:
print("当前没有可展示的活动")
return 0
for c in campaigns:
b = c.get("benefit") or {}
print(f"{c.get('campaignId')} {c.get('campaignKey')} {str(c.get('actionType', '')):13} "
f"{str(c.get('claimStatus', '')):9} {b.get('kind')}x{b.get('amount')}")
targets = [c for c in campaigns
if c.get("claimStatus") == "CLAIMABLE" and c.get("actionType") == "CLAIM_BENEFIT"]
if not targets:
print("\n没有 CLAIMABLE 的活动，无需签到")
return 0
if dry_run:
print(f"\n[dry-run] 将领取 {len(targets)} 条：{[t.get('campaignKey') for t in targets]}")
return 0
print()
claimed = False
for c in targets:
res = claim_one(session, c["campaignId"])
b = res.get("benefit") or {}
print(f"签到 {c.get('campaignKey')} -> HTTP {res.get('_http')} status={res.get('status')} "
f"replayed={res.get('replayed')} 权益={b.get('kind')}x{b.get('amount')} "
f"发放单={res.get('grantId')}")
if res.get("_http") == 200 and res.get("status") == "CLAIMED":
claimed = True
if claimed:
print("\n签到已到账，最新额度：")
print_quota(session, prefix="剩余额度(签到后)")
return 0
def show_usage(session: requests.Session, auth: dict, args) -> int:
"""usage 模式（默认）= 面板数据汇总 + 自动签到。"""
user = auth.get("user") or {}
summary = get(session, ENDPOINTS["credits-summary"][0])
streak = get(session, ENDPOINTS["seat-activity"][0])
heat = get(session, ENDPOINTS["credits-heatmap"][0],
{"days": args.days or 30, "organization_id": ""})
print(f"账号 : {user.get('name')} ({user.get('id')})")
print(f"token 到期: {auth.get('expiresAt')}")
print_quota(session)
if "_error" not in summary:
print(f"Credits : 累计 {summary.get('totalCredits')} / 峰值 {summary.get('peakCredits')}")
else:
print(f"Credits : 请求失败 {summary['_error']} {summary['_body'][:120]}")
if "_error" not in streak:
print(f"活跃天数 : 累计 {streak.get('cumulativeActiveDays')} 天 / 当前连续 "
f"{streak.get('currentConsecutiveDays')} 天 / 最长 {streak.get('maxConsecutiveDays')} 天"
f"（最后活跃 {streak.get('lastActiveDate')}）")
else:
print(f"活跃天数 : 请求失败 {streak['_error']}")
if "_error" not in heat:
items = heat.get("items") or []
used = [i for i in items if i.get("value")]
print(f"热力图 : {items[0].get('date', '?') if items else '?'} ~ "
f"{items[-1].get('date', '?') if items else '?'}，"
f"总和 {heat.get('total')}，有用量的天数 {len(used)}")
for i in used[-5:]:
print(f" {i.get('date')} {i.get('value')}")
print()
return do_claim(session, args.dry_run)
# ---------------------- 凭证加载 ----------------------
def load_auths(args) -> list[dict]:
"""凭证优先级：
1. --pat <pt-...> 命令行传 PAT（单账号，自动 exchange）
2. --auth-file <json> JSON 文件（支持两种格式，见下）
3. 同目录 auths.json 同上
4. 客户端本地解密 同 CN 脚本链路
auth-file JSON 支持两种条目格式：
A. {"pat": "pt-..."} -> 脚本自动 exchange
B. {"token": "jt-...", "refreshToken": "jrt-...", "user": {...}}
-> 直接用 token，若 401 自动用 refreshToken 刷新
"""
# 1) 命令行 PAT
if args.pat:
cred = exchange_pat(args.pat)
# 补齐 user 字段
s = make_session(cred["accessToken"])
info = get(s, "/api/v1/userinfo")
if "_error" not in info:
cred["user"] = info
return [cred]
# 2/3) auth-file
path = args.auth_file or (AUTH_FILE if os.path.exists(AUTH_FILE) else None)
if path:
result = []
with open(path, encoding="utf-8") as fh:
data = json.load(fh)
items = data if isinstance(data, list) else [data]
for item in items:
if "pat" in item:
# PAT 条目：自动 exchange
cred = exchange_pat(item["pat"])
cred["user"] = item.get("user") or {}
result.append(cred)
else:
# 直接给 token 的条目
result.append(item)
return result
# 4) 客户端本地解密（需先登录过国际版客户端）
return [load_auth_from_client()]
def run_account(auth: dict, args) -> int:
"""单账号执行；若 token 失效且有 refreshToken，自动刷新重试一次。"""
token = auth.get("token") or auth.get("accessToken")
refresh = auth.get("refreshToken")
for attempt in range(2):
session = make_session(token)
# 探测 token 是否有效
probe = get(session, "/api/v2/quota/usage")
if "_error" not in probe:
break
# 401 且有 refreshToken 时刷新重试
if attempt == 0 and refresh and probe.get("_error") in (401, 403):
print(" token 失效，尝试用 refreshToken 刷新...")
try:
new = refresh_job_token(refresh)
token = new.get("accessToken") or new.get("token") or token
refresh = new.get("refreshToken") or refresh
continue
except Exception as exc:
print(f" 刷新失败: {exc}")
return 1
return 1
if args.what == "claim":
return do_claim(session, args.dry_run)
if args.what in ENDPOINTS:
path, params = ENDPOINTS[args.what]
if args.what == "credits-heatmap" and args.days:
params = {**(params or {}), "days": args.days}
out = get(session, path, params)
print(json.dumps(out, ensure_ascii=False, indent=2))
return 1 if "_error" in out else 0
return show_usage(session, auth, args)
def main() -> int:
ap = argparse.ArgumentParser(
description="读取 Qoder 国际版 用量 / 签到",
epilog=(
"凭证来源（优先级从高到低）：\n"
" 1. --pat pt-xxxx 在 qoder.com 创建 PAT 后直接传入\n"
" 2. --auth-file x.json 多账号 JSON（支持 pat 或 token+refreshToken 两种条目）\n"
" 3. ./auths.json 同目录默认文件\n"
" 4. 客户端本地解密 需先在国际版客户端登录\n"
"\n"
"auth-file 条目示例：\n"
' [{"pat": "pt-aaa"}, {"token": "jt-bbb", "refreshToken": "jrt-ccc"}]\n'
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
ap.add_argument("what", nargs="?", default="usage",
help="usage | claim | " + " | ".join(ENDPOINTS))
ap.add_argument("--days", type=int, help="credits-heatmap 的天数")
ap.add_argument("--dry-run", action="store_true", help="claim 时只列出将要领取的活动")
ap.add_argument("--auth-file", help="账号凭证 JSON 路径")
ap.add_argument("--pat", help="Personal Access Token（pt-...），脚本会自动 exchange")
args = ap.parse_args()
rc = 0
for idx, auth in enumerate(load_auths(args), 1):
name = (auth.get("user") or {}).get("name") or "?"
print(f"===== 账号 {idx}: {name} =====")
try:
rc |= run_account(auth, args)
except Exception as exc:
print(f"账号 {name} 处理失败: {exc}", file=sys.stderr)
rc = 1
return rc
if __name__ == "__main__":
sys.exit(main())
风咕嘎 09-24 21:151楼
ceshi01Adx 09-24 21:182楼
MyCard 09-24 21:223楼