313 lines
12 KiB
Python
313 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""查询斗鱼和平小店的完整账号数据。
|
|
|
|
使用方式：
|
|
1. 修改下面的 COOKIE；必要时修改活动配置。
|
|
2. 在仓库根目录运行：python3 douyu_xpd_query.py
|
|
|
|
Cookie 只在本机使用，不要提交到 git 或发送给他人。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, unquote, urlsplit
|
|
|
|
import requests
|
|
|
|
|
|
# ========================== 只需要修改这里 ==========================
|
|
COOKIE = ""
|
|
|
|
# 当前和平小店配置；活动更换后以斗鱼页面/项目配置中的值为准。
|
|
ACT_ALIAS = "20260623KDQFH"
|
|
ACT_ID = "46195"
|
|
ROOM_ID = "9263298"
|
|
# =====================================================================
|
|
|
|
|
|
def dump_json(value: Any) -> None:
|
|
"""以完整 JSON 输出，中文不转义。"""
|
|
print(json.dumps(value, ensure_ascii=False, indent=2, default=str))
|
|
|
|
|
|
class QueryError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class XpdClient:
|
|
"""和平小店查询链路的最小独立实现。"""
|
|
|
|
UA = (
|
|
"Mozilla/5.0 (Linux; Android 12; HBN-AL00 Build/V417IR; wv) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
|
|
"Chrome/101.0.4951.61 Mobile Safari/537.36, Douyu_Android"
|
|
)
|
|
|
|
def __init__(self, cookie: str):
|
|
self.cookie = cookie.strip()
|
|
self.session = requests.Session()
|
|
self.session.trust_env = False
|
|
|
|
def _request(self, url: str, *, params: dict[str, Any], referer: str) -> requests.Response:
|
|
response = self.session.get(
|
|
url,
|
|
params=params,
|
|
headers={
|
|
"User-Agent": self.UA,
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Accept-Language": "zh-CN,zh;q=0.9",
|
|
"X-Requested-With": "air.tv.douyu.android",
|
|
"Referer": referer,
|
|
"Cookie": self.cookie,
|
|
},
|
|
timeout=(8, 20),
|
|
)
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
@staticmethod
|
|
def _parse_var(text: str, name: str) -> dict[str, Any]:
|
|
match = re.search(rf"var\s+{re.escape(name)}\s*=", text)
|
|
if not match:
|
|
raise QueryError(f"响应中找不到 var {name}: {text[:300]}")
|
|
chunk = text[match.end():].strip()
|
|
if chunk.endswith(";"):
|
|
chunk = chunk[:-1].rstrip()
|
|
try:
|
|
return json.loads(chunk)
|
|
except json.JSONDecodeError as exc:
|
|
raise QueryError(f"var {name} 不是合法 JSON: {text[:300]}") from exc
|
|
|
|
@staticmethod
|
|
def _int(value: Any) -> int | None:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
def _cookie_parts(self) -> dict[str, str]:
|
|
return {
|
|
item.split("=", 1)[0]: item.split("=", 1)[1]
|
|
for item in self.cookie.split("; ")
|
|
if "=" in item
|
|
}
|
|
|
|
def _token(self) -> str:
|
|
parts = self._cookie_parts()
|
|
for key in ("acf_uid", "acf_stk", "acf_ltkid"):
|
|
if not parts.get(key):
|
|
raise QueryError(f"Cookie 缺少 {key}")
|
|
return f"{parts['acf_uid']}_1_{parts['acf_stk']}_0_{parts['acf_ltkid']}"
|
|
|
|
def profile(self) -> dict[str, str]:
|
|
parts = self._cookie_parts()
|
|
return {
|
|
"uid": parts.get("acf_uid", ""),
|
|
"nickname": unquote(parts.get("acf_nickname", "")),
|
|
}
|
|
|
|
def embed(self) -> dict[str, Any]:
|
|
response = self._request(
|
|
"https://www.douyu.com/japi/carnival/nc/txEmbed/getIframeUrl",
|
|
params={"actAlias": ACT_ALIAS, "rid": ROOM_ID, "token": self._token()},
|
|
referer="https://www.douyu.com/topic/h5/hpxd01",
|
|
)
|
|
payload = response.json()
|
|
if payload.get("error") not in (0, "0", None):
|
|
raise QueryError(payload.get("msg") or f"获取 H5 参数失败: {payload}")
|
|
txurl = str((payload.get("data") or {}).get("txurlH5") or "")
|
|
if not txurl:
|
|
raise QueryError(f"响应没有 data.txurlH5: {payload}")
|
|
query = {key: values[0] for key, values in parse_qs(urlsplit(txurl).query).items() if values}
|
|
return {"query": query, "txurl_h5": txurl, "raw": payload}
|
|
|
|
def role(self, embed: dict[str, Any]) -> dict[str, Any]:
|
|
query = embed["query"]
|
|
response = self._request(
|
|
"https://apps.game.qq.com/daoju/igw/live",
|
|
params={
|
|
"acctype": "livelink", "_jsvar": "info",
|
|
"_service": "other.livelink.getrole.out", "_biz_code": "cjm",
|
|
"_act_id": ACT_ID, "_app_id": "2123", "isCode": "1",
|
|
"gameId": query.get("gameId", "cjm"), "actId": query.get("actId", ""),
|
|
"appId": query.get("appId", "bp_cf"), "livePlatId": query.get("livePlatId", "douyu"),
|
|
"code": query.get("code", ""), "timestamp": query.get("timestamp", ""),
|
|
"v": query.get("v", ""), "sig": query.get("sig", ""),
|
|
"authType": "delegate", "sAnchorId": ROOM_ID, "sVideoId": "",
|
|
},
|
|
referer="https://app.daoju.qq.com/",
|
|
)
|
|
info = self._parse_var(response.text, "info")
|
|
plat_id = info.get("platId")
|
|
return {
|
|
"game_open_id": str(info.get("gameOpenId") or ""),
|
|
"role_id": str(info.get("roleId") or ""),
|
|
"role_name": str(info.get("roleName") or ""),
|
|
"type": str(info.get("type") or ""),
|
|
# 0 是合法的 iOS 平台值，不能用 `or` 当作缺失处理。
|
|
"plat_id": str(plat_id) if plat_id is not None else "",
|
|
"raw": info,
|
|
}
|
|
|
|
def bind_info(self) -> dict[str, Any]:
|
|
response = self._request(
|
|
"https://www.douyu.com/japi/carnivalApi/v2/tencent/bindInfo",
|
|
params={"actAlias": ACT_ALIAS, "token": self._token()},
|
|
referer="https://www.douyu.com/",
|
|
)
|
|
payload = response.json()
|
|
if payload.get("error") not in (0, "0", None):
|
|
raise QueryError(payload.get("msg") or f"绑定信息查询失败: {payload}")
|
|
return payload
|
|
|
|
@staticmethod
|
|
def _area_id(role: dict[str, Any]) -> str:
|
|
"""优先使用 getrole 返回的大区，避免把 QQ/微信角色混查。"""
|
|
raw_area = role.get("raw", {}).get("area")
|
|
if raw_area not in (None, ""):
|
|
return str(raw_area)
|
|
return "1" if role.get("type") == "wx" else "2" if role.get("type") == "qq" else "1"
|
|
|
|
def _wallet_request(self, role: dict[str, Any], *, gamecoin: bool) -> dict[str, Any]:
|
|
areaid = self._area_id(role)
|
|
role_plat = role.get("plat_id")
|
|
plat = str(role_plat) if role_plat not in (None, "") else "1"
|
|
params = {
|
|
"_jsvar": "jbInfos" if gamecoin else "banlanceInfo",
|
|
"_service": "pay.idip.gamecoin.get" if gamecoin else "pay.midas.dq.get.ttpp",
|
|
"_app_id": "2123", "acctype": "ttpp", "areaid": areaid,
|
|
"eventid": "", "interwork": "0", "openid": role["game_open_id"],
|
|
"openkey": "openkey", "partition": "0", "pay_token": "",
|
|
"plat": plat, "plat_pc": "0", "roleid": role["role_id"],
|
|
"_biz_code": "cjm", "_act_id": ACT_ID, "_sid": "6",
|
|
}
|
|
if not gamecoin:
|
|
# 浏览器抓包使用毫秒时间戳；该参数也可避免复用旧请求。
|
|
params["_time"] = str(int(time.time() * 1000))
|
|
else:
|
|
params["coin_type"] = "1"
|
|
response = self._request("https://apps.game.qq.com/daoju/igw/live/", params=params, referer="https://app.daoju.qq.com/")
|
|
info = self._parse_var(response.text, params["_jsvar"])
|
|
return {"params": params, "raw": info}
|
|
|
|
def wallet(self, role: dict[str, Any]) -> dict[str, Any]:
|
|
"""查询点券钱包，raw 中保留服务端返回的全部字段。"""
|
|
result = self._wallet_request(role, gamecoin=False)
|
|
result["balance"] = self._int(result["raw"].get("balance"))
|
|
return result
|
|
|
|
def gamecoins(self, role: dict[str, Any]) -> dict[str, Any]:
|
|
"""查询车币、扭蛋币、夺宝碎片等全部 gamecoin 字段。"""
|
|
result = self._wallet_request(role, gamecoin=True)
|
|
raw = result["raw"]
|
|
names = {
|
|
"dq": "点券",
|
|
"jb": "车币",
|
|
"jb2": "扭蛋币",
|
|
"jb3": "夺宝碎片",
|
|
"jb4": "战备积分",
|
|
"jb5": "精英币",
|
|
}
|
|
currencies: dict[str, dict[str, Any]] = {}
|
|
for key, value in raw.items():
|
|
if re.fullmatch(r"(?:dq|jb|coin)\d*", key) and not key.endswith(("_name", "_rate")):
|
|
currencies[key] = {
|
|
"name": raw.get(f"{key}_name") or names.get(key, ""),
|
|
"value": self._int(value),
|
|
"rate": self._int(raw.get(f"{key}_rate")),
|
|
}
|
|
# 服务端有时只返回 jb4/jb5 的名称和倍率，不返回数值；保留为 None，
|
|
# 不把缺失误报成 0。gamecoin.dq 同样不覆盖 pay.midas 的点券 balance。
|
|
for key in ("jb4", "jb5"):
|
|
if key not in currencies and (f"{key}_name" in raw or f"{key}_rate" in raw):
|
|
currencies[key] = {
|
|
"name": raw.get(f"{key}_name") or names[key],
|
|
"value": self._int(raw.get(key)),
|
|
"rate": self._int(raw.get(f"{key}_rate")),
|
|
}
|
|
result["currencies"] = currencies
|
|
return result
|
|
|
|
def main() -> int:
|
|
if not COOKIE.strip():
|
|
print("请先在脚本顶部填写 COOKIE（完整斗鱼 Cookie 字符串）", file=sys.stderr)
|
|
return 2
|
|
|
|
client = XpdClient(COOKIE)
|
|
result: dict[str, Any] = {
|
|
"config": {
|
|
"act_alias": ACT_ALIAS,
|
|
"act_id": ACT_ID,
|
|
"room_id": ROOM_ID,
|
|
},
|
|
"profile": client.profile(),
|
|
}
|
|
|
|
try:
|
|
print("[1/6] 获取和平小店临时授权参数...", flush=True)
|
|
embed = client.embed()
|
|
result["embed"] = embed
|
|
|
|
print("[2/6] 获取绑定角色...", flush=True)
|
|
role = client.role(embed)
|
|
result["role"] = role
|
|
|
|
openid = str(role.get("game_open_id") or "")
|
|
roleid = str(role.get("role_id") or "")
|
|
raw_plat = role.get("plat_id")
|
|
plat = str(raw_plat) if raw_plat not in (None, "") else "1"
|
|
# 和平小店接口约定：微信大区 1，手Q大区 2。
|
|
areaid = client._area_id(role)
|
|
result["role_request"] = {
|
|
"openid": openid,
|
|
"roleid": roleid,
|
|
"plat": plat,
|
|
"areaid": areaid,
|
|
"wallet_key": f"{openid}|{roleid}|{plat}|{areaid}",
|
|
}
|
|
|
|
print("[3/6] 查询斗鱼侧小店绑定信息...", flush=True)
|
|
result["bind_info"] = client.bind_info()
|
|
|
|
if not openid or not roleid:
|
|
raise QueryError("没有获取到 gameOpenId/roleId，账号可能尚未绑定和平精英角色")
|
|
|
|
print("[4/6] 查询点券余额及钱包明细...", flush=True)
|
|
result["balance"] = client.wallet(role)
|
|
|
|
print("[5/6] 查询车币、扭蛋币、碎片等 gamecoin...", flush=True)
|
|
result["gamecoins"] = client.gamecoins(role)
|
|
|
|
print("[6/6] 输出角色诊断信息...", flush=True)
|
|
result["diagnosis"] = {
|
|
"message": "余额按 openid、roleid、plat、areaid 四项区分；不要把不同平台角色合并。",
|
|
"role_name": role.get("role_name", ""),
|
|
"role_type": role.get("type", ""),
|
|
"area": role.get("raw", {}).get("area"),
|
|
"partition": role.get("raw", {}).get("partition"),
|
|
"plat_id": role.get("plat_id", ""),
|
|
}
|
|
except (QueryError, requests.RequestException, OSError) as exc:
|
|
result["error"] = str(exc)
|
|
print(f"请求失败：{exc}", file=sys.stderr)
|
|
# 已成功拿到的字段仍然输出，方便定位是哪个接口失败。
|
|
dump_json(result)
|
|
return 1
|
|
except Exception as exc: # 保证脚本调试时也能输出已有数据
|
|
result["error"] = f"{type(exc).__name__}: {exc}"
|
|
print(f"未预期错误：{type(exc).__name__}: {exc}", file=sys.stderr)
|
|
dump_json(result)
|
|
return 1
|
|
|
|
dump_json(result)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|