yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(douyu): 新增和平小店查询模块(绑定角色/商品列表/点券余额)

Browse Source 
- activity_client: getIframeUrl 签发 code/sig,道聚城 getrole.out/recommend/balance 接口
  (2026-08 起需 isCode=1 + authType=delegate + sAnchorId 才能通过 Livelink 校验)
- 迁移 0016: accounts xpd_* 字段、douyu_config 小店配置(actAlias/actId/rid)、
  douyu_xpd_goods_snapshot 商品快照表
- runner/service/router/schema 注册 query_xpd_role/refresh_xpd_goods/query_xpd_balance
  三个任务,任务结果剥离 role.raw 防 openid 泄露
- 前端: 和平小店路由/菜单/DouyuTasksPage 三态/商品下拉/配置弹窗
This commit is contained in:
yml2213
2026-08-06 17:23:59 +08:00
parent f8de495029
commit fdc74eba8c
12 changed files with 702 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/douyu/activity_client.py
+230 -1
View File
@@ -4,10 +4,12 @@ from __future__ import annotations
import json
import random
import re
import string
import time
from collections.abc import Callable
from typing import Any
from urllib.parse import unquote
from urllib.parse import parse_qs, unquote, urlsplit
import requests
@@ -813,3 +815,230 @@ class DouyuActivityClient:
)
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_DAOJU_REFERER = "https://app.daoju.qq.com/"
XPD_GET_ROLE_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_UA = (
"Mozilla/5.0 (Linux; Android 12; HBN-AL00 Build/V417IR; wv) "
"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
"Chrome/101.0.4951.61 Mobile Safari/537.36, Douyu_Android"
)
@staticmethod
def _xpd_int(value: Any) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return None
@staticmethod
def _xpd_parse_var(text: str, varname: str) -> dict[str, Any]:
"""解析道聚城 `var xxx={...};` 形式的响应（容忍等号两侧空白）。"""
match = re.search(rf"var\s+{varname}\s*=", text)
if not match:
raise DouyuActivityError(f"响应中找不到 var {varname}=: {text[:200]}")
chunk = text[match.end():].strip()
if chunk.endswith(";"):
chunk = chunk[:-1]
try:
return json.loads(chunk)
except json.JSONDecodeError as exc:
raise DouyuActivityError(f"var {varname}= 解析失败: {chunk[:200]}") from exc
@classmethod
def _xpd_daoju_headers(cls) -> dict[str, str]:
return {
"User-Agent": cls.XPD_UA,
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9",
"X-Requested-With": "air.tv.douyu.android",
"Sec-Fetch-Site": "same-site",
"Sec-Fetch-Mode": "no-cors",
"Sec-Fetch-Dest": "script",
"Referer": cls.XPD_DAOJU_REFERER,
}
def _xpd_token(self) -> str:
"""拼斗鱼 token: acf_uid_1_acf_stk_0_acf_ltkid。"""
uid = cookie_value(self.cookie, "acf_uid")
stk = cookie_value(self.cookie, "acf_stk")
ltkid = cookie_value(self.cookie, "acf_ltkid")
if not uid or not stk or not ltkid:
raise DouyuActivityError("Cookie 缺少 acf_uid/acf_stk/acf_ltkid，无法获取小店参数")
return f"{uid}_1_{stk}_0_{ltkid}"
def xpd_embed_query(self, *, act_alias: str, rid: str) -> dict[str, Any]:
"""从斗鱼换取和平小店 H5 参数（code/sig/timestamp/actId 等，每次现签）。"""
payload = self._request_json(
"get",
self.XPD_EMBED_API,
"获取小店 H5 参数",
params={"actAlias": act_alias, "rid": rid, "token": self._xpd_token()},
headers={"Referer": self.XPD_H5_REFERER},
)
if payload.get("error") not in (0, "0", None):
raise DouyuActivityError(payload.get("msg") or "获取小店 H5 参数失败")
data = payload.get("data") or {}
txurl_h5 = str(data.get("txurlH5") or "")
if not txurl_h5:
raise DouyuActivityError(f"获取小店 H5 参数失败: {payload}")
query = parse_qs(urlsplit(txurl_h5).query)
embed_query = {key: values[0] for key, values in query.items() if values}
return {"query": embed_query, "txurl_h5": txurl_h5, "raw": payload}
def xpd_get_role(
self,
*,
embed_query: dict[str, str],
act_id: str,
rid: str = "",
) -> dict[str, Any]:
"""用小店 code/sig 换取当前绑定角色信息。
2026-08 起道聚城校验升级：必须带 isCode=1 + authType=delegate + sAnchorId(房间号)。
"""
params = {
"acctype": "livelink",
"_jsvar": "info",
"_service": "other.livelink.getrole.out",
"_biz_code": "cjm",
"_act_id": act_id,
"_app_id": "2123",
"isCode": "1",
"gameId": embed_query.get("gameId") or "cjm",
"actId": embed_query.get("actId") or "",
"appId": embed_query.get("appId") or "bp_cf",
"livePlatId": embed_query.get("livePlatId") or "douyu",
"code": embed_query.get("code") or "",
"timestamp": embed_query.get("timestamp") or "",
"v": embed_query.get("v") or "",
"sig": embed_query.get("sig") or "",
"authType": "delegate",
"sAnchorId": rid,
"sVideoId": "",
}
response = self._request(
"get",
self.XPD_GET_ROLE_API,
source="查询小店绑定角色",
params=params,
headers=self._xpd_daoju_headers(),
)
info = self._xpd_parse_var(response.text, "info")
return {
"game_open_id": str(info.get("gameOpenId") or ""),
"role_id": str(info.get("roleId") or ""),
"role_name": str(info.get("roleName") or ""),
"type": str(info.get("type") or ""),
"plat_id": str(info.get("platId") or ""),
"area": str(info.get("area") or ""),
"raw": info,
}
def xpd_list_goods(
self,
*,
embed_query: dict[str, str],
act_id: str,
openid: str,
roleid: str,
areaid: str = "1",
) -> dict[str, Any]:
"""查询和平小店推荐商品（category=76 全部商品）。"""
params = {
"_service": "act.recommend.query",
"_biz_code": "cjm",
"_app_id": "2123",
"app_id": "2123",
"page_begin": "0",
"page_num": "1",
"page_size": "200",
"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",
"userid": openid,
"roleid": roleid,
"area": "1",
"areaid": str(areaid),
"need_limit": "1",
}
response = self._request(
"get",
self.XPD_RECOMMEND_API,
source="刷新小店商品",
params=params,
headers=self._xpd_daoju_headers(),
)
recommend = self._xpd_parse_var(response.text, "recommend")
items = (((recommend.get("data") or {}).get("client_data") or {}).get("itemsdetail")) or []
goods = []
for item in items:
goods.append(
{
"commodity_id": str(item.get("iGoodsId") or ""),
"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,
}
)
return {"goods": goods, "raw": recommend}
def xpd_balance(
self,
*,
embed_query: dict[str, str],
act_id: str,
openid: str,
roleid: str,
plat: str,
areaid: str = "1",
) -> dict[str, Any]:
"""查询和平小店角色点券余额。"""
params = {
"_jsvar": "banlanceInfo",
"_service": "pay.midas.dq.get.ttpp",
"_app_id": "2123",
"acctype": "ttpp",
"areaid": str(areaid),
"eventid": "",
"interwork": "0",
"openid": openid,
"openkey": "openkey",
"partition": "0",
"pay_token": "",
"plat": str(plat),
"plat_pc": "0",
"roleid": roleid,
"_biz_code": "cjm",
"_act_id": act_id,
"_time": str(int(time.time())),
"_sid": "6",
}
response = self._request(
"get",
self.XPD_BALANCE_API,
source="查询小店点券余额",
params=params,
headers=self._xpd_daoju_headers(),
)
info = self._xpd_parse_var(response.text, "banlanceInfo")
return {"balance": self._xpd_int(info.get("balance")), "raw": info}