816 lines
32 KiB
Python
816 lines
32 KiB
Python
"""斗鱼活动、兑换与充值接口客户端。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import random
|
||
import string
|
||
from collections.abc import Callable
|
||
from typing import Any
|
||
from urllib.parse import unquote
|
||
|
||
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}
|