Files
live-hub-py/web/backend/services/huya_runner_recharge.py
T
yml2213 905a7ef22c fix(huya): 查询积分改走积分中心新接口(含成长等级)
旧活动接口 webActUI.getUserScore 对任何 sid 都返回 603「活动已结束」
(接口已下线,与工作台 SID/活动配置无关),刷新积分只会报错。

- 协议层新增两个 RPC(WSS + HTTP 双侧同签名):
  shopMiddleUI.getRedeemNeedGoods(tag2=当前积分、tag3=兑换门槛,
  兑换商品号 POINTS_REDEEM_GOODS_NUM=8315982,传错直接 500)
  与 GameBizUI.getGrowLevelUserInfo(成长值/成长等级/昵称/头像)
- 新增结构体 GetRedeemNeedGoodsReq / GetGrowLevelUserInfoReq,
  与 HuyaEliteWssSession 透传
- 服务层 _execute_query_points 主源改为积分中心接口,结果带
  points/redeem_threshold/grow_level/grow_value,并同步 account.points;
  接口无响应时才回退旧活动接口并如实报告其 603 返回
- 开通支付成功后的积分刷新同样改走积分中心接口
- 前端任务页积分结果优先读 points(兼容旧记录的 available_score),
  并在有成长等级时一并展示
- 新增 tests/test_huya_points_query.py:真机响应字节回归 + 请求参数
  字面量断言 + 服务层积分来源/回退语义(已做反向验证)

真机验证:HTTP 通道取到积分 280 / 门槛 20 / 成长等级 3
2026-10-15 15:32:46 +08:00

530 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""虎牙任务执行器:充值(由 huya_runner.py 按功能域拆分)。"""
from __future__ import annotations
import time
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from sqlalchemy.orm import Session
from core.huya.wss_client import SHOP_APP_SCENE
from ..models import HuyaAccount, HuyaRechargeGoodsSnapshot, HuyaTask
from .huya_shop_flow import run_open_flow
HUYA_RECHARGE_ACT_ID = 25135
# shopAppInfo.scene(2026-10-13 积分中心抓包实测 7;订单侧 scene 由
# core.huya.wss_client.ORDER_SCENE 决定,两者不是同一个值)
HUYA_RECHARGE_APP_SCENE = SHOP_APP_SCENE
HUYA_PAYMENT_POLL_SECONDS = 180
HUYA_PAYMENT_POLL_INTERVAL = 3
HUYA_RECHARGE_EXTRA_PRODUCTS = [
{
"spu_id": "hy-5879340",
"name": "精英宝典",
"task_name": "开通精英宝典",
"description": "得300积分丨解锁道具兑换权益",
"sort": 0,
},
]
from .huya_runner_core import HUYA_RECHARGE_SOURCE_ID
class RechargeMixin:
"""充值域:充值商品、下单与到账轮询。"""
if TYPE_CHECKING:
_stop: Any
payload: dict
@staticmethod
def _to_int(value: Any) -> int: ...
def _resolve_uid(self, account_info: dict) -> int: ...
def _push_log(self, level: str, message: str) -> None: ...
def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ...
def _open_wss_session(self, uid: int, cookie: str, kind: str): ...
def _shop_client(self, uid: int, cookie: str): ...
@staticmethod
def _format_local_time(timestamp: int) -> str: ...
@staticmethod
def _huya_order_status_label(status: int) -> str:
from core.huya.shop_structs import OrderStatus
labels = {
OrderStatus.DEPOSIT_WAIT_PAY: "待支付",
OrderStatus.DEPOSIT_PAID: "已支付",
OrderStatus.WAIT_DELIVER: "待发货",
OrderStatus.WAIT_RECEIVE: "待收货",
OrderStatus.FINISHED: "已完成",
OrderStatus.FINISHED_CLOSED: "已关闭",
OrderStatus.CANCELLED: "已取消",
OrderStatus.BALANCE_WAIT_PAY: "尾款待支付",
OrderStatus.CANCELLED_BALANCE_EXPIRED: "尾款超时取消",
}
return labels.get(int(status or 0), str(status or "未知"))
@staticmethod
def _is_huya_order_paid(order) -> bool:
from core.huya.shop_structs import OrderStatus
paid_statuses = {
OrderStatus.DEPOSIT_PAID,
OrderStatus.WAIT_DELIVER,
OrderStatus.WAIT_RECEIVE,
OrderStatus.FINISHED,
OrderStatus.FINISHED_CLOSED,
}
return (
int(getattr(order, "payTime", 0) or 0) > 0
or int(getattr(order, "orderStatus", 0) or 0) in paid_statuses
)
def _wait_recharge_payment(
self,
client: Any,
uid: int,
guid: str,
cookie: str,
order_id: int,
sid: int,
result: dict,
) -> tuple[str, dict | None]:
deadline = time.time() + HUYA_PAYMENT_POLL_SECONDS
order_id_text = str(order_id)
last_order = None
while not self._stop.is_set() and time.time() < deadline:
detail = client.order_detail(
uid=uid, guid=guid, cookie=cookie, order_id=order_id, timeout=10.0
)
checked_at = datetime.now(UTC).isoformat()
order = getattr(detail, "order", None) if detail is not None else None
if order is not None and (
not getattr(order, "orderId", 0)
or str(getattr(order, "orderId", "")) == order_id_text
):
last_order = order.to_dict()
status = int(getattr(order, "orderStatus", 0) or 0)
result.update(
{
"payment_checked_at": checked_at,
"payment_order": last_order,
"payment_order_status": status,
"payment_order_status_label": self._huya_order_status_label(
status
),
}
)
if self._is_huya_order_paid(order):
result.update(
{
"payment_status": "paid",
"payment_status_label": "已支付",
"payment_paid": True,
"payment_paid_at": checked_at,
}
)
return "paid", last_order
else:
# 兼容旧 HTTP 网关不提供 orderDetailV5 的情况。
resp = client.query_user_order_list(
uid=uid,
guid=guid,
cookie=cookie,
offset=0,
page_size=10,
order_type=1,
status=0,
timeout=10.0,
)
for legacy_order in getattr(resp, "orders", []) if resp else []:
if str(getattr(legacy_order, "orderId", "")) != order_id_text:
continue
last_order = legacy_order.to_dict()
status = int(getattr(legacy_order, "orderStatus", 0) or 0)
result.update(
{
"payment_checked_at": checked_at,
"payment_order": last_order,
"payment_order_status": status,
"payment_order_status_label": self._huya_order_status_label(
status
),
}
)
if self._is_huya_order_paid(legacy_order):
result.update(
{
"payment_status": "paid",
"payment_status_label": "已支付",
"payment_paid": True,
"payment_paid_at": checked_at,
}
)
return "paid", last_order
result.setdefault("payment_checked_at", checked_at)
if self._stop.wait(HUYA_PAYMENT_POLL_INTERVAL):
break
result.update(
{
"payment_status": "timeout" if not self._stop.is_set() else "stopped",
"payment_status_label": "等待支付超时"
if not self._stop.is_set()
else "已停止监听",
"payment_paid": False,
"payment_timeout_seconds": HUYA_PAYMENT_POLL_SECONDS,
}
)
if last_order:
result["payment_order"] = last_order
return result["payment_status"], last_order
@staticmethod
def _normalize_pay_channel(value) -> str:
text = str(value or "").strip()
lowered = text.lower()
if lowered in {"weixin", "wx", "wechat", "微信"}:
return "Weixin"
return "Zfb"
@staticmethod
def _pay_channel_label(value: str) -> str:
return "微信" if value == "Weixin" else "支付宝"
@staticmethod
def _recharge_price_text(price: int | None) -> str:
if not price:
return ""
return f"{price / 100:.2f}元"
def _execute_refresh_recharge_goods(
self,
worker_db: Session,
task: HuyaTask,
account: HuyaAccount,
account_info: dict,
config_info: dict,
):
pid = self._to_int(config_info.get("room_pid"))
if not pid:
self._mark_task(worker_db, task, "failed", "请先配置虎牙直播间 ID")
return
uid = self._resolve_uid(account_info)
if not uid:
self._mark_task(worker_db, task, "failed", "无法从账号或 Cookie 解析 yyuid")
return
cookie = account_info.get("cookie") or ""
if not cookie:
self._mark_task(worker_db, task, "failed", "账号 Cookie 为空")
return
client: Any = self._shop_client(uid, cookie)
task_resp = client.get_act_task_detail(
uid=uid, cookie=cookie, act_id=HUYA_RECHARGE_ACT_ID
)
if task_resp is None:
self._mark_task(worker_db, task, "error", "虎牙充值任务详情接口无响应")
return
task_result = task_resp.to_dict()
if task_resp.status != 200:
self._mark_task(
worker_db,
task,
"failed",
task_resp.msg or f"虎牙充值任务详情获取失败: {task_resp.status}",
task_result,
)
return
candidates: list[dict] = []
seen: set[str] = set()
def add_candidate(item: dict):
spu_id = str(item.get("spu_id") or "").strip()
if not spu_id or spu_id in seen:
return
seen.add(spu_id)
candidates.append(item)
for item in HUYA_RECHARGE_EXTRA_PRODUCTS:
add_candidate(dict(item))
for index, item in enumerate(task_result.get("tasks", []), start=1):
if int(item.get("task_type") or 0) != 67:
continue
add_candidate(
{
"spu_id": item.get("spu_id") or "",
"name": item.get("name") or "",
"task_id": str(item.get("task_id") or ""),
"task_name": item.get("name") or "",
"description": item.get("description") or "",
"icon": item.get("icon") or "",
"task_url": item.get("task_url") or "",
"prizes": item.get("prizes") or [],
"sort": index,
}
)
if not candidates:
self._mark_task(
worker_db, task, "failed", "未从活动任务中发现充值商品", task_result
)
return
now = datetime.now(UTC)
goods: list[dict] = []
failed: list[dict] = []
for candidate in candidates:
spu_id = candidate["spu_id"]
detail_resp = client.get_goods_info(
uid=uid,
guid="",
cookie=cookie,
pid=pid,
spu_id=spu_id,
sku_id=0,
game_id="",
source_id=HUYA_RECHARGE_SOURCE_ID,
app_scene=HUYA_RECHARGE_APP_SCENE,
)
if detail_resp is None:
failed.append({"spu_id": spu_id, "message": "商品详情接口无响应"})
continue
detail = detail_resp.to_dict()
if detail_resp.code != 200 or not detail.get("sku_id"):
failed.append(
{
"spu_id": spu_id,
"message": detail_resp.message
or f"商品详情获取失败: {detail_resp.code}",
"detail": detail,
}
)
continue
item = {
**candidate,
**detail,
"spu_id": detail.get("spu_id") or spu_id,
"sku_id": str(detail.get("sku_id") or ""),
"name": detail.get("name") or candidate.get("name") or spu_id,
"description": detail.get("description")
or candidate.get("description")
or "",
"icon": detail.get("icon") or candidate.get("icon") or "",
"task_id": candidate.get("task_id") or "",
"task_name": candidate.get("task_name") or candidate.get("name") or "",
"raw_order": int(candidate.get("sort") or 0),
}
goods.append(item)
worker_db.query(HuyaRechargeGoodsSnapshot).filter(
HuyaRechargeGoodsSnapshot.account_id == account.id,
HuyaRechargeGoodsSnapshot.sid
== (self._to_int(config_info.get("sid")) or 2203),
).delete(synchronize_session=False)
for item in goods:
worker_db.add(
HuyaRechargeGoodsSnapshot(
account_id=account.id,
sid=self._to_int(config_info.get("sid")) or 2203,
spu_id=item["spu_id"],
sku_id=item["sku_id"],
name=item["name"],
price=item.get("price") or None,
stock=item.get("stock") or None,
buy_limit=item.get("buy_limit") or None,
icon=item.get("icon") or "",
description=item.get("description") or "",
task_id=item.get("task_id") or "",
task_name=item.get("task_name") or "",
raw=item,
updated_at=now,
)
)
account.status = "recharge_goods_refreshed"
account.updated_at = now
message = f"已刷新充值商品 {len(goods)} 个"
if failed:
message += f",失败 {len(failed)} 个"
result = {
"act_id": HUYA_RECHARGE_ACT_ID,
"goods_count": len(goods),
"failed_count": len(failed),
"goods": goods,
"failed": failed,
"task_detail": task_result,
}
self._mark_task(
worker_db, task, "success" if goods else "failed", message, result
)
def _execute_create_recharge_order(
self,
worker_db: Session,
task: HuyaTask,
account: HuyaAccount,
account_info: dict,
config_info: dict,
):
pid = self._to_int(config_info.get("room_pid"))
if not pid:
self._mark_task(worker_db, task, "failed", "请先配置虎牙直播间 ID")
return
spu_id = str(self.payload.get("spu_id") or "").strip()
if not spu_id:
self._mark_task(worker_db, task, "failed", "请选择充值商品")
return
count = self._to_int(self.payload.get("count")) or 1
count = max(1, min(count, 999))
pay_channel = self._normalize_pay_channel(
self.payload.get("pay_channel") or config_info.get("pay_channel")
)
uid = self._resolve_uid(account_info)
if not uid:
self._mark_task(worker_db, task, "failed", "无法从账号或 Cookie 解析 yyuid")
return
cookie = account_info.get("cookie") or ""
if not cookie:
self._mark_task(worker_db, task, "failed", "账号 Cookie 为空")
return
snapshot = (
worker_db.query(HuyaRechargeGoodsSnapshot)
.filter(
HuyaRechargeGoodsSnapshot.account_id == account.id,
HuyaRechargeGoodsSnapshot.sid
== (self._to_int(config_info.get("sid")) or 2203),
HuyaRechargeGoodsSnapshot.spu_id == spu_id,
)
.first()
)
payload_sku_id = self._to_int(self.payload.get("sku_id"))
sku_id = payload_sku_id or self._to_int(snapshot.sku_id if snapshot else "")
product_name = str(
self.payload.get("product_name")
or (snapshot.name if snapshot else "")
or spu_id
)
unit_price = int(snapshot.price or 0) if snapshot else 0
client: Any = self._shop_client(uid, cookie)
outcome = run_open_flow(
client,
uid=uid,
cookie=cookie,
pid=pid,
spu_id=spu_id,
sku_id=sku_id,
count=count,
pay_channel=pay_channel,
snapshot_price=unit_price,
log=lambda message: self._push_log("info", message),
stopped=self._stop.is_set,
)
detail = outcome.detail
if not outcome.ok:
self._mark_task(
worker_db,
task,
"error" if "无响应" in outcome.message else "failed",
outcome.message,
{
"goods": detail,
"supplier_name": outcome.supplier_name,
"promotion": outcome.promotion,
"promotion_info": outcome.promotion_info,
},
)
return
sku_id = outcome.sku_id or sku_id
product_name = outcome.product_name or product_name
unit_price = outcome.unit_price or unit_price
order_result = outcome.order_raw
pay_result = outcome.pay_raw
amount = outcome.amount
result = {
"spu_id": spu_id,
"sku_id": sku_id,
"product_name": product_name,
"count": count,
"unit_price": unit_price,
"amount": amount,
"amount_text": self._recharge_price_text(amount),
"pay_channel": pay_channel,
"pay_channel_label": self._pay_channel_label(pay_channel),
"order_id": outcome.order_id,
"app_order_id": outcome.app_order_id,
"pay_order_id": outcome.pay_order_id,
"pay_url": outcome.pay_url,
"payment_status": "pending",
"payment_status_label": "等待支付",
"payment_paid": False,
"goods": detail,
"order": order_result,
"pay": pay_result,
"supplier_name": outcome.supplier_name,
"promotion": outcome.promotion,
"promotion_info": outcome.promotion_info,
}
account.status = "recharge_order_created"
account.updated_at = datetime.now(UTC)
message = f"{product_name} x{count} {self._pay_channel_label(pay_channel)} {result['amount_text']}"
self._update_task_progress(
worker_db, task, "running", f"{message},等待扫码支付", result
)
self._push_log(
"info", f"[{uid}] 已生成虎牙支付二维码,开始监听订单 {outcome.order_id}"
)
payment_status, payment_order = self._wait_recharge_payment(
client=client,
uid=uid,
guid="",
cookie=cookie,
order_id=outcome.order_id,
sid=self._to_int(config_info.get("sid")) or 2203,
result=result,
)
account.updated_at = datetime.now(UTC)
if payment_status == "paid":
account.status = "recharge_paid"
# 支付后刷新积分:旧活动接口 getUserScore 已下线(603 活动已结束),
# 改走积分中心 getRedeemNeedGoods(tag2 = 当前积分)。
post_redeem = client.get_redeem_need_goods(uid=uid, guid="", cookie=cookie)
if post_redeem is not None and post_redeem.code == 200:
result["post_payment_points"] = self._to_int(post_redeem.value("tag2"))
account.points = result["post_payment_points"]
paid_message = f"支付成功: {product_name} x{count} {result['amount_text']}"
if payment_order and payment_order.get("pay_time"):
paid_message += f",支付时间 {self._format_local_time(int(payment_order['pay_time']) // 1000)}"
self._mark_task(worker_db, task, "success", paid_message, result)
return
if payment_status == "stopped":
account.status = "recharge_order_created"
self._mark_task(
worker_db, task, "stopped", f"{message},已停止监听支付", result
)
return
account.status = "recharge_order_created"
timeout_message = f"{message},{result['payment_status_label']}"
self._mark_task(worker_db, task, "timeout", timeout_message, result)