Qoder国际版签到脚本

"""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())