"""虎牙任务执行器：充值（由 huya_runner.py 按功能域拆分）。""" from __future__ import annotations import time from datetime import datetime, timezone from typing import Any, TYPE_CHECKING from sqlalchemy.orm import Session from core.huya import HuyaHttpClient from ..models import HuyaAccount, HuyaRechargeGoodsSnapshot, HuyaTask HUYA_RECHARGE_ACT_ID = 25135 HUYA_RECHARGE_SCENE = 4 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 if TYPE_CHECKING: from .huya_runner import HuyaBatchRunner 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: ... @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, 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: 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, ) checked_at = datetime.now(timezone.utc).isoformat() if resp is not None and getattr(resp, "orders", None): for order in resp.orders: if str(getattr(order, "orderId", "")) != order_id_text: continue 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 break else: result["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 = HuyaHttpClient( logger=lambda msg: self._push_log("info", f"[{uid}] {msg}") ) 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(timezone.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="0", source_id=HUYA_RECHARGE_SOURCE_ID, scene=HUYA_RECHARGE_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).delete(synchronize_session=False) for item in goods: worker_db.add( HuyaRechargeGoodsSnapshot( 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.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 = HuyaHttpClient( logger=lambda msg: self._push_log("info", f"[{uid}] {msg}") ) detail_resp = client.get_goods_info( uid=uid, guid="", cookie=cookie, pid=pid, spu_id=spu_id, sku_id=sku_id or 0, game_id="0", source_id=HUYA_RECHARGE_SOURCE_ID, scene=HUYA_RECHARGE_SCENE, ) if detail_resp is None: self._mark_task(worker_db, task, "error", "虎牙充值商品详情接口无响应") return detail = detail_resp.to_dict() if detail_resp.code != 200: self._mark_task( worker_db, task, "failed", detail_resp.message or f"虎牙充值商品详情获取失败: {detail_resp.code}", detail, ) return sku_id = int(detail.get("sku_id") or sku_id or 0) product_name = detail.get("name") or product_name unit_price = int(detail.get("price") or unit_price or 0) if not sku_id: self._mark_task( worker_db, task, "failed", "充值商品缺少 SKU，请先刷新充值商品列表", detail, ) return order_resp = client.create_order( uid=uid, guid="", cookie=cookie, pid=pid, spu_id=spu_id, sku_id=sku_id, item_count=count, source_id=HUYA_RECHARGE_SOURCE_ID, game_id="0", scene=HUYA_RECHARGE_SCENE, order_type=6, ) if order_resp is None: self._mark_task(worker_db, task, "error", "虎牙下单接口无响应") return order_result = order_resp.to_dict() if order_resp.code != 200 or not order_resp.orderId: self._mark_task( worker_db, task, "failed", order_resp.message or f"虎牙下单失败: {order_resp.code}", {"goods": detail, "order": order_result}, ) return pay_resp = client.pay_order_submit( uid=uid, guid="", cookie=cookie, order_id=order_resp.orderId, pay_type=pay_channel, pid=pid, source_id=HUYA_RECHARGE_SOURCE_ID, scene=HUYA_RECHARGE_SCENE, item_count=count, ) if pay_resp is None: self._mark_task( worker_db, task, "error", "虎牙支付接口无响应", {"goods": detail, "order": order_result}, ) return pay_result = pay_resp.to_dict() if pay_resp.code != 200 or not pay_resp.payUrl: self._mark_task( worker_db, task, "failed", pay_resp.message or f"虎牙支付二维码生成失败: {pay_resp.code}", {"goods": detail, "order": order_result, "pay": pay_result}, ) return amount = int(pay_resp.amount or unit_price * count or 0) 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": order_resp.orderId, "app_order_id": pay_resp.appOrderId, "pay_order_id": pay_resp.payOrderId, "pay_url": pay_resp.payUrl, "payment_status": "pending", "payment_status_label": "等待支付", "payment_paid": False, "goods": detail, "order": order_result, } account.status = "recharge_order_created" account.updated_at = datetime.now(timezone.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}] 已生成虎牙支付二维码，开始监听订单 {order_resp.orderId}" ) payment_status, payment_order = self._wait_recharge_payment( client=client, uid=uid, guid="", cookie=cookie, order_id=order_resp.orderId, result=result, ) account.updated_at = datetime.now(timezone.utc) if payment_status == "paid": account.status = "recharge_paid" 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, "success", timeout_message, result)