"""斗鱼活动、兑换与充值接口客户端。""" from __future__ import annotations import json import re import time from collections.abc import Callable from typing import Any from urllib.parse import parse_qs, unquote, urlsplit, urlunsplit 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" CREATE_EXCHANGE_ORDER_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/createExchangeOrder" PAY_EXCHANGE_ORDER_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/payExchangeOrder" EXCHANGE_ORDER_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/getExchangeOrderList" EXCHANGE_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchangeList" # 2026-08-30 页面新增链路（对齐浏览器）： # 预兑 user/queryExchange -> user/confirmExchange；取消预兑 user/queryCancelExchange -> user/cancelExchange； # 预约到货 user/sub / user/cancelSub；批量兑换 batchExchangeLimit；单步兑换 exchange；自动兑换开关。 ELITE_USER_INFO_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/getUserInfo" GOODS_DETAIL_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/storedetail" EXCHANGE_GIFT_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchange" PRE_EXCHANGE_CHECK_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/queryExchange" PRE_EXCHANGE_CONFIRM_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/confirmExchange" PRE_EXCHANGE_CANCEL_CHECK_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/queryCancelExchange" PRE_EXCHANGE_CANCEL_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/cancelExchange" COMMODITY_SUBSCRIBE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/sub" COMMODITY_UNSUBSCRIBE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/user/cancelSub" BATCH_EXCHANGE_LIMIT_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/batchExchangeLimit" AUTO_CONVERT_QUERY_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/activity/automaticallyConvert" AUTO_CONVERT_SAVE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/activity/automaticallyConvertSave" EXCHANGE_HOME_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/activity/exchangeHome" 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, } @staticmethod def _safe_url(url: str) -> str: """移除 query/fragment，避免将 Cookie、签名或一次性参数写入日志。""" parsed = urlsplit(url) return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) def _redirect_error( self, method: str, url: str, source: str, response: requests.Response, ) -> DouyuActivityError: location = response.headers.get("Location", "") safe_location = self._safe_url(location) if location else "-" trace = ( f"method={method.upper()} url={self._safe_url(url)} " f"location={safe_location} history={response.status_code}->{safe_location}" ) self.logger(f"{source}: 已拦截重定向 | {trace}") return DouyuActivityError(f"{source} 请求被重定向，Cookie 已失效，请重新登录（{trace}）") 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) # 活动 JSON 接口不应自动跳到登录页；忽略调用方传入的覆盖值。 kwargs.pop("allow_redirects", None) try: response = self.session.request( method, url, headers=req_headers, timeout=kwargs.pop("timeout", self.timeout), allow_redirects=False, **kwargs, ) except requests.RequestException as exc: raise DouyuActivityError( f"{source} 请求失败: {method.upper()} {self._safe_url(url)}: {exc}" ) from exc self._merge_response_cookie(response) if 300 <= response.status_code < 400: raise self._redirect_error(method, url, source, 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, *, force_refresh: bool = False) -> str: """获取 cvl_csrf_token；优先复用 Cookie 中已有值，与浏览器行为一致。 2026-08-30 抓包实证：页面只读 Cookie `cvl_csrf_token` 的值作为 csrfToken 复用（同一 token 跨多次接口、跨十几秒复用），Cookie 里没有/失效时才调用 generateCsrf，生成的 token 写回 Cookie 继续复用。 force_refresh=True 用于服务端提示 token 失效后的强制刷新。 """ if not force_refresh: token = cookie_value(self.cookie, "cvl_csrf_token") if token: return 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 @staticmethod def _is_csrf_error(payload: dict[str, Any]) -> bool: """判断响应是否为 csrfToken 校验失败（此时应刷新并重试一次）。""" return "csrf" in str(payload.get("msg") or "").lower() @staticmethod def _peace_page_referer(rid: str = "") -> str: """精英手册首页 Referer（对齐浏览器 getUserInfo 等接口）。""" if rid: return f"https://www.douyu.com/pages/live-peace-handbook/web?ditchname=pass0&roomId={rid}" return "https://www.douyu.com/pages/live-peace-handbook/web?ditchname=pass0" @staticmethod def _peace_shop_referer(rid: str = "") -> str: """精英手册商城页 Referer（对齐浏览器 create/pay/storedetail 等接口）。""" if rid: return f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}" return "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0" def _post_csrf_form( self, url: str, source: str, fields: dict[str, Any], referer: str, ) -> dict[str, Any]: """提交带 csrfToken 的 multipart 表单（对齐浏览器请求层）。 浏览器 csrf 配置 {tvk: cvl_csrf_token, tn: csrfToken} 会把 Cookie 中的 token 值拼进表单；这里同样复用 Cookie 值，仅在服务端提示 token 失效时 刷新一次后重试。 """ token = self.csrf_token() payload = self._request_json( "post", url, source, files=self._multipart_fields({**fields, "csrfToken": token}), headers={"Referer": referer}, ) if payload.get("error") not in (0, "0") and self._is_csrf_error(payload): self.logger(f"{source}: csrfToken 校验失败，刷新后重试一次") token = self.csrf_token(force_refresh=True) payload = self._request_json( "post", url, source, files=self._multipart_fields({**fields, "csrfToken": token}), headers={"Referer": referer}, ) return payload 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} @staticmethod def _multipart_fields(data: dict[str, Any]) -> dict[str, tuple[None, str]]: """Build requests multipart fields without uploading files.""" return {key: (None, str(value)) for key, value in data.items()} def create_exchange_order( self, *, manual_id: str, rid: str, commodity_id: str, num: int = 1, ) -> dict[str, Any]: """Lock stock for an activity commodity before spending points.""" if num < 1: raise DouyuActivityError("锁单数量必须大于 0") payload = self._post_csrf_form( self.CREATE_EXCHANGE_ORDER_API, "锁定兑换商品", { "manualID": manual_id, "rid": rid, "commodityID": commodity_id, "num": num, }, referer=self._peace_shop_referer(rid), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "锁定兑换商品失败") data = payload.get("data") or {} order_id = str(data.get("orderId") or "").strip() if not order_id: raise DouyuActivityError("锁定兑换商品成功，但响应中没有 orderId") locked_at = int(time.time()) try: expire_seconds = int(data.get("expireSeconds") or 0) except (TypeError, ValueError): expire_seconds = 0 if expire_seconds > 0: expires_at = locked_at + expire_seconds else: expires_at = None return { "order_id": order_id, "commodity_id": commodity_id, "commodity_name": data.get("commodityName") or "", "commodity_image": data.get("commodityImage") or "", "score": data.get("score"), "num": data.get("num", num), "expire_seconds": data.get("expireSeconds"), "locked_at": locked_at, "expires_at": expires_at, "raw": payload, } def pay_exchange_order(self, *, manual_id: str, order_id: str, rid: str = "") -> dict[str, Any]: """Pay a previously locked exchange order with handbook points.""" normalized_order_id = str(order_id).strip() if not normalized_order_id: raise DouyuActivityError("锁单订单号不能为空") payload = self._post_csrf_form( self.PAY_EXCHANGE_ORDER_API, "支付锁单", { "orderID": normalized_order_id, "manualID": manual_id, }, referer=self._peace_shop_referer(rid), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "支付锁单失败") data = payload.get("data") or {} return { "order_id": str(data.get("orderId") or normalized_order_id), "exchange_id": str(data.get("exchangeId") or ""), "commodity_type": data.get("commodityType"), "commodity_image": data.get("commodityImage") or "", "exchange_num": data.get("exchangeNum"), "s_type": data.get("sType"), "g_type": data.get("gType"), "raw": payload, } def exchange_orders(self, *, manual_id: str, page: int = 1, page_size: int = 10) -> dict[str, Any]: """Query current locked exchange orders.""" payload = self._request_json( "get", self.EXCHANGE_ORDER_LIST_API, "查询锁单", params={"page": page, "pageSize": page_size, "manualID": manual_id}, headers={ "Origin": "", "Referer": "https://www.douyu.com/pages/live-peace-handbook/web/lock-order?ditchname=pass0", }, ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询锁单失败") data = payload.get("data") or {} return { "orders": data.get("list") or [], "total": data.get("total") or 0, "raw": payload, } def exchange_goods(self, *, manual_id: str, rid: str, commodity_id: str) -> dict[str, Any]: """Lock an activity commodity and immediately pay the locked order.""" locked = self.create_exchange_order( manual_id=manual_id, rid=rid, commodity_id=commodity_id, num=1, ) paid = self.pay_exchange_order(manual_id=manual_id, order_id=locked["order_id"]) return { "commodity_id": commodity_id, "order_id": locked["order_id"], "exchange_id": paid["exchange_id"], "commodity_image": paid["commodity_image"] or locked["commodity_image"], "exchange_num": paid["exchange_num"], "lock_order": locked, "payment": paid, } def elite_user_info(self, *, manual_id: str, rid: str = "") -> dict[str, Any]: """查询精英手册状态（对齐页面 getUserInfo，含 manualType/manualScore）。 浏览器每次进入手册页都会先请求该接口，用于判断商品是否"精英专享"、 是否已达兑换上限等状态；兑换前先查一次与页面行为一致。 """ payload = self._request_json( "get", self.ELITE_USER_INFO_API, "查询手册状态", params={"manualID": manual_id}, headers={"Origin": "", "Referer": self._peace_page_referer(rid)}, ) 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, } def query_goods_detail(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]: """查询商品详情 storedetail（对齐浏览器 getGoodsDetail）。 详情里带有兑换状态机所需的全部标记：preExchange.isExists / subscribeType / exchangeInfo.userStatus / batchExchange / roomLimited / storeStatus 等。 """ payload = self._request_json( "get", self.GOODS_DETAIL_API, "查询商品详情", params={"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"}, headers={"Origin": "", "Referer": self._peace_shop_referer(rid)}, ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询商品详情失败") return {"detail": payload.get("data") or {}, "raw": payload} @staticmethod def resolve_exchange_plan( detail: dict[str, Any], *, manual_type: Any = None, batch_num_limit: Any = None, ) -> dict[str, Any]: """对齐浏览器兑换按钮状态机（$e / Ve / Ge）给出自动化兑换计划。 浏览器判定顺序（2026-08-30 bundle index.4872272b.js / index.0539fbf8.js）： 存在预扣订单 -> 已达兑换上限 -> 无库存 -> 限定房间 -> 活动未开启(预约/预兑) -> 精英专享未开通 -> 已预兑等待开放 -> 立即兑换(经典锁单/支付)。 action: classic 经典链路 createExchangeOrder -> payExchangeOrder（批量商品 num 受上限约束） pre_exchange 预兑商品（subscribeType=2 且未开启）：user/queryExchange -> user/confirmExchange subscribe 预约到货商品（subscribeType=1 且未开启）：user/sub wait 已预兑（userStatus=1）但预扣订单尚未生成，等待开放后重试 blocked 明确不可兑换，text 为原因（文案对齐浏览器/服务端） """ detail = detail or {} pre = detail.get("preExchange") or {} pre_is_exists = int(pre.get("isExists") or 0) exchange_info = detail.get("exchangeInfo") or {} user_status = int(exchange_info.get("userStatus") or 0) subscribe_type = int(detail.get("subscribeType") or 0) status = int(detail.get("status") or 0) store_status = int(detail.get("storeStatus") or 0) open_status = int(detail.get("openStatus") or 0) unlimited = ( int(detail.get("unlimitedMode") or 0) == 1 and int(detail.get("unlimitedStatus") or 0) == 1 ) room_limited = int(detail.get("roomLimited") or 0) == 1 is_limit_room = int(detail.get("isLimitRoom") or 0) == 1 elite_limited = int(detail.get("eliteLimited") or 0) == 1 batch_exchange = int(detail.get("batchExchange") or 0) > 0 max_num = 1 if batch_exchange and batch_num_limit is not None: try: max_num = max(1, int(batch_num_limit or 1)) except (TypeError, ValueError): max_num = 1 plan: dict[str, Any] = { "action": "classic", "text": "", "max_num": max_num, "batch_exchange": batch_exchange, "pre_exchange": bool(pre_is_exists), "subscribe_type": subscribe_type, "user_status": user_status, } if pre_is_exists: plan.update(action="classic", text="存在预扣订单，继续兑换") elif status == 1 or (subscribe_type == 2 and user_status == 2): plan.update(action="blocked", text="已达兑换上限") elif store_status == 0 and not unlimited: plan.update(action="blocked", text="无库存") elif room_limited and not is_limit_room: plan.update(action="blocked", text="限定房间兑换，请前往指定直播间") elif open_status == 0: if subscribe_type == 1: plan.update(action="subscribe", text="活动未开启，预约到货") elif subscribe_type == 2: plan.update(action="pre_exchange", text="活动未开启，预兑") else: plan.update(action="blocked", text="活动未开启") elif elite_limited and manual_type is not None and int(manual_type or 0) != 1: plan.update(action="blocked", text="需要开通精英手册，才可以兑换本道具") elif subscribe_type == 2 and user_status == 1: plan.update(action="wait", text="已预兑，等待开放后继续兑换") else: plan.update(action="classic", text="立即兑换") return plan def batch_exchange_limit(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]: """查询批量兑换上限（对齐页面 batchExchangeLimit，仅 batchExchange=1 商品）。""" payload = self._post_csrf_form( self.BATCH_EXCHANGE_LIMIT_API, "查询批量兑换上限", {"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"}, referer=self._peace_shop_referer(rid), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询批量兑换上限失败") data = payload.get("data") or {} return {"limit": data.get("limit"), "raw": payload} def pre_exchange_check(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]: """查询预兑状态（对齐页面 user/queryExchange，预兑弹窗前必查）。""" payload = self._request_json( "get", self.PRE_EXCHANGE_CHECK_API, "查询预兑状态", params={"manualID": manual_id, "commodityID": commodity_id}, headers={"Origin": "", "Referer": self._peace_shop_referer()}, ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询预兑状态失败") return {"data": payload.get("data") or {}, "raw": payload} def pre_exchange_cancel_check(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]: """查询取消预兑信息（对齐页面 user/queryCancelExchange）。""" payload = self._request_json( "get", self.PRE_EXCHANGE_CANCEL_CHECK_API, "查询取消预兑信息", params={"manualID": manual_id, "commodityID": commodity_id}, headers={"Origin": "", "Referer": self._peace_shop_referer()}, ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询取消预兑信息失败") return {"data": payload.get("data") or {}, "raw": payload} def confirm_pre_exchange(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]: """确认预兑（对齐页面 user/confirmExchange，成功提示"预兑成功"）。""" payload = self._post_csrf_form( self.PRE_EXCHANGE_CONFIRM_API, "确认预兑", {"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"}, referer=self._peace_shop_referer(rid), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "确认预兑失败") return {"raw": payload} def cancel_pre_exchange(self, *, manual_id: str, commodity_id: str, rid: str = "") -> dict[str, Any]: """取消预兑（对齐页面 user/cancelExchange）。""" payload = self._post_csrf_form( self.PRE_EXCHANGE_CANCEL_API, "取消预兑", {"manualID": manual_id, "commodityID": commodity_id, "rid": rid or "0"}, referer=self._peace_shop_referer(rid), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "取消预兑失败") return {"raw": payload} def subscribe_commodity(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]: """预约到货（对齐页面 user/sub，成功提示"预约成功"）。""" payload = self._post_csrf_form( self.COMMODITY_SUBSCRIBE_API, "预约到货", {"manualID": manual_id, "commodityID": commodity_id}, referer=self._peace_shop_referer(), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "预约到货失败") return {"raw": payload} def unsubscribe_commodity(self, *, manual_id: str, commodity_id: str) -> dict[str, Any]: """取消预约到货（对齐页面 user/cancelSub）。""" payload = self._post_csrf_form( self.COMMODITY_UNSUBSCRIBE_API, "取消预约到货", {"manualID": manual_id, "commodityID": commodity_id}, referer=self._peace_shop_referer(), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "取消预约到货失败") return {"raw": payload} def exchange_gift(self, *, manual_id: str, commodity_id: str, rid: str = "", num: int = 1) -> dict[str, Any]: """单步兑换（对齐页面 postExchangeGift -> /ordnc/activity/peace/exchange）。 2026-08-30 bundle 中新链路，页面主要用于兑换中心/自动兑换场景；字段按页面 调用（{manualID, commodityID, num, rid} + csrfToken）对齐。 """ payload = self._post_csrf_form( self.EXCHANGE_GIFT_API, "单步兑换", { "manualID": manual_id, "commodityID": commodity_id, "num": max(1, int(num)), "rid": rid or "0", }, referer=self._peace_shop_referer(rid), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "兑换失败") data = payload.get("data") or {} return { "exchange_id": str(data.get("exchangeId") or ""), "commodity_type": data.get("commodityType"), "commodity_image": data.get("commodityImage") or "", "exchange_num": data.get("exchangeNum"), "s_type": data.get("sType"), "g_type": data.get("gType"), "raw": payload, } def auto_convert_switch(self, *, manual_id: str) -> dict[str, Any]: """查询自动兑换开关（对齐页面 activity/automaticallyConvert）。""" payload = self._request_json( "get", self.AUTO_CONVERT_QUERY_API, "查询自动兑换开关", params={"manualID": manual_id}, headers={"Origin": "", "Referer": self._peace_shop_referer()}, ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询自动兑换开关失败") return {"data": payload.get("data") or {}, "raw": payload} def set_auto_convert(self, *, manual_id: str, switch: Any) -> dict[str, Any]: """设置自动兑换开关（对齐页面 automaticallyConvertSave {manualID, switch}）。""" payload = self._post_csrf_form( self.AUTO_CONVERT_SAVE_API, "设置自动兑换开关", {"manualID": manual_id, "switch": switch}, referer=self._peace_shop_referer(), ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "设置自动兑换开关失败") return {"raw": payload} def exchange_home_info(self) -> dict[str, Any]: """查询兑换中心信息（对齐页面 activity/exchangeHome）。""" payload = self._request_json( "get", self.EXCHANGE_HOME_API, "查询兑换中心", headers={"Origin": "", "Referer": self._peace_shop_referer()}, ) if payload.get("error") not in (0, "0"): raise DouyuActivityError(payload.get("msg") or "查询兑换中心失败") return {"data": payload.get("data") or {}, "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") plat_id = info.get("platId") area = info.get("area") 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 "", "area": str(area) if area is not None else "", "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() * 1000)), "_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, }