1357 lines
54 KiB
Python
1357 lines
54 KiB
Python
"""斗鱼活动、兑换与充值接口客户端。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import random
|
||
import re
|
||
import string
|
||
import time
|
||
from collections.abc import Callable
|
||
from typing import Any
|
||
from urllib.parse import parse_qs, unquote, urlsplit
|
||
|
||
import requests
|
||
|
||
from .cookie_utils import cookie_pairs, cookie_value, normalize_cookie_pairs, normalize_douyu_cookie
|
||
|
||
|
||
PC_UA = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/133.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
|
||
class DouyuActivityError(RuntimeError):
|
||
"""斗鱼活动接口异常。"""
|
||
|
||
|
||
class DouyuActivityClient:
|
||
"""封装斗鱼活动页、兑换、充值和送礼相关接口。"""
|
||
|
||
CSRF_API = "https://www.douyu.com/japi/carnival/nc/common/generateCsrf"
|
||
ACF_CCN_API = "https://www.douyu.com/curl/csrfApi/getCsrfCookie?"
|
||
ACF_CCN_FALLBACK_API = "https://www.douyu.com/curl/csrfNlApi/getCsrfCookie"
|
||
SUBSCRIBE_API = "https://www.douyu.com/wgapi/livenc/asubscribe/userSubStatus"
|
||
ROLE_PARAM_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/roleParam"
|
||
BIND_INFO_API = "https://www.douyu.com/japi/carnivalApi/tencent/bindInfo"
|
||
BIND_INFO_V2_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/bindInfo"
|
||
BIND_API = "https://www.douyu.com/japi/carnivalApi/tencent/bind"
|
||
GOODS_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/storehome"
|
||
EXCHANGE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchange"
|
||
EXCHANGE_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchangeList"
|
||
CREDIT_BALANCE_API = "https://www.douyu.com/japi/oms/web/peace/credit/balance"
|
||
CREDIT_DIFF_API = "https://www.douyu.com/japi/oms/web/peace/credit/diff"
|
||
PEACE_ITEM_API = "https://www.douyu.com/japi/oms/web/peace/item"
|
||
PEACE_TIME_API = "https://www.douyu.com/japi/oms/web/peace/time"
|
||
PAY_QR_API = "https://www.douyu.com/japi/oms/web/pay/getQrCode"
|
||
ESPORTS_BIND_INFO_API = "https://www.douyu.com/wgapi/ordnc/live/memberthird/tencent/bindInfo"
|
||
ESPORTS_BIND_ROLE_PARAM_API = "https://www.douyu.com/wgapi/ordnc/live/memberthird/tencent/bindRoleParam"
|
||
ESPORTS_CSRF_API = "https://www.douyu.com/wgapi/livenc/liveweb/csrfApi/getCsrfCookie"
|
||
ESPORTS_AGREEMENT_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/agreement"
|
||
ESPORTS_ACT_BIND_API = "https://www.douyu.com/wgapi/ordnc/live/memberthird/tencent/actBind"
|
||
ESPORTS_USER_INFO_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/getUserInfo"
|
||
ESPORTS_GOODS_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/storehome"
|
||
ESPORTS_EXCHANGE_API = "https://www.douyu.com/wgapi/ordnc/activity/peacematch/exchange"
|
||
GOLD_QR_API = "https://cz.douyu.com/m/gold/getQrCode"
|
||
GOLD_ACCOUNT_API = "https://cz.douyu.com/item/gold/account"
|
||
EXCHANGE_BALANCE_API = "https://www.douyu.com/wjapi/nc/exchange/fim"
|
||
DONATE_API = "https://www.douyu.com/japi/gift/donate/mainsite/v3"
|
||
|
||
def __init__(
|
||
self,
|
||
cookie: str,
|
||
*,
|
||
logger: Callable[[str], None] | None = None,
|
||
timeout: tuple[float, float] = (8, 20),
|
||
):
|
||
self.cookie = normalize_douyu_cookie(cookie)
|
||
self.timeout = timeout
|
||
self.logger = logger or (lambda _msg: None)
|
||
self.session = requests.Session()
|
||
self.session.trust_env = False
|
||
self.session.headers.update({
|
||
"User-Agent": PC_UA,
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
})
|
||
self._load_cookie(self.cookie)
|
||
|
||
@staticmethod
|
||
def profile_from_cookie(cookie: str) -> dict[str, str]:
|
||
"""从斗鱼 Cookie 提取账号基础信息。"""
|
||
nickname = cookie_value(cookie, "acf_nickname")
|
||
return {
|
||
"uid": cookie_value(cookie, "acf_uid"),
|
||
"nickname": unquote(nickname) if nickname else "",
|
||
}
|
||
|
||
def _load_cookie(self, cookie: str) -> None:
|
||
for key, value in cookie_pairs(cookie):
|
||
self.session.cookies.set(key, value, domain=".douyu.com", path="/")
|
||
|
||
def _merge_response_cookie(self, response: requests.Response) -> None:
|
||
if not response.cookies:
|
||
return
|
||
self.cookie = normalize_cookie_pairs([
|
||
*cookie_pairs(self.cookie),
|
||
*[(key, value) for key, value in response.cookies.items()],
|
||
])
|
||
self._load_cookie(self.cookie)
|
||
|
||
def _set_cookie_value(self, key: str, value: str) -> None:
|
||
"""写入单个 Cookie 值，确保手动 Cookie 请求头与 Session 一致。"""
|
||
self.cookie = normalize_cookie_pairs([
|
||
*cookie_pairs(self.cookie),
|
||
(key, value),
|
||
])
|
||
self._load_cookie(self.cookie)
|
||
|
||
def _headers(self, referer: str = "https://www.douyu.com/") -> dict[str, str]:
|
||
return {
|
||
"User-Agent": PC_UA,
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
"Origin": "https://www.douyu.com",
|
||
"Referer": referer,
|
||
"Cookie": self.cookie,
|
||
}
|
||
|
||
def _request(
|
||
self,
|
||
method: str,
|
||
url: str,
|
||
*,
|
||
source: str,
|
||
headers: dict[str, str] | None = None,
|
||
**kwargs,
|
||
) -> requests.Response:
|
||
req_headers = self._headers()
|
||
if headers:
|
||
req_headers.update(headers)
|
||
response = self.session.request(
|
||
method,
|
||
url,
|
||
headers=req_headers,
|
||
timeout=kwargs.pop("timeout", self.timeout),
|
||
**kwargs,
|
||
)
|
||
self._merge_response_cookie(response)
|
||
self.logger(f"{source}: {response.status_code}")
|
||
response.raise_for_status()
|
||
return response
|
||
|
||
def _json(self, response: requests.Response, source: str) -> dict[str, Any]:
|
||
text = response.text.strip()
|
||
if not text:
|
||
raise DouyuActivityError(f"{source} 返回为空")
|
||
try:
|
||
return response.json()
|
||
except json.JSONDecodeError as exc:
|
||
raise DouyuActivityError(f"{source} 返回不是 JSON: {text[:200]}") from exc
|
||
|
||
def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict[str, Any]:
|
||
return self._json(self._request(method, url, source=source, **kwargs), source)
|
||
|
||
def csrf_token(self) -> str:
|
||
"""获取 cvl_csrf_token。"""
|
||
payload = self._request_json(
|
||
"post",
|
||
self.CSRF_API,
|
||
"生成 CSRF",
|
||
data="{}",
|
||
headers={
|
||
"Content-Type": "application/json;charset=UTF-8",
|
||
"Referer": "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0",
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "生成 CSRF 失败")
|
||
token = cookie_value(self.cookie, "cvl_csrf_token")
|
||
if not token:
|
||
raise DouyuActivityError("响应 Cookie 中没有 cvl_csrf_token")
|
||
return token
|
||
|
||
def acf_ccn(self, *, refresh_subscribe: bool = True) -> str:
|
||
"""获取 acf_ccn；必要时通过订阅接口刷新一次。"""
|
||
response = self._request(
|
||
"get",
|
||
self.ACF_CCN_API,
|
||
source="获取 acf_ccn",
|
||
headers={
|
||
"Origin": "",
|
||
"Referer": "https://www.douyu.com/pages/ord-task-center?clientType=web&panelSource=1&rid=0",
|
||
},
|
||
)
|
||
ctn = cookie_value(self.cookie, "acf_ccn")
|
||
if not ctn:
|
||
response = self._request(
|
||
"get",
|
||
self.ACF_CCN_FALLBACK_API,
|
||
source="获取 acf_ccn 备用接口",
|
||
headers={"Origin": ""},
|
||
)
|
||
ctn = cookie_value(self.cookie, "acf_ccn")
|
||
if not ctn:
|
||
raise DouyuActivityError(f"获取 acf_ccn 失败: {response.text[:200]}")
|
||
|
||
if refresh_subscribe:
|
||
payload = self._request_json(
|
||
"post",
|
||
self.SUBSCRIBE_API,
|
||
"刷新订阅状态",
|
||
data={"subscribe": "cmn", "ctn": ctn},
|
||
headers={
|
||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||
"Referer": "https://www.douyu.com/",
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0"):
|
||
raise DouyuActivityError(payload.get("msg") or "刷新订阅状态失败")
|
||
ctn = cookie_value(self.cookie, "acf_ccn") or ctn
|
||
return ctn
|
||
|
||
def get_bind_qr(self, act_alias: str) -> dict[str, Any]:
|
||
"""生成腾讯游戏绑定二维码链接（v2 接口）。"""
|
||
ctn = self.acf_ccn(refresh_subscribe=True)
|
||
payload = self._request_json(
|
||
"post",
|
||
self.ROLE_PARAM_API,
|
||
"获取绑定二维码",
|
||
params={"actAlias": act_alias},
|
||
headers={
|
||
"Referer": "https://www.douyu.com/",
|
||
},
|
||
)
|
||
url = (payload.get("data") or {}).get("url") or ""
|
||
if not url:
|
||
raise DouyuActivityError(payload.get("msg") or "获取绑定二维码失败")
|
||
return {"url": url, "ctn": ctn, "raw": payload}
|
||
|
||
@staticmethod
|
||
def _first_present(data: dict[str, Any], *keys: str):
|
||
"""按候选 key 取第一个非 None 值。"""
|
||
for key in keys:
|
||
if key in data and data.get(key) is not None:
|
||
return data.get(key)
|
||
return None
|
||
|
||
def bind_info(self, act_alias: str, *, v2: bool = False) -> dict[str, Any]:
|
||
"""查询绑定游戏账号信息。"""
|
||
url = self.BIND_INFO_V2_API if v2 else self.BIND_INFO_API
|
||
payload = self._request_json(
|
||
"get",
|
||
url,
|
||
"查询绑定信息",
|
||
params={"actAlias": act_alias},
|
||
headers={"Origin": "", "Referer": "https://www.douyu.com/"},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "查询绑定信息失败")
|
||
parsed = self._parse_bind_info(payload, act_alias=act_alias, api_version="v2" if v2 else "v1")
|
||
self._log_bind_info(parsed)
|
||
return parsed
|
||
|
||
def _log_bind_info(self, parsed: dict[str, Any]) -> None:
|
||
"""输出绑定信息接口返回的角色明细（用于排查两个接口/多个 alias 的角色差异）。"""
|
||
self.logger(
|
||
f"查询绑定信息明细 api={parsed.get('api_version') or '-'} "
|
||
f"act={parsed.get('act_alias') or '-'} "
|
||
f"role={parsed.get('role_name') or '-'} "
|
||
f"area={parsed.get('area_name') or '-'} plat={parsed.get('plat_name') or '-'} "
|
||
f"nick={parsed.get('nickname') or '-'} "
|
||
f"bound_account={1 if parsed.get('is_bound_account') else 0} "
|
||
f"bound_role={1 if parsed.get('is_bound_role') else 0} "
|
||
f"bound_act={1 if parsed.get('is_bound_act') else 0} "
|
||
f"need_act={1 if parsed.get('need_bind_act') else 0} "
|
||
f"need_role={1 if parsed.get('need_bind_role') else 0} "
|
||
f"wait={parsed.get('change_role_wait_time') if parsed.get('change_role_wait_time') is not None else '-'} "
|
||
f"can_change_role={parsed.get('can_change_role')} "
|
||
f"can_change_time={parsed.get('can_change_time') if parsed.get('can_change_time') is not None else '-'}"
|
||
)
|
||
|
||
def _parse_bind_info(self, payload: dict[str, Any], *, act_alias: str, api_version: str) -> dict[str, Any]:
|
||
"""统一解析不同活动的腾讯游戏绑定信息。"""
|
||
data = payload.get("data") or {}
|
||
game_account = data.get("gameAccount") or {}
|
||
game_role = data.get("gameRole") or {}
|
||
tx_act = data.get("txAct") or {}
|
||
bind_account = data.get("bindAccount")
|
||
bind_role = data.get("bindRole")
|
||
bind_act = data.get("bindAct")
|
||
# 换绑倒计时在不同 alias/版本字段可能不一致，尽量兼容。
|
||
change_role_wait_time = self._first_present(
|
||
data,
|
||
"changeRoleWaitTime",
|
||
"change_role_wait_time",
|
||
"changeBindWaitTime",
|
||
"changBindTime",
|
||
)
|
||
if change_role_wait_time is None and isinstance(tx_act, dict):
|
||
change_role_wait_time = self._first_present(
|
||
tx_act,
|
||
"changeRoleWaitTime",
|
||
"changeBindWaitTime",
|
||
"changBindTime",
|
||
)
|
||
can_change_role = self._first_present(
|
||
data,
|
||
"canChangeRole",
|
||
"can_change_role",
|
||
"canChangeBind",
|
||
)
|
||
if can_change_role is None and isinstance(tx_act, dict):
|
||
can_change_role = self._first_present(tx_act, "canChangeRole", "canChangeBind")
|
||
can_change_time = self._first_present(
|
||
data,
|
||
"canChangeTime",
|
||
"can_change_time",
|
||
)
|
||
if can_change_time is None and isinstance(tx_act, dict):
|
||
can_change_time = self._first_present(tx_act, "canChangeTime", "can_change_time")
|
||
return {
|
||
"act_alias": act_alias,
|
||
"api_version": api_version,
|
||
"bind_account": bind_account,
|
||
"bind_role": bind_role,
|
||
"bind_act": bind_act,
|
||
"need_bind_role": data.get("needBindRole"),
|
||
"need_bind_act": data.get("needBindAct"),
|
||
"is_bound_account": str(bind_account) == "1",
|
||
"is_bound_role": str(bind_role) == "1",
|
||
"is_bound_act": str(bind_act) == "1",
|
||
"nickname": game_account.get("nick") or "",
|
||
"role_name": game_role.get("roleName") or "",
|
||
"area_name": game_role.get("areaName") or "",
|
||
"plat_name": game_role.get("platName") or "",
|
||
"tx_act": tx_act if isinstance(tx_act, dict) else {},
|
||
"change_role_wait_time": change_role_wait_time,
|
||
"can_change_role": can_change_role,
|
||
"can_change_time": can_change_time,
|
||
"raw": payload,
|
||
}
|
||
|
||
def esports_bind_info(self, act_alias: str) -> dict[str, Any]:
|
||
"""电竞手册唯一的角色状态查询，返回角色、绑定状态和换绑冷却。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.ESPORTS_BIND_INFO_API,
|
||
"查询电竞手册绑定信息",
|
||
params={"actAlias": act_alias},
|
||
headers={"Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "查询电竞手册绑定信息失败")
|
||
parsed = self._parse_bind_info(payload, act_alias=act_alias, api_version="esports")
|
||
self._log_bind_info(parsed)
|
||
return parsed
|
||
|
||
def get_esports_bind_qr(self, act_alias: str) -> dict[str, Any]:
|
||
"""生成电竞手册的腾讯角色绑定二维码链接。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.ESPORTS_BIND_ROLE_PARAM_API,
|
||
"获取电竞手册绑定二维码",
|
||
params={"actAlias": act_alias},
|
||
headers={"Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "获取电竞手册绑定二维码失败")
|
||
data = payload.get("data") or {}
|
||
# 电竞手册网页实际打开 data.url；gameRoleBindUrl 是兼容旧入口。
|
||
bind_url = data.get("url") or data.get("gameRoleBindUrl") or ""
|
||
if not bind_url:
|
||
raise DouyuActivityError(payload.get("msg") or "电竞手册绑定二维码链接为空")
|
||
return {
|
||
"url": bind_url,
|
||
"bind_url": bind_url,
|
||
"role_param_url": data.get("url") or "",
|
||
"game_role_bind_url": data.get("gameRoleBindUrl") or "",
|
||
"act_alias": act_alias,
|
||
"raw": payload,
|
||
}
|
||
|
||
def esports_ctn(self, *, referer: str) -> str:
|
||
"""获取电竞手册页面当前 ctn，仅在 Cookie 缺失时走页面 CSRF 接口。"""
|
||
ctn = cookie_value(self.cookie, "acf_ccn")
|
||
if ctn:
|
||
return ctn
|
||
|
||
response = self._request(
|
||
"get",
|
||
self.ESPORTS_CSRF_API,
|
||
source="获取电竞手册 ctn",
|
||
headers={"Origin": "", "Referer": referer},
|
||
)
|
||
ctn = cookie_value(self.cookie, "acf_ccn")
|
||
if not ctn:
|
||
raise DouyuActivityError(f"获取电竞手册 ctn 失败: {response.text[:200]}")
|
||
return ctn
|
||
|
||
def confirm_esports_bind(self, act_alias: str, *, room_id: str = "9263298") -> dict[str, Any]:
|
||
"""自动完成协议前置请求后，通过 actBind 确认电竞手册绑定。"""
|
||
referer = (
|
||
"https://www.douyu.com/pages/live-peace-pel/web"
|
||
f"?roomId={room_id or '9263298'}&isAnchorSide=0"
|
||
)
|
||
|
||
# 页面成功流程固定为 agreement -> actBind；协议不作为独立业务状态暴露。
|
||
# 两次 POST 都使用当前 Cookie 中的 acf_ccn，不额外主动刷新 CSRF。
|
||
agreement_ctn = self.esports_ctn(referer=referer)
|
||
agreement = self._request_json(
|
||
"post",
|
||
self.ESPORTS_AGREEMENT_API,
|
||
"同意电竞手册协议",
|
||
files={"ctn": (None, agreement_ctn)},
|
||
headers={"Referer": referer},
|
||
)
|
||
if agreement.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(agreement.get("msg") or "同意电竞手册协议失败")
|
||
|
||
bind_ctn = self.esports_ctn(referer=referer)
|
||
bind = self._request_json(
|
||
"post",
|
||
self.ESPORTS_ACT_BIND_API,
|
||
"完成电竞手册绑定",
|
||
files={
|
||
"actAlias": (None, act_alias),
|
||
"ctn": (None, bind_ctn),
|
||
},
|
||
headers={"Referer": referer},
|
||
)
|
||
if bind.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(bind.get("msg") or "完成电竞手册绑定失败")
|
||
return {"agreement": agreement, "act_bind": bind}
|
||
|
||
def confirm_bind(self, act_alias: str) -> dict[str, Any]:
|
||
"""确认绑定当前扫码选择的游戏角色。"""
|
||
token = self.csrf_token()
|
||
payload = self._request_json(
|
||
"post",
|
||
self.BIND_API,
|
||
"确认绑定",
|
||
data={"actAlias": act_alias, "csrfToken": token},
|
||
headers={
|
||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||
"Referer": "https://www.douyu.com/",
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0"):
|
||
raise DouyuActivityError(payload.get("msg") or "确认绑定失败")
|
||
return {"csrf_token": token, "raw": payload}
|
||
|
||
def create_elite_qr(self, *, ctn: str, act_alias: str, amount: int, room_id: str) -> dict[str, Any]:
|
||
"""生成精英宝典支付二维码。"""
|
||
# E 语言实现会在 peace/item 和 getQrCode 两次请求里显式拼上 acf_ccn=ctn。
|
||
# 这里也同步 Cookie，避免接口只看 Cookie 时误判未绑定游戏账号。
|
||
self._set_cookie_value("acf_ccn", ctn)
|
||
item_payload = self._request_json(
|
||
"post",
|
||
self.PEACE_ITEM_API,
|
||
"校验精英宝典",
|
||
json={"itemId": 1, "ctn": ctn},
|
||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||
)
|
||
if item_payload.get("error") not in (0, "0"):
|
||
raise DouyuActivityError(item_payload.get("msg") or "精英宝典校验失败")
|
||
|
||
payload = self._request_json(
|
||
"post",
|
||
self.PAY_QR_API,
|
||
"生成精英宝典支付码",
|
||
json={
|
||
"bizType": "peace",
|
||
"biz": {"actAlias": act_alias, "amount": amount, "roomId": int(room_id or 0)},
|
||
"ctn": ctn,
|
||
},
|
||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||
)
|
||
qr = (payload.get("data") or {}).get("qr") or ""
|
||
if not qr:
|
||
raise DouyuActivityError(payload.get("msg") or "生成精英宝典支付码失败")
|
||
return {"pay_url": qr, "ctn": ctn, "act_alias": act_alias, "item": item_payload, "raw": payload}
|
||
|
||
def create_esports_qr(self, *, ctn: str, act_alias: str, amount: int, room_id: str) -> dict[str, Any]:
|
||
"""生成电竞手册支付二维码。"""
|
||
self._set_cookie_value("acf_ccn", ctn)
|
||
time_payload = self._request_json(
|
||
"get",
|
||
self.PEACE_TIME_API,
|
||
"校验电竞手册开通时间",
|
||
headers={"Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
|
||
)
|
||
if time_payload.get("error") not in (0, "0"):
|
||
raise DouyuActivityError(time_payload.get("msg") or "电竞手册当前不可开通")
|
||
|
||
payload = self._request_json(
|
||
"post",
|
||
self.PAY_QR_API,
|
||
"生成电竞手册支付码",
|
||
json={
|
||
"bizType": "peacePel",
|
||
"biz": {"actAlias": act_alias, "amount": amount, "roomId": int(room_id or 0)},
|
||
"ctn": ctn,
|
||
},
|
||
headers={
|
||
"Content-Type": "application/json;charset=UTF-8",
|
||
"Referer": "https://www.douyu.com/pages/live-peace-pel/web/",
|
||
},
|
||
)
|
||
data = payload.get("data") or {}
|
||
qr = data.get("qr") or ""
|
||
if not qr:
|
||
raise DouyuActivityError(payload.get("msg") or "生成电竞手册支付码失败")
|
||
return {
|
||
"pay_url": qr,
|
||
"pay_token": data.get("token") or "",
|
||
"pay_ttl": data.get("ttl"),
|
||
"ctn": ctn,
|
||
"act_alias": act_alias,
|
||
"time": time_payload,
|
||
"raw": payload,
|
||
}
|
||
|
||
def esports_user_info(self, *, manual_id: str) -> dict[str, Any]:
|
||
"""查询电竞手册的开通状态和积分。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.ESPORTS_USER_INFO_API,
|
||
"查询电竞手册状态",
|
||
params={"manualID": manual_id},
|
||
headers={"Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-pel/web/"},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "查询电竞手册状态失败")
|
||
user_info = (payload.get("data") or {}).get("userInfo") or {}
|
||
return {
|
||
"manual_id": manual_id,
|
||
"manual_type": user_info.get("manualType"),
|
||
"manual_score": user_info.get("manualScore"),
|
||
"expire_time": user_info.get("expireTime"),
|
||
"user_info": user_info,
|
||
"raw": payload,
|
||
}
|
||
|
||
@staticmethod
|
||
def _esports_shop_referer(rid: str) -> str:
|
||
return (
|
||
"https://www.douyu.com/pages/live-peace-pel/web/shop"
|
||
f"?roomId={rid or '0'}&isAnchorSide=0"
|
||
)
|
||
|
||
def list_esports_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
|
||
"""查询电竞手册皮肤商城。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.ESPORTS_GOODS_API,
|
||
"刷新电竞皮肤商城",
|
||
params={"manualID": manual_id, "rid": rid},
|
||
headers={
|
||
"Origin": "",
|
||
"Referer": self._esports_shop_referer(rid),
|
||
},
|
||
)
|
||
goods = (payload.get("data") or {}).get("list") or []
|
||
return {
|
||
"goods": goods,
|
||
"score": (payload.get("data") or {}).get("score"),
|
||
"raw": payload,
|
||
}
|
||
|
||
def exchange_esports_goods(
|
||
self,
|
||
*,
|
||
manual_id: str,
|
||
rid: str,
|
||
commodity_id: str,
|
||
quantity: int = 1,
|
||
ctn: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""兑换电竞手册皮肤商城商品。"""
|
||
referer = self._esports_shop_referer(rid)
|
||
ctn_value = ctn or self.esports_ctn(referer=referer)
|
||
self._set_cookie_value("acf_ccn", ctn_value)
|
||
payload = self._request_json(
|
||
"post",
|
||
self.ESPORTS_EXCHANGE_API,
|
||
"兑换电竞皮肤",
|
||
data={
|
||
"manualID": manual_id,
|
||
"commodityID": commodity_id,
|
||
"rid": rid or "0",
|
||
"num": str(max(1, quantity)),
|
||
"ctn": ctn_value,
|
||
},
|
||
headers={
|
||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||
"Origin": "",
|
||
"Referer": referer,
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0"):
|
||
raise DouyuActivityError(payload.get("msg") or "兑换电竞皮肤失败")
|
||
return {
|
||
"commodity_id": commodity_id,
|
||
"quantity": max(1, quantity),
|
||
"ctn": ctn_value,
|
||
"raw": payload,
|
||
}
|
||
|
||
def create_gold_qr(self, *, amount: int, pay_type: int = 1, product_id: str = "DYTV_Product_Gold_1") -> dict[str, Any]:
|
||
"""生成鱼翅充值支付二维码。"""
|
||
payload = self._request_json(
|
||
"post",
|
||
self.GOLD_QR_API,
|
||
"生成鱼翅充值码",
|
||
data={
|
||
"pay_type": str(pay_type),
|
||
"product_id": product_id,
|
||
"number": str(amount),
|
||
"toname": "",
|
||
"source": "dy_box",
|
||
},
|
||
headers={
|
||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||
"Origin": "https://cz.douyu.com",
|
||
"Referer": "https://cz.douyu.com/",
|
||
},
|
||
)
|
||
data = payload.get("data") or {}
|
||
pay_url = data.get("code_url") or payload.get("code_url") or ""
|
||
token = data.get("token") or payload.get("token") or ""
|
||
if not pay_url:
|
||
raise DouyuActivityError(payload.get("msg") or "生成鱼翅充值码失败")
|
||
return {"pay_url": pay_url, "token": token, "raw": payload}
|
||
|
||
def _donate_gift(
|
||
self,
|
||
*,
|
||
gift_count: int,
|
||
room_id: str,
|
||
gift_id: str,
|
||
skin_id: str,
|
||
source: str,
|
||
) -> dict[str, Any]:
|
||
"""赠送直播间礼物。"""
|
||
payload = self._request_json(
|
||
"post",
|
||
self.DONATE_API,
|
||
source,
|
||
data={
|
||
"giftId": gift_id,
|
||
"giftCount": str(gift_count),
|
||
"roomId": room_id,
|
||
"bizExt": json.dumps({"isMainGift": 1}, separators=(",", ":")),
|
||
"skinId": skin_id,
|
||
},
|
||
headers={
|
||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||
"Referer": f"https://www.douyu.com/{room_id}",
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0") and payload.get("msg") != "success":
|
||
raise DouyuActivityError(payload.get("msg") or f"{source}失败")
|
||
return {"raw": payload}
|
||
|
||
def donate_elite_gift(
|
||
self,
|
||
*,
|
||
gift_count: int,
|
||
room_id: str,
|
||
gift_id: str,
|
||
skin_id: str,
|
||
) -> dict[str, Any]:
|
||
"""赠送精英令礼物。"""
|
||
return self._donate_gift(
|
||
gift_count=gift_count,
|
||
room_id=room_id,
|
||
gift_id=gift_id,
|
||
skin_id=skin_id,
|
||
source="赠送精英令",
|
||
)
|
||
|
||
def donate_esports_gift(
|
||
self,
|
||
*,
|
||
gift_name: str,
|
||
gift_count: int,
|
||
room_id: str,
|
||
gift_id: str,
|
||
skin_id: str,
|
||
) -> dict[str, Any]:
|
||
"""赠送电竞手册任务指定礼物。"""
|
||
return self._donate_gift(
|
||
gift_count=gift_count,
|
||
room_id=room_id,
|
||
gift_id=gift_id,
|
||
skin_id=skin_id,
|
||
source=f"赠送{gift_name}",
|
||
)
|
||
|
||
def query_points(self, *, uid: str, ctn: str) -> dict[str, Any]:
|
||
"""查询手册积分余额。"""
|
||
payload = self._request_json(
|
||
"post",
|
||
self.CREDIT_BALANCE_API,
|
||
"查询手册积分",
|
||
json={"uid": str(uid), "ctn": ctn},
|
||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||
)
|
||
data = payload.get("data") or {}
|
||
return {"points": data.get("balance"), "raw": payload}
|
||
|
||
def query_points_diff(self, *, uid: str, ctn: str) -> dict[str, Any]:
|
||
"""查询手册积分差额。"""
|
||
payload = self._request_json(
|
||
"post",
|
||
self.CREDIT_DIFF_API,
|
||
"查询手册积分差额",
|
||
json={"userId": str(uid), "ctn": ctn},
|
||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||
)
|
||
data = payload.get("data") or {}
|
||
return {"diff": data.get("diff"), "raw": payload}
|
||
|
||
def list_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
|
||
"""刷新活动商品列表。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.GOODS_API,
|
||
"刷新商品列表",
|
||
params={"manualID": manual_id, "rid": rid},
|
||
headers={
|
||
"Origin": "",
|
||
"Referer": f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}",
|
||
},
|
||
)
|
||
goods = (payload.get("data") or {}).get("list") or []
|
||
return {"goods": goods, "raw": payload}
|
||
|
||
def query_limited_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
|
||
"""查询当前账号限兑商品。"""
|
||
result = self.list_goods(manual_id=manual_id, rid=rid)
|
||
limited = [
|
||
item for item in result["goods"]
|
||
if str(item.get("status") or "") == "1"
|
||
]
|
||
return {"limited_goods": limited, "raw": result["raw"]}
|
||
|
||
def exchange_records(self, *, manual_id: str, limit: int = 100, offset: int = 0) -> dict[str, Any]:
|
||
"""查询兑换记录。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.EXCHANGE_LIST_API,
|
||
"查询兑换记录",
|
||
params={"manualID": manual_id, "limit": limit, "offset": offset},
|
||
headers={"Origin": "", "Referer": "https://www.douyu.com/"},
|
||
)
|
||
records = (payload.get("data") or {}).get("list") or []
|
||
return {"records": records, "raw": payload}
|
||
|
||
def exchange_goods(self, *, manual_id: str, rid: str, commodity_id: str, ctn: str | None = None) -> dict[str, Any]:
|
||
"""兑换活动商品。"""
|
||
ctn_value = ctn or cookie_value(self.cookie, "acf_ccn") or self.acf_ccn(refresh_subscribe=False)
|
||
token = self.csrf_token()
|
||
randstr = "".join(random.choices(string.ascii_letters + string.digits, k=16))
|
||
data = {
|
||
"manualID": manual_id,
|
||
"rid": rid,
|
||
"commodityID": commodity_id,
|
||
"ctn": ctn_value,
|
||
"csrfToken": token,
|
||
"randstr": randstr,
|
||
"token[error][code]": "-1",
|
||
"token[data]": "",
|
||
}
|
||
payload = self._request_json(
|
||
"post",
|
||
self.EXCHANGE_API,
|
||
"兑换商品",
|
||
data=data,
|
||
headers={
|
||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||
"Referer": f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}",
|
||
"sec-ch-ua": f'"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
|
||
"sec-ch-ua-mobile": "?0",
|
||
"sec-ch-ua-platform": '"Windows"',
|
||
"Sec-Fetch-Site": "same-origin",
|
||
"Sec-Fetch-Mode": "cors",
|
||
"Sec-Fetch-Dest": "empty",
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0"):
|
||
raise DouyuActivityError(payload.get("msg") or "兑换商品失败")
|
||
return {
|
||
"commodity_id": commodity_id,
|
||
"ctn": ctn_value,
|
||
"csrf_token": token,
|
||
"randstr": randstr,
|
||
"raw": payload,
|
||
}
|
||
|
||
def gold_account(self) -> dict[str, Any]:
|
||
"""查询 cz 侧鱼翅余额。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.GOLD_ACCOUNT_API,
|
||
"查询鱼翅余额",
|
||
headers={"Origin": "https://cz.douyu.com", "Referer": "https://cz.douyu.com/"},
|
||
)
|
||
data = payload.get("data") or {}
|
||
return {"gold": data.get("gold"), "raw": payload}
|
||
|
||
def exchange_balance(self) -> dict[str, Any]:
|
||
"""查询钱包兑换中心余额。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.EXCHANGE_BALANCE_API,
|
||
"查询钱包兑换余额",
|
||
params={"appCode": "YJTX"},
|
||
headers={
|
||
"Origin": "",
|
||
"Referer": "https://www.douyu.com/member/walletcenter",
|
||
},
|
||
)
|
||
data = payload.get("data") or {}
|
||
return {"count": data.get("count"), "raw": payload}
|
||
|
||
# ---- 和平小店（腾讯道聚城 H5）----
|
||
XPD_EMBED_API = "https://www.douyu.com/japi/carnival/nc/txEmbed/getIframeUrl"
|
||
XPD_H5_REFERER = "https://www.douyu.com/topic/h5/hpxd01"
|
||
XPD_ROLE_PARAM_API = "https://apiv2.douyucdn.cn/japi/carnivalApi/tencent/roleParam"
|
||
XPD_BIND_INFO_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/bindInfo"
|
||
XPD_DAOJU_REFERER = "https://app.daoju.qq.com/"
|
||
XPD_GET_ROLE_API = "https://apps.game.qq.com/daoju/igw/live"
|
||
XPD_ORDER_API = "https://apps.game.qq.com/daoju/igw/live/"
|
||
# 2026-08 抓包(8.6-和平小店)实证:本期走 xn_live_cjm 变体(旧期为 recommend_live/common)
|
||
XPD_RECOMMEND_API = "https://apps.game.qq.com/daoju/v3/recommend_xn_live_cjm/common"
|
||
XPD_BALANCE_API = "https://apps.game.qq.com/daoju/igw/live/"
|
||
XPD_LIVE_BUY_API = "https://apps.game.qq.com/daoju/igw/livebuy/"
|
||
XPD_GOODS_PAGE_SIZE = 10
|
||
XPD_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"
|
||
)
|
||
|
||
@staticmethod
|
||
def _xpd_int(value: Any) -> int | None:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _xpd_parse_var(text: str, varname: str) -> dict[str, Any]:
|
||
"""解析道聚城 `var xxx={...};` 形式的响应（容忍等号两侧空白）。"""
|
||
match = re.search(rf"var\s+{varname}\s*=", text)
|
||
if not match:
|
||
raise DouyuActivityError(f"响应中找不到 var {varname}=: {text[:200]}")
|
||
chunk = text[match.end():].strip()
|
||
if chunk.endswith(";"):
|
||
chunk = chunk[:-1]
|
||
try:
|
||
return json.loads(chunk)
|
||
except json.JSONDecodeError as exc:
|
||
raise DouyuActivityError(f"var {varname}= 解析失败: {chunk[:200]}") from exc
|
||
|
||
@classmethod
|
||
def _xpd_daoju_headers(cls) -> dict[str, str]:
|
||
return {
|
||
"User-Agent": cls.XPD_UA,
|
||
"Accept": "*/*",
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
"X-Requested-With": "air.tv.douyu.android",
|
||
"Sec-Fetch-Site": "same-site",
|
||
"Sec-Fetch-Mode": "no-cors",
|
||
"Sec-Fetch-Dest": "script",
|
||
"Referer": cls.XPD_DAOJU_REFERER,
|
||
}
|
||
|
||
def _xpd_token(self) -> str:
|
||
"""拼斗鱼 token: acf_uid_1_acf_stk_0_acf_ltkid。"""
|
||
uid = cookie_value(self.cookie, "acf_uid")
|
||
stk = cookie_value(self.cookie, "acf_stk")
|
||
ltkid = cookie_value(self.cookie, "acf_ltkid")
|
||
if not uid or not stk or not ltkid:
|
||
raise DouyuActivityError("Cookie 缺少 acf_uid/acf_stk/acf_ltkid，无法获取小店参数")
|
||
return f"{uid}_1_{stk}_0_{ltkid}"
|
||
|
||
def xpd_embed_query(self, *, act_alias: str, rid: str) -> dict[str, Any]:
|
||
"""从斗鱼换取和平小店 H5 参数（code/sig/timestamp/actId 等，每次现签）。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.XPD_EMBED_API,
|
||
"获取小店 H5 参数",
|
||
params={"actAlias": act_alias, "rid": rid, "token": self._xpd_token()},
|
||
headers={"Referer": self.XPD_H5_REFERER},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "获取小店 H5 参数失败")
|
||
data = payload.get("data") or {}
|
||
txurl_h5 = str(data.get("txurlH5") or "")
|
||
if not txurl_h5:
|
||
raise DouyuActivityError(f"获取小店 H5 参数失败: {payload}")
|
||
query = parse_qs(urlsplit(txurl_h5).query)
|
||
embed_query = {key: values[0] for key, values in query.items() if values}
|
||
return {"query": embed_query, "txurl_h5": txurl_h5, "raw": payload}
|
||
|
||
def xpd_get_role(
|
||
self,
|
||
*,
|
||
embed_query: dict[str, str],
|
||
act_id: str,
|
||
rid: str = "",
|
||
) -> dict[str, Any]:
|
||
"""用小店 code/sig 换取当前绑定角色信息。
|
||
|
||
2026-08 起道聚城校验升级：必须带 isCode=1 + authType=delegate + sAnchorId(房间号)。
|
||
"""
|
||
params = {
|
||
"acctype": "livelink",
|
||
"_jsvar": "info",
|
||
"_service": "other.livelink.getrole.out",
|
||
"_biz_code": "cjm",
|
||
"_act_id": act_id,
|
||
"_app_id": "2123",
|
||
"isCode": "1",
|
||
"gameId": embed_query.get("gameId") or "cjm",
|
||
"actId": embed_query.get("actId") or "",
|
||
"appId": embed_query.get("appId") or "bp_cf",
|
||
"livePlatId": embed_query.get("livePlatId") or "douyu",
|
||
"code": embed_query.get("code") or "",
|
||
"timestamp": embed_query.get("timestamp") or "",
|
||
"v": embed_query.get("v") or "",
|
||
"sig": embed_query.get("sig") or "",
|
||
"authType": "delegate",
|
||
"sAnchorId": rid,
|
||
"sVideoId": "",
|
||
}
|
||
response = self._request(
|
||
"get",
|
||
self.XPD_GET_ROLE_API,
|
||
source="查询小店绑定角色",
|
||
params=params,
|
||
headers=self._xpd_daoju_headers(),
|
||
)
|
||
info = self._xpd_parse_var(response.text, "info")
|
||
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 ""),
|
||
"plat_id": str(info.get("platId") or ""),
|
||
"area": str(info.get("area") or ""),
|
||
"raw": info,
|
||
}
|
||
|
||
def xpd_list_goods(
|
||
self,
|
||
*,
|
||
embed_query: dict[str, str],
|
||
act_id: str,
|
||
openid: str,
|
||
roleid: str,
|
||
areaid: str = "1",
|
||
) -> dict[str, Any]:
|
||
"""查询和平小店全部商品，排除热门抢购分类并拉取全部分页。"""
|
||
base_params = {
|
||
"_service": "act.recommend.query",
|
||
"_biz_code": "cjm",
|
||
"_app_id": "2123",
|
||
"app_id": "2123",
|
||
"fields": (
|
||
"iRelationGoodsid,dtModifyTime,sBuyLimitInfo,sGoodsWaterMark,dtShowBeginTime,"
|
||
"dtShowEndTime,iSort,iGoodsId,sGoodsName,iJbPrice,iJbOrgPrice,iJb2Price,"
|
||
"iJb2OrgPrice,iPrice,iOrgPrice,sGoodsPic,sGoodsDesc,sPayType,dtBeginTime,"
|
||
"dtEndTime,iActionId,sExtShowInfo,sExtInfo,iCategoryId,dtRushBegin,dtRushEnd"
|
||
),
|
||
"actionFields": "iActionId,sActionName,dtBeginTime,dtEndTime",
|
||
"order_by": "dtShowBeginTime",
|
||
"orderColumnBatch": '[{"key":"iSort","sort":"desc"},{"key":"dtShowBeginTime","sort":"desc"},{"key":"iGoodsId","sort":"asc"}]',
|
||
"desc": "1",
|
||
"_act_id": act_id,
|
||
"actid": act_id,
|
||
# category=76 是“热门抢购”；全部商品页明确将其排除。
|
||
"excludeFields": "iCategoryId_76",
|
||
"userid": openid,
|
||
"roleid": roleid,
|
||
"area": "1",
|
||
"areaid": str(areaid),
|
||
"need_limit": "1",
|
||
}
|
||
goods: list[dict[str, Any]] = []
|
||
seen_ids: set[str] = set()
|
||
pages: list[dict[str, Any]] = []
|
||
offset = 0
|
||
page_num = 1
|
||
|
||
while True:
|
||
params = {
|
||
**base_params,
|
||
"page_begin": str(offset),
|
||
"page_num": str(page_num),
|
||
"page_size": str(self.XPD_GOODS_PAGE_SIZE),
|
||
}
|
||
response = self._request(
|
||
"get",
|
||
self.XPD_RECOMMEND_API,
|
||
source=f"刷新小店全部商品第 {page_num} 页",
|
||
params=params,
|
||
headers=self._xpd_daoju_headers(),
|
||
)
|
||
recommend = self._xpd_parse_var(response.text, "recommend")
|
||
if str(recommend.get("errcode") or "0") != "0":
|
||
raise DouyuActivityError(str(recommend.get("errdesc") or "小店商品查询失败"))
|
||
pages.append(recommend)
|
||
items = (((recommend.get("data") or {}).get("client_data") or {}).get("itemsdetail")) or []
|
||
if not isinstance(items, list):
|
||
raise DouyuActivityError("小店商品响应格式异常")
|
||
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
commodity_id = str(item.get("iGoodsId") or "")
|
||
if not commodity_id or commodity_id in seen_ids:
|
||
continue
|
||
seen_ids.add(commodity_id)
|
||
goods.append(
|
||
{
|
||
"commodity_id": commodity_id,
|
||
"name": str(item.get("sGoodsName") or ""),
|
||
"price": self._xpd_int(item.get("iPrice")),
|
||
"org_price": self._xpd_int(item.get("iOrgPrice")),
|
||
"category": str(item.get("iCategoryId") or ""),
|
||
"goods_left": self._xpd_int(item.get("iGoodsLeft")),
|
||
"raw": item,
|
||
}
|
||
)
|
||
|
||
if len(items) < self.XPD_GOODS_PAGE_SIZE:
|
||
break
|
||
offset += self.XPD_GOODS_PAGE_SIZE
|
||
page_num += 1
|
||
if page_num > 100:
|
||
raise DouyuActivityError("小店商品分页超出安全上限")
|
||
|
||
return {"goods": goods, "raw": {"pages": pages}}
|
||
|
||
def xpd_purchase_records(
|
||
self,
|
||
*,
|
||
embed_query: dict[str, str],
|
||
act_id: str,
|
||
page_size: int = 10,
|
||
max_pages: int = 100,
|
||
) -> dict[str, Any]:
|
||
"""查询和平小店购买记录，接口参数来自道聚城 H5 抓包。"""
|
||
page_size = max(1, min(int(page_size), 50))
|
||
records: list[dict[str, Any]] = []
|
||
pages: list[dict[str, Any]] = []
|
||
total = 0
|
||
|
||
for page in range(1, max_pages + 1):
|
||
params = {
|
||
"_service": "order.list",
|
||
"_jsvar": "info",
|
||
"acctype": "livelink",
|
||
"_app_id": "2123",
|
||
"hist": "0",
|
||
"ps": str(page_size),
|
||
"pn": str(page),
|
||
"_biz_code": "cjm",
|
||
"_act_id": str(act_id),
|
||
"authType": "delegate",
|
||
"actId": embed_query.get("actId") or "",
|
||
"appId": embed_query.get("appId") or "bp_cf",
|
||
"gameId": embed_query.get("gameId") or "cjm",
|
||
"livePlatId": embed_query.get("livePlatId") or "douyu",
|
||
"code": embed_query.get("code") or "",
|
||
"timestamp": embed_query.get("timestamp") or "",
|
||
"v": embed_query.get("v") or "",
|
||
"sig": embed_query.get("sig") or "",
|
||
"_sid": "8",
|
||
"set": "8",
|
||
}
|
||
response = self._request(
|
||
"get",
|
||
self.XPD_ORDER_API,
|
||
source=f"查询小店购买记录第 {page} 页",
|
||
params=params,
|
||
headers=self._xpd_daoju_headers(),
|
||
)
|
||
info = self._xpd_parse_var(response.text, "info")
|
||
if str(info.get("ret") or "0") != "0":
|
||
raise DouyuActivityError(str(info.get("msg") or "小店购买记录查询失败"))
|
||
data = info.get("data") if isinstance(info.get("data"), dict) else {}
|
||
page_records = data.get("list") if isinstance(data.get("list"), list) else []
|
||
total = self._xpd_int(data.get("cnt")) or total
|
||
pages.append(info)
|
||
|
||
for record in page_records:
|
||
if not isinstance(record, dict):
|
||
continue
|
||
goods_info = record.get("sGoodsInfo")
|
||
if isinstance(goods_info, str):
|
||
try:
|
||
goods_info = json.loads(goods_info)
|
||
except json.JSONDecodeError:
|
||
goods_info = {}
|
||
goods_list = goods_info.get("list") if isinstance(goods_info, dict) else []
|
||
goods = (
|
||
goods_list[0]
|
||
if isinstance(goods_list, list) and goods_list and isinstance(goods_list[0], dict)
|
||
else {}
|
||
)
|
||
app_ext = record.get("sAppExt")
|
||
if isinstance(app_ext, str):
|
||
try:
|
||
app_ext = json.loads(app_ext)
|
||
except json.JSONDecodeError:
|
||
app_ext = {}
|
||
records.append({
|
||
"serial": str(record.get("sSerialNum") or ""),
|
||
"goods_name": str(
|
||
goods.get("sGoodsName")
|
||
or goods.get("sPacketName")
|
||
or record.get("sActionDesc")
|
||
or ""
|
||
),
|
||
"goods_image": str(goods.get("sGoodsPic") or (app_ext or {}).get("sGoodsPic") or ""),
|
||
"buy_time": str(record.get("dtBuyTime") or ""),
|
||
"pay_time": str(record.get("dtPayTime") or ""),
|
||
"status": str(record.get("iStatus") or ""),
|
||
"price": self._xpd_int(record.get("iPrice")),
|
||
"pay_amount": self._xpd_int(record.get("iPayAmount")),
|
||
"pay_type": str(record.get("sPayType") or ""),
|
||
"role_name": str(record.get("sRoleName") or ""),
|
||
"send_status": str(goods.get("iSendStatus") or ""),
|
||
"delivery_message": str(goods.get("deliveryMsg") or ""),
|
||
"raw": record,
|
||
})
|
||
|
||
if len(page_records) < page_size or len(records) >= total:
|
||
break
|
||
|
||
return {"records": records, "total": total or len(records), "raw": {"pages": pages}}
|
||
|
||
def xpd_bind_qr(self, *, act_alias: str) -> dict[str, Any]:
|
||
"""生成和平小店绑定二维码链接（微信扫码进入 livelink 小程序绑定角色）。
|
||
|
||
返回 data.url（可生成二维码）与 gameRoleBindUrl（直接角色绑定页）。
|
||
"""
|
||
payload = self._request_json(
|
||
"post",
|
||
self.XPD_ROLE_PARAM_API,
|
||
"生成小店绑定二维码",
|
||
params={"client_sys": "android"},
|
||
data={
|
||
"token": self._xpd_token(),
|
||
"uid": cookie_value(self.cookie, "acf_uid") or "",
|
||
"client": "mobile",
|
||
"actAlias": act_alias,
|
||
},
|
||
headers={
|
||
"User-Device": "MDExZTUxMzEyYmYxMmY5MmQ3YmY1M2M4MDA3MDcxMTF8djcuOS4y",
|
||
"aid": "android1",
|
||
"User-Agent": "android/7.9.2 (android 9; ; M973Q)",
|
||
"Accept-Language": "zh-CN",
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "生成小店绑定二维码失败")
|
||
data = payload.get("data") or {}
|
||
url = str(data.get("url") or "")
|
||
if not url:
|
||
raise DouyuActivityError(f"生成小店绑定二维码失败: {payload}")
|
||
return {
|
||
"url": url,
|
||
"game_role_bind_url": str(data.get("gameRoleBindUrl") or ""),
|
||
"nick_name": str((data.get("queryParams") or {}).get("nickName") or ""),
|
||
"raw": payload,
|
||
}
|
||
|
||
def xpd_bind_info(self, *, act_alias: str) -> dict[str, Any]:
|
||
"""查询和平小店绑定信息（bindInfo）。"""
|
||
payload = self._request_json(
|
||
"get",
|
||
self.XPD_BIND_INFO_API,
|
||
"查询小店绑定信息",
|
||
params={"actAlias": act_alias, "token": self._xpd_token()},
|
||
)
|
||
if payload.get("error") not in (0, "0", None):
|
||
raise DouyuActivityError(payload.get("msg") or "查询小店绑定信息失败")
|
||
data = payload.get("data") or {}
|
||
game_role = data.get("gameRole") if isinstance(data.get("gameRole"), dict) else {}
|
||
game_account = data.get("gameAccount") if isinstance(data.get("gameAccount"), dict) else {}
|
||
return {
|
||
"bind_account": self._xpd_int(data.get("bindAccount")),
|
||
"bind_role": self._xpd_int(data.get("bindRole")),
|
||
"need_bind_role": self._xpd_int(data.get("needBindRole")),
|
||
"nick": str(game_account.get("nick") or ""),
|
||
"role_name": str(game_role.get("roleName") or ""),
|
||
"area_name": str(game_role.get("areaName") or ""),
|
||
"plat_name": str(game_role.get("platName") or ""),
|
||
"raw": payload,
|
||
}
|
||
|
||
def xpd_balance(
|
||
self,
|
||
*,
|
||
embed_query: dict[str, str],
|
||
act_id: str,
|
||
openid: str,
|
||
roleid: str,
|
||
plat: str,
|
||
areaid: str = "1",
|
||
) -> dict[str, Any]:
|
||
"""查询和平小店角色点券余额。"""
|
||
params = {
|
||
"_jsvar": "banlanceInfo",
|
||
"_service": "pay.midas.dq.get.ttpp",
|
||
"_app_id": "2123",
|
||
"acctype": "ttpp",
|
||
"areaid": str(areaid),
|
||
"eventid": "",
|
||
"interwork": "0",
|
||
"openid": openid,
|
||
"openkey": "openkey",
|
||
"partition": "0",
|
||
"pay_token": "",
|
||
"plat": str(plat),
|
||
"plat_pc": "0",
|
||
"roleid": roleid,
|
||
"_biz_code": "cjm",
|
||
"_act_id": act_id,
|
||
"_time": str(int(time.time())),
|
||
"_sid": "6",
|
||
}
|
||
response = self._request(
|
||
"get",
|
||
self.XPD_BALANCE_API,
|
||
source="查询小店点券余额",
|
||
params=params,
|
||
headers=self._xpd_daoju_headers(),
|
||
)
|
||
info = self._xpd_parse_var(response.text, "banlanceInfo")
|
||
return {"balance": self._xpd_int(info.get("balance")), "raw": info}
|
||
|
||
def xpd_fragments(
|
||
self,
|
||
*,
|
||
embed_query: dict[str, str],
|
||
act_id: str,
|
||
openid: str,
|
||
roleid: str,
|
||
plat: str,
|
||
areaid: str = "1",
|
||
) -> dict[str, Any]:
|
||
"""查询和平小店角色扭蛋碎片数量（gamecoin jb2）。"""
|
||
params = {
|
||
"_jsvar": "jbInfos",
|
||
"_service": "pay.idip.gamecoin.get",
|
||
"_app_id": "2123",
|
||
"acctype": "ttpp",
|
||
"areaid": str(areaid),
|
||
"eventid": "",
|
||
"interwork": "0",
|
||
"openid": openid,
|
||
"openkey": "openkey",
|
||
"partition": "0",
|
||
"pay_token": "",
|
||
"plat": str(plat),
|
||
"plat_pc": "0",
|
||
"roleid": roleid,
|
||
"_biz_code": "cjm",
|
||
"_act_id": act_id,
|
||
"coin_type": "1",
|
||
"_sid": "6",
|
||
}
|
||
response = self._request(
|
||
"get",
|
||
self.XPD_BALANCE_API,
|
||
source="查询小店扭蛋碎片",
|
||
params=params,
|
||
headers=self._xpd_daoju_headers(),
|
||
)
|
||
info = self._xpd_parse_var(response.text, "jbInfos")
|
||
return {"fragments": self._xpd_int(info.get("jb2")), "raw": info}
|
||
|
||
def xpd_exchange_goods(
|
||
self,
|
||
*,
|
||
embed_query: dict[str, str],
|
||
act_id: str,
|
||
rid: str,
|
||
commodity_id: str,
|
||
price: int,
|
||
picture: str = "",
|
||
pay_type: int = 1,
|
||
action_id: str = "",
|
||
) -> dict[str, Any]:
|
||
"""兑换和平小店商品。
|
||
|
||
道聚城的 code/sig 为短时授权参数，调用方必须在每次兑换前重新获取。
|
||
该接口会实际扣除点券或扭蛋碎片，因此遇到网络错误不做自动重试。
|
||
"""
|
||
if pay_type not in (1, 5):
|
||
raise DouyuActivityError("小店兑换仅支持点券或扭蛋碎片")
|
||
if not commodity_id:
|
||
raise DouyuActivityError("小店兑换缺少商品 ID")
|
||
if price < 0:
|
||
raise DouyuActivityError("小店兑换商品价格无效")
|
||
|
||
params = {
|
||
"_service": "buy.plug.svr.web",
|
||
"_plug_id": "7200",
|
||
"_jsvar": "buyInfo",
|
||
"acctype": "livelink",
|
||
"_app_id": "2123",
|
||
"iActionId": action_id or act_id,
|
||
"propid": commodity_id,
|
||
"paytype": str(pay_type),
|
||
"source": rid,
|
||
"sAnchorId": rid,
|
||
"attach": json.dumps({"ch": "0"}, separators=(",", ":")),
|
||
"apptype": "3",
|
||
"_biz_code": "cjm",
|
||
"_act_id": act_id,
|
||
"sLiveUserInfo": json.dumps(
|
||
{"sAnchorId": rid, "sVideoId": 0}, separators=(",", ":"),
|
||
),
|
||
"appext": json.dumps(
|
||
{"user_price": price, "sGoodsPic": picture, "paytype": pay_type},
|
||
separators=(",", ":"),
|
||
),
|
||
"isCode": "1",
|
||
"authType": "delegate",
|
||
"_sid": "8",
|
||
"set": "8",
|
||
}
|
||
for key in ("gameId", "actId", "appId", "livePlatId", "code", "timestamp", "v", "sig"):
|
||
value = str(embed_query.get(key) or "")
|
||
if value:
|
||
params[key] = value
|
||
|
||
response = self._request(
|
||
"get",
|
||
self.XPD_LIVE_BUY_API,
|
||
source="兑换小店商品",
|
||
params=params,
|
||
headers=self._xpd_daoju_headers(),
|
||
)
|
||
info = self._xpd_parse_var(response.text, "buyInfo")
|
||
if str(info.get("ret")) != "0" or str(info.get("msg") or "").lower() != "ok":
|
||
raise DouyuActivityError(str(info.get("msg") or "小店兑换失败"))
|
||
return {
|
||
"commodity_id": commodity_id,
|
||
"pay_type": pay_type,
|
||
"price": price,
|
||
"serial": str(info.get("serial") or info.get("sSerialNum") or ""),
|
||
"event_id": str(info.get("event_id") or ""),
|
||
"new_balance": self._xpd_int(info.get("newBalance")),
|
||
"raw": info,
|
||
}
|