feat(huya): 统一积分兑换并完善兑换记录

This commit is contained in:
yml
2026-11-13 19:03:55 +08:00
parent b89ee4fd18
commit 711be0e2b6
17 changed files with 2064 additions and 193 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
"""虎牙活动默认配置。"""
HUYA_DEFAULT_ROOM_PID = "1199650619883"
HUYA_DEFAULT_SID = "2203"
HUYA_DEFAULT_SID = "2277"
HUYA_DEFAULT_OUTER_ACT_ID = "17096"
# 2026-11-09 最新活动页抓包:绑定走 gid=3203、bActId=0。
HUYA_DEFAULT_BIND_GID = "3203"
+13 -2
View File
@@ -31,10 +31,21 @@ def create_browser_exchange_kit(
raise HTTPException(status_code=400, detail="该账号没有可用 Cookie,请先登录")
config = ensure_huya_config(db)
room_pid = huya_config_value("room_pid", config.room_pid)
points_page_code = huya_config_value("points_page_code", config.points_page_code)
points_source_id = huya_config_value("points_source_id", config.points_source_id)
target = req.target if req and req.target else DEFAULT_EXCHANGE_TARGET
targets = exchange_target_options(room_pid)
targets = exchange_target_options(
room_pid,
points_page_code=points_page_code,
points_source_id=points_source_id,
)
try:
target_url = exchange_target_url(target, room_pid)
target_url = exchange_target_url(
target,
room_pid,
points_page_code=points_page_code,
points_source_id=points_source_id,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
pairs = dict(cookie_pairs(normalize_huya_cookie(account.cookie)))
+16 -3
View File
@@ -165,11 +165,24 @@ def list_goods(
db: Session = Depends(get_db),
current: User = Depends(require_permission("huya:task")),
):
"""查看已缓存的虎牙商品快照。"""
rows = db.query(HuyaGoodsSnapshot).order_by(HuyaGoodsSnapshot.id.desc()).all()
"""查看最近一次从积分中心 ``queryAllocation`` 刷新的商品快照。"""
rows = (
db.query(HuyaGoodsSnapshot)
.filter(HuyaGoodsSnapshot.account_id.is_not(None))
.order_by(
HuyaGoodsSnapshot.updated_at.desc(),
HuyaGoodsSnapshot.id.desc(),
)
.all()
)
if not rows:
return []
newest = rows[0].updated_at
latest = {}
for row in rows:
latest.setdefault((row.sid, row.product_id), row)
if row.updated_at != newest:
continue
latest.setdefault(row.product_id, row)
return list(latest.values())
+47 -7
View File
@@ -10,6 +10,10 @@
from __future__ import annotations
from ..huya_defaults import (
HUYA_DEFAULT_POINTS_PAGE_CODE,
HUYA_DEFAULT_POINTS_SOURCE_ID,
)
from .huya_runner_bind import HUYA_BIND_ROOM_ID, HUYA_BIND_ZT_UUID
from .huya_runner_core import HUYA_RECHARGE_SOURCE_ID
@@ -20,7 +24,7 @@ EXCHANGE_TARGET_ESPORTS = "esports"
#: 未指定目标时使用的默认页(保持历史行为:精英宝典)
DEFAULT_EXCHANGE_TARGET = EXCHANGE_TARGET_ELITE
#: 精英宝典复用绑定/充值活动 uuid;电竞宝典为独立的 zt 活动 uuid
#: 旧 zt 活动 uuid 仅保留给旧调用兼容;精英宝典现统一走积分中心页。
HUYA_ELITE_ZT_UUID = HUYA_BIND_ZT_UUID
HUYA_ESPORTS_ZT_UUID = "50ea007f"
@@ -52,25 +56,61 @@ def _zt_exchange_url(zt_uuid: str, room_pid: str) -> str:
)
def exchange_target_url(key: str, room_pid: str) -> str:
"""返回指定目标页地址;pid 由调用方(路由)归一化,空值在这里会拼出无 pid 的地址。"""
def _points_exchange_url(
room_pid: str,
*,
page_code: str = HUYA_DEFAULT_POINTS_PAGE_CODE,
source_id: str = HUYA_DEFAULT_POINTS_SOURCE_ID,
) -> str:
pid = str(room_pid or "").strip()
return (
"https://m-shop.huya.com/points.html"
f"?pagecode={page_code}"
f"&sourceId={source_id}"
f"&pid={pid}"
f"&anchorUid={pid}"
f"&roomid={HUYA_BIND_ROOM_ID}"
)
def exchange_target_url(
key: str,
room_pid: str,
*,
points_page_code: str = HUYA_DEFAULT_POINTS_PAGE_CODE,
points_source_id: str = HUYA_DEFAULT_POINTS_SOURCE_ID,
) -> str:
"""返回指定目标页地址;pid 由调用方(路由)归一化,空值在这里会拼出无 pid 的地址。"""
if key == EXCHANGE_TARGET_LIVE_ROOM:
return f"https://www.huya.com/{HUYA_BIND_ROOM_ID}"
if key == EXCHANGE_TARGET_ELITE:
return _zt_exchange_url(HUYA_ELITE_ZT_UUID, pid)
return _points_exchange_url(
room_pid,
page_code=points_page_code,
source_id=points_source_id,
)
if key == EXCHANGE_TARGET_ESPORTS:
return _zt_exchange_url(HUYA_ESPORTS_ZT_UUID, pid)
return _zt_exchange_url(HUYA_ESPORTS_ZT_UUID, str(room_pid or "").strip())
raise ValueError(f"不支持的兑换目标:{key}")
def exchange_target_options(room_pid: str) -> list[dict[str, str]]:
def exchange_target_options(
room_pid: str,
*,
points_page_code: str = HUYA_DEFAULT_POINTS_PAGE_CODE,
points_source_id: str = HUYA_DEFAULT_POINTS_SOURCE_ID,
) -> list[dict[str, str]]:
"""返回三个可选目标页,顺序固定,供前端选择器渲染。"""
return [
{
"key": key,
"label": EXCHANGE_TARGET_LABELS[key],
"url": exchange_target_url(key, room_pid),
"url": exchange_target_url(
key,
room_pid,
points_page_code=points_page_code,
points_source_id=points_source_id,
),
}
for key in EXCHANGE_TARGET_KEYS
]
@@ -0,0 +1,443 @@
"""虎牙积分中心兑换的唯一编排入口。
2026-11-13 HAR 实证:
* 积分不足时,先通过商城 ``createOrderV5 / payOrderSubmitV5`` 购买服务端推荐的
积分包;
* 支付到账后,再调用 ``getRedeemNeedGoods`` 确认积分;
* 积分足够后使用 ``createOrderV6`` 兑换目标道具,纯积分兑换不再发起支付。
精英宝典工作台、通用任务页以及后续浏览器入口都只调用本模块,不再直接实现
``scoreExchangePrize`` 或商城下单分支。
"""
from __future__ import annotations
import time
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any
from .huya_shop_flow import ShopOrderOutcome, run_open_flow
EXCHANGE_STATUS_EXCHANGED = "exchanged"
EXCHANGE_STATUS_PAYMENT_REQUIRED = "payment_required"
EXCHANGE_STATUS_FAILED = "failed"
def _int(value: Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
def _text(value: Any) -> str:
return str(value or "").strip()
def normalize_spu_id(value: Any) -> str:
"""把数值或 ``hy-`` 前缀 SPU 统一成接口使用的字符串形式。"""
text = _text(value)
if not text:
return ""
if text.startswith("hy-"):
return text
if text.isdigit():
return f"hy-{text}"
return text
def spu_number(value: Any) -> int:
text = normalize_spu_id(value)
if "-" in text:
text = text.rsplit("-", 1)[-1]
return _int(text)
def _response_code(response: Any) -> int:
if response is None:
return 0
return _int(
getattr(response, "code", None)
or getattr(response, "status", None)
or getattr(response, "value", lambda: 0)("tag0")
)
def _response_message(response: Any) -> str:
if response is None:
return ""
return _text(
getattr(response, "message", "")
or getattr(response, "msg", "")
or getattr(response, "value", lambda: "")(*("tag1",))
)
def _value(response: Any, *path: str) -> Any:
if response is None:
return None
getter = getattr(response, "value", None)
if not callable(getter):
return None
return getter(*path)
def _stock_remaining(value: Any) -> int | None:
stock = _int(value)
return stock if stock > 0 else None
def _parse_plan(item: Any) -> dict[str, Any] | None:
if not isinstance(item, dict):
return None
spu_id = normalize_spu_id(item.get("tag0"))
sku_id = _int(item.get("tag1"))
name = _text(item.get("tag2"))
if not spu_id or not sku_id:
return None
gain_score = _int(item.get("tag5"))
price = _int(item.get("tag6"))
custom_count = max(1, _int(item.get("tag7")) or 1)
credit_type = _int(item.get("tag9"))
once_max = _int(item.get("tag21"))
count = custom_count if credit_type == 2 else 1
if once_max > 0:
count = min(count, once_max)
groups = item.get("tag8")
group_items = groups if isinstance(groups, list) else []
return {
"spu_id": spu_id,
"sku_id": sku_id,
"name": name or spu_id,
"gain_score": gain_score,
"price": price,
"count": count,
"credit_type": credit_type,
"once_max": once_max,
"group_items": group_items,
"remaining": _stock_remaining(item.get("tag20")),
}
@dataclass
class RedeemState:
code: int
message: str
user_score: int
redeem_need_score: int
need_buy_score: int
meet_precondition: bool
full_score: bool
plans: list[dict[str, Any]] = field(default_factory=list)
raw: dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> dict[str, Any]:
return {
"code": self.code,
"message": self.message,
"user_score": self.user_score,
"redeem_need_score": self.redeem_need_score,
"need_buy_score": self.need_buy_score,
"meet_precondition": self.meet_precondition,
"full_score": self.full_score,
"plans": self.plans,
}
@dataclass
class PointsExchangeOutcome:
status: str
message: str
state: RedeemState | None = None
target: dict[str, Any] = field(default_factory=dict)
order: dict[str, Any] = field(default_factory=dict)
plan: dict[str, Any] | None = None
purchase: ShopOrderOutcome | None = None
detail: dict[str, Any] = field(default_factory=dict)
def _parse_redeem_state(response: Any) -> RedeemState:
code = _response_code(response)
plans_raw = _value(response, "tag7")
plans = [
plan
for plan in (
_parse_plan(item)
for item in (plans_raw if isinstance(plans_raw, list) else [])
)
if plan is not None
]
return RedeemState(
code=code,
message=_response_message(response),
user_score=_int(_value(response, "tag2")),
redeem_need_score=_int(_value(response, "tag3")),
need_buy_score=_int(_value(response, "tag4")),
meet_precondition=_int(_value(response, "tag5")) == 1,
full_score=_int(_value(response, "tag6")) == 1,
plans=plans,
raw=response.to_dict() if hasattr(response, "to_dict") else {},
)
def get_redeem_state(
client: Any,
*,
uid: int,
cookie: str,
spu_id: str,
) -> RedeemState:
response = client.get_redeem_need_goods(
uid=uid,
guid="",
cookie=cookie,
goods_num=spu_number(spu_id),
)
return _parse_redeem_state(response)
def _get_target_goods(
client: Any,
*,
uid: int,
cookie: str,
pid: int,
spu_id: str,
) -> dict[str, Any]:
response = client.get_goods_info(
uid=uid,
guid="",
cookie=cookie,
pid=pid,
spu_id=normalize_spu_id(spu_id),
sku_id=0,
game_id="",
)
if response is None:
raise RuntimeError("兑换商品详情接口无响应")
if _response_code(response) != 200:
raise RuntimeError(_response_message(response) or "兑换商品详情获取失败")
detail = response.to_dict() if hasattr(response, "to_dict") else {}
sku_id = _int(detail.get("sku_id"))
if not sku_id:
raise RuntimeError("兑换商品缺少 SKU,请先刷新商品列表")
return {
"spu_id": normalize_spu_id(spu_id),
"sku_id": sku_id,
"name": _text(detail.get("name")) or normalize_spu_id(spu_id),
"price": _int(detail.get("price")),
"goods_type": _int(detail.get("goods_type")),
"virtual_type": _int(detail.get("virtual_type")),
"detail": detail,
}
def _create_redeem_order(
client: Any,
*,
uid: int,
cookie: str,
pid: int,
target: dict[str, Any],
) -> PointsExchangeOutcome:
response = client.create_order_v6(
uid=uid,
guid="",
cookie=cookie,
pid=pid,
items=[
{
"spu_id": target["spu_id"],
"sku_id": target["sku_id"],
"item_count": 1,
"goods_type": target.get("goods_type", 0),
"virtual_type": target.get("virtual_type", 0),
}
],
)
if response is None:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
"兑换下单接口无响应",
target=target,
)
order = response.to_dict() if hasattr(response, "to_dict") else {}
if _response_code(response) != 200:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
_response_message(response) or "积分兑换下单失败",
target=target,
order=order,
)
return PointsExchangeOutcome(
EXCHANGE_STATUS_EXCHANGED,
f"兑换成功: {target['name']}",
target=target,
order=order,
)
def select_points_plan(
state: RedeemState,
) -> dict[str, Any] | None:
if not state.plans:
return None
simple = [plan for plan in state.plans if not plan.get("group_items")]
candidates = simple or list(state.plans)
enough = [
plan
for plan in candidates
if plan["gain_score"] * max(1, plan["count"]) >= state.need_buy_score
]
pool = enough or candidates
return min(
pool,
key=lambda plan: (
plan["price"] * max(1, plan["count"]),
plan["gain_score"] * max(1, plan["count"]),
),
)
def prepare_exchange(
client: Any,
*,
uid: int,
cookie: str,
pid: int,
spu_id: str,
pay_channel: str = "Zfb",
stopped: Callable[[], bool] | None = None,
log: Callable[[str], None] | None = None,
) -> PointsExchangeOutcome:
"""检查兑换状态;直接兑换或返回一张待支付的积分包订单。"""
state = get_redeem_state(client, uid=uid, cookie=cookie, spu_id=spu_id)
if state.code != 200:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
state.message or f"兑换状态查询失败: {state.code}",
state=state,
)
if not state.meet_precondition:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
"当前账号未满足兑换前置条件,请先开通对应宝典",
state=state,
)
try:
target = _get_target_goods(
client, uid=uid, cookie=cookie, pid=pid, spu_id=spu_id
)
except RuntimeError as exc:
return PointsExchangeOutcome(EXCHANGE_STATUS_FAILED, str(exc), state=state)
if state.full_score:
outcome = _create_redeem_order(
client, uid=uid, cookie=cookie, pid=pid, target=target
)
outcome.state = state
return outcome
plan = select_points_plan(state)
if plan is None:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
"积分不足,服务端没有返回可购买的积分包",
state=state,
target=target,
)
if plan.get("group_items"):
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
"服务端返回组合积分包,当前版本无法安全自动下单",
state=state,
target=target,
plan=plan,
)
purchase = run_open_flow(
client,
uid=uid,
cookie=cookie,
pid=pid,
spu_id=plan["spu_id"],
sku_id=plan["sku_id"],
count=int(plan["count"]),
pay_channel=pay_channel,
snapshot_price=int(plan["price"]),
log=log,
stopped=stopped,
)
if not purchase.ok:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
purchase.message,
state=state,
target=target,
plan=plan,
purchase=purchase,
detail=purchase.detail,
)
return PointsExchangeOutcome(
EXCHANGE_STATUS_PAYMENT_REQUIRED,
f"积分不足,已生成积分包支付码: {plan['name']}",
state=state,
target=target,
plan=plan,
purchase=purchase,
)
def complete_exchange_after_payment(
client: Any,
*,
uid: int,
cookie: str,
pid: int,
spu_id: str,
attempts: int = 5,
interval: float = 3.0,
sleep: Callable[[float], None] = time.sleep,
) -> PointsExchangeOutcome:
"""支付成功后等待积分到账并提交 V6 兑换。"""
state: RedeemState | None = None
last_message = "积分尚未到账"
for index in range(max(1, attempts)):
state = get_redeem_state(client, uid=uid, cookie=cookie, spu_id=spu_id)
if state.code != 200:
last_message = state.message or f"兑换状态查询失败: {state.code}"
elif not state.meet_precondition:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
"支付成功,但账号仍未满足兑换前置条件",
state=state,
)
elif state.full_score:
try:
target = _get_target_goods(
client, uid=uid, cookie=cookie, pid=pid, spu_id=spu_id
)
except RuntimeError as exc:
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED, str(exc), state=state
)
outcome = _create_redeem_order(
client, uid=uid, cookie=cookie, pid=pid, target=target
)
outcome.state = state
return outcome
else:
last_message = (
f"积分到账中:当前 {state.user_score},还需 {state.need_buy_score}"
)
if index + 1 < max(1, attempts):
sleep(interval)
return PointsExchangeOutcome(
EXCHANGE_STATUS_FAILED,
last_message,
state=state,
)
+238 -143
View File
@@ -11,6 +11,14 @@ from sqlalchemy.orm import Session
from core.huya.wss_client import POINTS_REDEEM_GOODS_NUM
from ..models import HuyaAccount, HuyaGoodsSnapshot, HuyaTask
from .huya_pay_link import pay_qr_path, resolve_pay_targets
from .huya_points_exchange import (
EXCHANGE_STATUS_EXCHANGED,
EXCHANGE_STATUS_PAYMENT_REQUIRED,
complete_exchange_after_payment,
normalize_spu_id,
prepare_exchange,
)
# 积分查询超时:HTTP 通道正常几百毫秒,10 秒足够;失败时别让整批任务长时间挂住。
HUYA_POINTS_QUERY_TIMEOUT_SECONDS = 10.0
@@ -37,6 +45,8 @@ class GoodsMixin:
self, uid: int, cookie: str, b_act_id_int: int, outer_act_id: str = ""
): ...
def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ...
def _wait_recharge_payment(self, *args: Any, **kwargs: Any): ...
def _wait_until(self, when: datetime, uid: int) -> bool: ...
def _parse_scheduled_time(self, value: Any) -> datetime | None: ...
@@ -216,16 +226,6 @@ class GoodsMixin:
account_info: dict,
config_info: dict,
):
sid = str(self.payload.get("sid") or config_info.get("sid") or "").strip()
if not sid:
self._mark_task(worker_db, task, "failed", "请先配置虎牙活动 SID")
return
sid_int = self._to_int(sid)
if not sid_int:
self._mark_task(worker_db, task, "failed", f"虎牙活动 SID 无效: {sid}")
return
uid = self._resolve_uid(account_info)
if not uid:
self._mark_task(worker_db, task, "failed", "无法从账号或 Cookie 解析 yyuid")
@@ -236,7 +236,50 @@ class GoodsMixin:
self._mark_task(worker_db, task, "failed", "账号 Cookie 为空")
return
client: Any = self._activity_client(uid, cookie)
from core.huya.shop_structs import extract_points_center_sid
client: Any = self._http_client()
configured_sid = self._to_int(self.payload.get("sid") or config_info.get("sid"))
sid_int = configured_sid
sid_source = "config"
try:
page_code = (
str(config_info.get("points_page_code") or "").strip()
or "heping_baodian_credit_center_page"
)
source_id = (
str(config_info.get("points_source_id") or "").strip()
or "yellowcarlist"
)
scene = self._to_int(config_info.get("points_scene")) or 7
allocation = client.play_mall_query_allocation(
uid=uid,
guid="",
cookie=cookie,
page_code=page_code,
source_id=source_id,
scene=scene,
deep=True,
)
payload = getattr(allocation, "payload", None)
discovered_sid = extract_points_center_sid(payload)
if discovered_sid:
sid_int = discovered_sid
sid_source = "queryAllocation"
except Exception as exc: # noqa: BLE001 自动发现失败时回退配置值
self._push_log(
"warning",
f"[{uid}] 自动发现兑换活动 SID 失败,回退配置值: {exc}",
)
if not sid_int:
self._mark_task(
worker_db,
task,
"failed",
"无法确定当前兑换活动 SID,请先刷新积分中心商品",
)
return
response = client.get_user_prize_records(uid=uid, cookie=cookie, sid=sid_int)
if response is None:
self._mark_task(worker_db, task, "error", "虎牙兑换记录接口无响应")
@@ -244,6 +287,8 @@ class GoodsMixin:
result = response.to_dict()
result["sid"] = sid_int
result["sid_source"] = sid_source
result["configured_sid"] = configured_sid
if response.status != 200:
self._mark_task(
worker_db,
@@ -277,16 +322,6 @@ class GoodsMixin:
account_info: dict,
config_info: dict,
):
sid = str(self.payload.get("sid") or config_info.get("sid") or "").strip()
if not sid:
self._mark_task(worker_db, task, "failed", "请先配置虎牙活动 SID")
return
sid_int = self._to_int(sid)
if not sid_int:
self._mark_task(worker_db, task, "failed", f"虎牙活动 SID 无效: {sid}")
return
uid = self._resolve_uid(account_info)
if not uid:
self._mark_task(worker_db, task, "failed", "无法从账号或 Cookie 解析 yyuid")
@@ -297,29 +332,36 @@ class GoodsMixin:
self._mark_task(worker_db, task, "failed", "账号 Cookie 为空")
return
client: Any = self._activity_client(uid, cookie)
response = client.get_act_prize_list(uid=uid, cookie=cookie, sid=sid_int)
if response is None:
self._mark_task(worker_db, task, "error", "虎牙商品列表接口无响应")
return
from .huya_collab_service import fetch_points_center_gifts
result = response.to_dict()
result["sid"] = sid_int
if response.status != 200:
self._mark_task(
worker_db,
task,
"failed",
response.msg or f"虎牙商品列表刷新失败: {response.status}",
result,
)
return
goods = [
item
for item in result.get("goods", [])
if item.get("product_id") and item.get("name")
sid_int = self._to_int(config_info.get("sid")) or 2203
page_code = (
str(config_info.get("points_page_code") or "").strip()
or "heping_baodian_credit_center_page"
)
source_id = (
str(config_info.get("points_source_id") or "").strip() or "yellowcarlist"
)
scene = self._to_int(config_info.get("points_scene")) or 7
page_id = self._to_int(config_info.get("points_page_id")) or None
tab_ids = [
int(part)
for part in str(config_info.get("points_tab_ids") or "")
.replace(",", ",")
.replace(" ", ",")
.split(",")
if part.isdigit()
]
goods, message = fetch_points_center_gifts(
uid,
cookie,
client=self._http_client(),
page_code=page_code,
source_id=source_id,
scene=scene,
page_id=page_id,
tab_ids=tab_ids or None,
)
now = datetime.now(UTC)
worker_db.query(HuyaGoodsSnapshot).filter(
HuyaGoodsSnapshot.account_id == account.id,
@@ -339,10 +381,26 @@ class GoodsMixin:
)
)
result = {
"source": "PlayMallUI.queryAllocation",
"sid": sid_int,
"page_code": page_code,
"source_id": source_id,
"scene": scene,
"goods": goods,
}
if not goods:
self._mark_task(worker_db, task, "failed", message, result)
return
account.status = "goods_refreshed"
account.updated_at = now
message = f"已刷新商品 {len(goods)} 个"
self._mark_task(worker_db, task, "success", message, {**result, "goods": goods})
self._mark_task(
worker_db,
task,
"success",
f"已刷新积分中心商品 {len(goods)} 个",
result,
)
def _execute_exchange_goods(
self,
@@ -352,18 +410,8 @@ class GoodsMixin:
account_info: dict,
config_info: dict,
):
sid = str(self.payload.get("sid") or config_info.get("sid") or "").strip()
if not sid:
self._mark_task(worker_db, task, "failed", "请先配置虎牙活动 SID")
return
sid_int = self._to_int(sid)
if not sid_int:
self._mark_task(worker_db, task, "failed", f"虎牙活动 SID 无效: {sid}")
return
product_id = self._to_int(self.payload.get("product_id"))
if not product_id:
spu_id = normalize_spu_id(self.payload.get("product_id"))
if not spu_id:
self._mark_task(worker_db, task, "failed", "请选择兑换商品")
return
@@ -377,20 +425,6 @@ class GoodsMixin:
self._mark_task(worker_db, task, "failed", "账号 Cookie 为空")
return
snapshot = (
worker_db.query(HuyaGoodsSnapshot)
.filter(
HuyaGoodsSnapshot.account_id == account.id,
HuyaGoodsSnapshot.sid == sid_int,
HuyaGoodsSnapshot.product_id == str(product_id),
)
.first()
)
product_name = str(
self.payload.get("product_name")
or (snapshot.name if snapshot else "")
or product_id
)
scheduled_at = self._parse_scheduled_time(self.payload.get("scheduled_at"))
if self.payload.get("scheduled_at") and scheduled_at is None:
self._mark_task(worker_db, task, "failed", "定时兑换时间格式无效")
@@ -403,92 +437,153 @@ class GoodsMixin:
self._mark_task(worker_db, task, "stopped", "兑换任务已停止")
return
client: Any = self._activity_client(uid, cookie)
detail = client.get_act_prize_detail(
uid=uid, cookie=cookie, sid=sid_int, pid=product_id
)
if detail is None:
self._mark_task(worker_db, task, "error", "虎牙兑换详情接口无响应")
pid = self._to_int(config_info.get("room_pid"))
if not pid:
self._mark_task(worker_db, task, "failed", "请先配置虎牙直播间 ID")
return
detail_result = detail.to_dict()
prize = detail.prize
if detail.status != 200 or prize is None:
pay_channel = self._normalize_pay_channel(
self.payload.get("pay_channel") or config_info.get("pay_channel")
)
client: Any = self._http_client()
prepared = prepare_exchange(
client,
uid=uid,
cookie=cookie,
pid=pid,
spu_id=spu_id,
pay_channel=pay_channel,
stopped=self._stop.is_set,
log=lambda message: self._push_log("info", message),
)
common = {
"spu_id": spu_id,
"product_name": (prepared.target.get("name") if prepared.target else "")
or str(self.payload.get("product_name") or spu_id),
"scheduled_at": scheduled_at.isoformat() if scheduled_at else "",
"executed_at": datetime.now(UTC).isoformat(),
"redeem_state": prepared.state.to_dict() if prepared.state else None,
"target": prepared.target,
"exchange_plan": prepared.plan,
}
if prepared.status == EXCHANGE_STATUS_EXCHANGED:
result = {
**common,
"exchange_mode": "points_only",
"exchange_order": prepared.order,
"payment_status": "not_required",
"payment_status_label": "积分直兑",
}
account.status = "goods_exchanged"
account.updated_at = datetime.now(UTC)
if prepared.state:
account.points = prepared.state.user_score
self._mark_task(worker_db, task, "success", prepared.message, result)
return
if prepared.status != EXCHANGE_STATUS_PAYMENT_REQUIRED or not prepared.purchase:
self._mark_task(
worker_db,
task,
"failed",
detail.msg or f"虎牙兑换详情获取失败: {detail.status}",
detail_result,
prepared.message,
{**common, **prepared.detail},
)
return
now_ts = int(time.time())
if prize.isCanExchange == 0:
self._mark_task(
worker_db, task, "failed", "该商品当前不可兑换", detail_result
)
return
if prize.isShowNum and prize.leftNum <= 0:
self._mark_task(worker_db, task, "failed", "该商品库存不足", detail_result)
return
if prize.exchangeStartTime and now_ts < prize.exchangeStartTime:
self._mark_task(
worker_db, task, "failed", "该商品尚未开始兑换", detail_result
)
return
if prize.exchangeEndTime and now_ts > prize.exchangeEndTime:
self._mark_task(
worker_db, task, "failed", "该商品兑换已结束", detail_result
)
return
score = client.query_user_score(uid=uid, cookie=cookie, sid=sid_int)
if score is None:
self._mark_task(
worker_db, task, "error", "兑换前积分查询无响应", detail_result
)
return
if score.status != 200 or score.available_score < prize.newScore:
detail_result["score"] = score.to_dict()
self._mark_task(worker_db, task, "failed", "可用积分不足", detail_result)
return
response = client.score_exchange_prize(
uid=uid, cookie=cookie, sid=sid_int, pid=product_id
)
if response is None:
self._mark_task(worker_db, task, "error", "虎牙兑换接口无响应")
return
result = response.to_dict()
result.update(
{
"sid": sid_int,
"product_id": str(product_id),
"product_name": product_name,
"scheduled_at": scheduled_at.isoformat() if scheduled_at else "",
"executed_at": datetime.now(UTC).isoformat(),
"goods": snapshot.raw if snapshot else None,
"prize_detail": detail_result,
}
purchase = prepared.purchase
pay_targets = resolve_pay_targets(purchase.pay_url)
result = {
**common,
"exchange_mode": "buy_points_then_exchange",
"points_package": prepared.plan,
"points_purchase": {
"order_id": purchase.order_id,
"app_order_id": purchase.app_order_id,
"pay_order_id": purchase.pay_order_id,
"product_name": purchase.product_name,
"count": purchase.count,
"unit_price": purchase.unit_price,
"amount": purchase.amount,
"supplier_name": purchase.supplier_name,
"order": purchase.order_raw,
"pay": purchase.pay_raw,
},
"pay_url": purchase.pay_url,
"pay_qr_code": pay_targets.qr_code,
"pay_cashier_url": pay_targets.cashier_url,
"pay_qr_path": pay_qr_path(task.id) if task is not None else "",
"payment_status": "pending",
"payment_status_label": "等待支付积分包",
"payment_paid": False,
}
account.status = "exchange_payment_pending"
account.updated_at = datetime.now(UTC)
self._update_task_progress(
worker_db,
task,
"running",
f"积分不足,需先购买 {prepared.plan['name'] if prepared.plan else '积分包'},等待扫码支付",
result,
)
if response.status != 200:
payment_status, _payment_order = self._wait_recharge_payment(
client=client,
uid=uid,
guid="",
cookie=cookie,
order_id=purchase.order_id,
sid=self._to_int(config_info.get("sid")) or 2203,
result=result,
)
if payment_status == "paid":
completed = complete_exchange_after_payment(
client,
uid=uid,
cookie=cookie,
pid=pid,
spu_id=spu_id,
)
result.update(
{
"post_payment_redeem_state": (
completed.state.to_dict() if completed.state else None
),
"exchange_order": completed.order,
"target": completed.target or prepared.target,
}
)
if completed.status == EXCHANGE_STATUS_EXCHANGED:
account.status = "goods_exchanged"
account.updated_at = datetime.now(UTC)
if completed.state:
account.points = completed.state.user_score
self._mark_task(
worker_db,
task,
"success",
f"积分包支付成功,{completed.message}",
result,
)
else:
self._mark_task(
worker_db,
task,
"failed",
f"积分包支付成功,但兑换失败:{completed.message}",
result,
)
return
if payment_status == "stopped":
self._mark_task(
worker_db,
task,
"failed",
response.msg or f"虎牙兑换失败: {response.status}",
"stopped",
"积分包支付监听已停止",
result,
)
return
post_score = client.query_user_score(uid=uid, cookie=cookie, sid=sid_int)
post_records = client.get_user_prize_records(
uid=uid, cookie=cookie, sid=sid_int
self._mark_task(
worker_db,
task,
"timeout",
f"积分包支付超时,订单 #{purchase.order_id} 仍未完成",
result,
)
if post_score is not None:
result["post_exchange_score"] = post_score.to_dict()
account.points = post_score.available_score
if post_records is not None:
result["post_exchange_records"] = post_records.to_dict()
account.status = "goods_exchanged"
account.updated_at = datetime.now(UTC)
message = response.msg or f"兑换成功: {product_name}"
self._mark_task(worker_db, task, "success", message, result)