521 lines
20 KiB
Python
521 lines
20 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"
|
|
PAY_QR_API = "https://www.douyu.com/japi/oms/web/pay/getQrCode"
|
|
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 "查询绑定信息失败")
|
|
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")
|
|
return {
|
|
"act_alias": act_alias,
|
|
"api_version": "v2" if v2 else "v1",
|
|
"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,
|
|
"raw": payload,
|
|
}
|
|
|
|
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_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_elite_gift(
|
|
self,
|
|
*,
|
|
gift_count: int,
|
|
room_id: str,
|
|
gift_id: str,
|
|
skin_id: str,
|
|
) -> dict[str, Any]:
|
|
"""赠送精英令礼物。"""
|
|
payload = self._request_json(
|
|
"post",
|
|
self.DONATE_API,
|
|
"赠送精英令",
|
|
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 "赠送精英令失败")
|
|
return {"raw": payload}
|
|
|
|
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}
|