diff --git a/web/backend/services/huya_runner.py b/web/backend/services/huya_runner.py index 518f7ba..559fc85 100644 --- a/web/backend/services/huya_runner.py +++ b/web/backend/services/huya_runner.py @@ -1,1455 +1,27 @@ -"""虎牙任务批次执行器。""" +"""虎牙任务批次执行器（入口聚合；功能域已拆分到 huya_runner_*.py）。""" + +from __future__ import annotations -import asyncio -import time -import threading from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import datetime, timezone -from typing import Optional -from loguru import logger -from sqlalchemy.orm import Session, joinedload +from sqlalchemy.orm import joinedload -from core.huya import HuyaHttpClient -from core.huya.cookie_utils import normalize_huya_cookie from ..database import SessionLocal -from ..models import HuyaAccount, HuyaGoodsSnapshot, HuyaRechargeGoodsSnapshot, HuyaTask -from .huya_service import HUYA_CONFIG_FIELDS, cookie_value, ensure_huya_config, huya_config_value +from datetime import datetime, timezone +from ..models import HuyaAccount, HuyaTask +from core.huya.cookie_utils import normalize_huya_cookie +from .huya_service import HUYA_CONFIG_FIELDS, ensure_huya_config, huya_config_value +from .huya_runner_core import HuyaBatchRunnerCore, huya_batch_registry # noqa: F401 (huya_batch_registry 供 routers 重导出) +from .huya_runner_bind import BindMixin +from .huya_runner_goods import GoodsMixin +from .huya_runner_recharge import RechargeMixin -HUYA_RECHARGE_ACT_ID = 25135 -HUYA_RECHARGE_SOURCE_ID = "yellowcarlist" -HUYA_RECHARGE_SCENE = 4 -HUYA_PAYMENT_POLL_SECONDS = 180 -HUYA_PAYMENT_POLL_INTERVAL = 3 -HUYA_BIND_ROLE_POLL_SECONDS = 180 -HUYA_BIND_ROLE_POLL_INTERVAL = 3 -HUYA_BIND_ZT_UUID = "b02faae1" -HUYA_BIND_ROOM_ID = "30596253" -HUYA_RECHARGE_EXTRA_PRODUCTS = [ - { - "spu_id": "hy-5879340", - "name": "精英宝典", - "task_name": "开通精英宝典", - "description": "得300积分丨解锁道具兑换权益", - "sort": 0, - }, -] - -class HuyaBatchRunner: - """批量执行虎牙任务，通过队列推送实时日志。""" - - def __init__( - self, - db: Session, - batch_id: str, - task_type: str, - payload: Optional[dict] = None, - log_queue: Optional[asyncio.Queue] = None, - loop: Optional[asyncio.AbstractEventLoop] = None, - concurrency: int = 3, - ): - self.db = db - self.batch_id = batch_id - self.task_type = task_type - self.payload = payload or {} - self.log_queue = log_queue - self.loop = loop - self.concurrency = max(1, min(concurrency, 10)) - self._stop = threading.Event() - self._counter_lock = threading.Lock() - self._started = 0 - - def stop(self): - self._stop.set() - - def _push_log(self, level: str, message: str): - if level == "result": - try: - huya_batch_registry.mark_finished(self.batch_id) - except NameError: - pass - if level != "result" and message: - log_func = getattr(logger, level, logger.info) - log_func(f"[huya] {message}") - if self.log_queue and self.loop: - asyncio.run_coroutine_threadsafe( - self.log_queue.put({"level": level, "message": message}), - self.loop, - ) - - @staticmethod - def _account_name(account_info: dict) -> str: - return ( - account_info.get("nickname") - or account_info.get("username") - or account_info.get("uid") - or f"#{account_info.get('account_id')}" - ) - - @staticmethod - def _to_int(value) -> int: - text = str(value or "").strip() - return int(text) if text.isdigit() else 0 - - def _resolve_uid(self, account_info: dict) -> int: - cookie = account_info.get("cookie") or "" - return ( - self._to_int(account_info.get("yyuid")) - or self._to_int(account_info.get("uid")) - or self._to_int(cookie_value(cookie, "yyuid")) - or self._to_int(cookie_value(cookie, "udb_uid")) - ) - - @staticmethod - def _role_name(bind_status) -> str: - account_data = bind_status.accountData - return account_data.gameRole.roleName or "" - - @staticmethod - def _has_bind_role(bind_status) -> bool: - return bool(bind_status and HuyaBatchRunner._role_name(bind_status)) - - @staticmethod - def _bind_role_result(bind_status) -> dict: - account_data = bind_status.accountData - game_account = account_data.gameAccount - game_role = account_data.gameRole - return { - "game_title": bind_status.gameName, - "role_name": HuyaBatchRunner._role_name(bind_status), - "change_bind_day": bind_status.changeBindDay, - "is_bind_account": account_data.isBindAcount, - "is_bind_role": account_data.isBindRole, - "is_need_act_check": account_data.isNeedActCheck, - "change_bind_time": account_data.changBindTime, - "game_account": game_account.to_dict(), - "game_role": game_role.to_dict(), - } - - @staticmethod - def _role_channel(bind_status) -> str: - game_role = bind_status.accountData.gameRole - parts = [game_role.platName, game_role.areaName] - return " / ".join(part for part in parts if part) - - @staticmethod - def _format_local_time(timestamp: int) -> str: - if not timestamp: - return "" - return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S") - - @staticmethod - def _parse_scheduled_time(value) -> datetime | None: - text = str(value or "").strip() - if not text: - return None - try: - normalized = text.replace("Z", "+00:00") - dt = datetime.fromisoformat(normalized) - except ValueError: - return None - if dt.tzinfo is None: - return dt.astimezone() - return dt - - def _wait_until(self, when: datetime, uid: int) -> bool: - target = when.timestamp() - local_text = self._format_local_time(int(target)) - self._push_log("info", f"[{uid}] 定时兑换等待到 {local_text}") - while not self._stop.is_set(): - remaining = target - time.time() - if remaining <= 0: - return True - time.sleep(min(0.2, max(0.02, remaining))) - return False - - @classmethod - def _bind_change_state(cls, bind_status) -> dict: - account_data = bind_status.accountData - is_bound = bool(account_data.isBindAcount and account_data.isBindRole) - change_time = int(account_data.changBindTime or 0) - now = int(datetime.now(timezone.utc).timestamp()) - can_change = not is_bound or not change_time or change_time <= now - return { - "is_bound": is_bound, - "can_change_bind": can_change, - "change_bind_time": change_time, - "change_available_at": cls._format_local_time(change_time), - "change_bind_day": int(bind_status.changeBindDay or 0), - } - - @classmethod - def _bind_ready_result(cls, bind_status) -> dict: - role_info = cls._bind_role_result(bind_status) - return { - **role_info, - **cls._bind_change_state(bind_status), - "bind_status": bind_status.to_dict(), - "bind_ready_for_confirm": bool(role_info["role_name"]), - "bind_phase": "role_ready" if role_info["role_name"] else "waiting_role", - } - - def _resolve_bind_status( - self, - client: HuyaHttpClient, - uid: int, - cookie: str, - b_act_id_int: int, - ): - """按活动页逻辑解析绑定状态，优先返回含角色的状态。""" - outer_status = client.check_user_bind_game_account( - uid=uid, - cookie=cookie, - b_act_id=b_act_id_int, - is_use_outer_act_id=1, - ) - if outer_status is None: - return None, {} - - query_result = { - "bind_status_source": "outer", - "outer_bind_status": outer_status.to_dict(), - } - if outer_status.status != 200: - return outer_status, query_result - - chosen_status = outer_status - account_data = outer_status.accountData - should_check_inner = ( - not self._has_bind_role(outer_status) - and bool(account_data.isNeedActCheck or not account_data.isBindAcount or not account_data.isBindRole) - ) - if should_check_inner: - inner_status = client.check_user_bind_game_account( - uid=uid, - cookie=cookie, - b_act_id=b_act_id_int, - is_use_outer_act_id=0, - ) - if inner_status is not None: - query_result["inner_bind_status"] = inner_status.to_dict() - if inner_status.status == 200 and self._has_bind_role(inner_status): - chosen_status = inner_status - query_result["bind_status_source"] = "inner" - - query_result["bind_status"] = chosen_status.to_dict() - return chosen_status, query_result - - def _apply_role_to_account(self, account: HuyaAccount, bind_status, status: str): - role_name = self._role_name(bind_status) - account.status = status - account.game_name = role_name or account.game_name - account.game_channel = self._role_channel(bind_status) or account.game_channel - account.updated_at = datetime.now(timezone.utc) - - @staticmethod - def _bind_redirect_url(config_info: dict) -> str: - room_pid = str(config_info.get("room_pid") or "").strip() - if not room_pid: - return "" - return ( - f"https://zt.huya.com/{HUYA_BIND_ZT_UUID}/pc/index.html" - f"?sourceId={HUYA_RECHARGE_SOURCE_ID}" - f"&pid={room_pid}" - f"&anchorUid={room_pid}" - f"&roomid={HUYA_BIND_ROOM_ID}" - ) - - def _wait_bind_role_result( - self, - client: HuyaHttpClient, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - uid: int, - cookie: str, - b_act_id_int: int, - result: dict, - ) -> tuple[str, dict]: - """生成二维码后轮询扫码状态与角色同步，直到识别到角色、超时或停止。""" - deadline = time.monotonic() + HUYA_BIND_ROLE_POLL_SECONDS - qrcode_token = str(result.get("qrcode_token") or "") - qrcode_finished = not qrcode_token - result["bind_polling"] = True - self._update_task_progress(worker_db, task, "running", "已生成绑定小程序码，等待扫码绑定", result) - - while not self._stop.is_set() and time.monotonic() < deadline: - if self._stop.wait(HUYA_BIND_ROLE_POLL_INTERVAL): - break - - if qrcode_token and not qrcode_finished: - qrcode_status = client.get_livelink_qrcode_status(qrcode_token, timeout=10.0) - if qrcode_status is not None: - result["qrcode_status"] = qrcode_status - if qrcode_status["is_expired"] or qrcode_status["is_failure"]: - result.update({ - "bind_phase": "qrcode_expired", - "bind_ready_for_confirm": False, - "bind_polling": False, - }) - self._update_task_progress( - worker_db, - task, - "running", - "绑定小程序码已失效，请重新获取", - result, - ) - return "", result - if qrcode_status["is_completed"]: - qrcode_finished = True - result["bind_phase"] = "qrcode_completed" - elif qrcode_status["is_scan"]: - result["bind_phase"] = "qrcode_scanned" - else: - result["bind_phase"] = "waiting_scan" - - bind_status, bind_query_result = self._resolve_bind_status( - client=client, - uid=uid, - cookie=cookie, - b_act_id_int=b_act_id_int, - ) - if bind_status is None: - continue - result.update(bind_query_result) - - if bind_status.status != 200: - result.update({ - "bind_phase": "role_check_failed", - "bind_status": bind_status.to_dict(), - }) - self._update_task_progress( - worker_db, - task, - "running", - bind_status.msg or "等待绑定角色同步", - result, - ) - continue - - previous_phase = result.get("bind_phase") - ready = self._bind_ready_result(bind_status) - # 角色未就绪时保留扫码阶段文案，避免状态来回跳。 - if not ready["role_name"] and previous_phase in { - "waiting_scan", - "qrcode_scanned", - "qrcode_completed", - }: - ready["bind_phase"] = previous_phase - result.update(ready) - result["bind_polling"] = True - role_name = ready["role_name"] - if role_name: - self._apply_role_to_account(account, bind_status, "game_queried") - result["bind_polling"] = False - self._update_task_progress( - worker_db, - task, - "running", - f"已识别角色: {role_name}，待确认绑定", - result, - ) - return role_name, result - - if result.get("bind_phase") == "qrcode_completed": - message = "小程序绑定已完成，等待角色同步" - elif result.get("bind_phase") == "qrcode_scanned": - message = "已扫码，等待小程序绑定完成" - else: - message = "已生成绑定小程序码，等待扫码绑定" - self._update_task_progress(worker_db, task, "running", message, result) - - result.update({ - "bind_phase": "role_timeout" if not self._stop.is_set() else "stopped", - "bind_ready_for_confirm": False, - "bind_polling": False, - }) - return "", result - - def _mark_task( - self, - worker_db: Session, - task: HuyaTask, - status: str, - message: str, - result: Optional[dict] = None, - ): - task.status = status - task.message = message - task.result = result - task.finished_at = datetime.now(timezone.utc) - worker_db.commit() - - def _update_task_progress( - self, - worker_db: Session, - task: HuyaTask, - status: str, - message: str, - result: Optional[dict] = None, - ): - task.status = status - task.message = message - if result is not None: - task.result = result - worker_db.commit() - - @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 "未知")) - - @classmethod - def _is_huya_order_paid(cls, 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: HuyaHttpClient, - 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 - - def _execute_query_points( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - 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") - return - - cookie = account_info.get("cookie") or "" - if not cookie: - self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") - return - - client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - response = client.query_user_score(uid=uid, cookie=cookie, sid=sid_int) - if response is None: - self._mark_task(worker_db, task, "error", "虎牙积分接口无响应") - return - - 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 - - points = response.available_score - account.points = points - account.status = "points_queried" - account.updated_at = datetime.now(timezone.utc) - self._mark_task(worker_db, task, "success", f"积分: {points}", result) - - def _execute_query_exchange_records( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - 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") - return - - cookie = account_info.get("cookie") or "" - if not cookie: - self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") - return - - client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - response = client.get_user_prize_records(uid=uid, cookie=cookie, sid=sid_int) - if response is None: - self._mark_task(worker_db, task, "error", "虎牙兑换记录接口无响应") - return - - 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 - - records = result.get("records", []) - for index, item in enumerate(records, start=1): - item["index"] = index - item["exchange_time_text"] = self._format_local_time(int(item.get("exchange_time") or 0)) - if item.get("score") is not None: - item["score_text"] = f"{int(item.get('score') or 0)}积分" - - account.status = "exchange_records_queried" - account.updated_at = datetime.now(timezone.utc) - count = len(records) - message = f"兑换记录 {count} 条" if count else "暂无兑换记录" - self._mark_task(worker_db, task, "success", message, result) - - def _execute_refresh_goods( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - 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") - return - - cookie = account_info.get("cookie") or "" - if not cookie: - self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") - return - - client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - 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 - - 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") - ] - now = datetime.now(timezone.utc) - worker_db.query(HuyaGoodsSnapshot).delete(synchronize_session=False) - for item in goods: - worker_db.add(HuyaGoodsSnapshot( - product_id=item["product_id"], - name=item["name"], - price=item["price"], - remain_text=item["remain_text"], - raw=item, - updated_at=now, - )) - - account.status = "goods_refreshed" - account.updated_at = now - message = f"已刷新商品 {len(goods)} 个" - self._mark_task(worker_db, task, "success", message, {**result, "goods": goods}) - - def _execute_exchange_goods( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - 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: - self._mark_task(worker_db, task, "failed", "请选择兑换商品") - 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 - - snapshot = worker_db.query(HuyaGoodsSnapshot).filter( - 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", "定时兑换时间格式无效") - return - if scheduled_at and scheduled_at.timestamp() > time.time(): - if not self._wait_until(scheduled_at, uid): - self._mark_task(worker_db, task, "stopped", "兑换任务已停止") - return - - client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - 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(timezone.utc).isoformat(), - "goods": snapshot.raw if snapshot else None, - }) - if response.status != 200: - self._mark_task( - worker_db, - task, - "failed", - response.msg or f"虎牙兑换失败: {response.status}", - result, - ) - return - - account.status = "goods_exchanged" - account.updated_at = datetime.now(timezone.utc) - message = response.msg or f"兑换成功: {product_name}" - self._mark_task(worker_db, task, "success", message, result) - - @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 = 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 = 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) - - def _execute_get_bind_qr( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - account_info: dict, - config_info: dict, - ): - b_act_id = str(self.payload.get("bind_act_id") or config_info.get("bind_act_id") or "").strip() - if not b_act_id: - self._mark_task(worker_db, task, "failed", "请先配置虎牙绑定 bActId") - return - - b_act_id_int = self._to_int(b_act_id) - if not b_act_id_int: - self._mark_task(worker_db, task, "failed", f"虎牙绑定 bActId 无效: {b_act_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 = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - bind_status, bind_query_result = self._resolve_bind_status( - client=client, - uid=uid, - cookie=cookie, - b_act_id_int=b_act_id_int, - ) - if bind_status is None: - self._mark_task(worker_db, task, "error", "虎牙绑定状态接口无响应") - return - if bind_status.status != 200: - self._mark_task( - worker_db, - task, - "failed", - bind_status.msg or f"虎牙绑定状态查询失败: {bind_status.status}", - bind_query_result or bind_status.to_dict(), - ) - return - - role_info = self._bind_role_result(bind_status) - change_state = self._bind_change_state(bind_status) - if role_info["role_name"]: - self._apply_role_to_account(account, bind_status, account.status or "imported") - if not change_state["can_change_bind"]: - role_name = role_info["role_name"] or "当前角色" - available_at = change_state["change_available_at"] - result = { - "bind_act_id": b_act_id_int, - **role_info, - **change_state, - "bind_status": bind_status.to_dict(), - } - self._mark_task( - worker_db, - task, - "failed", - f"{role_name} 暂不能更换，{available_at} 后可更换", - result, - ) - return - - live_link = client.get_live_link_param( - uid=uid, - cookie=cookie, - b_act_id=b_act_id_int, - game_auth_scene=bind_status.gameAuthScene, - ) - if live_link is None: - self._mark_task(worker_db, task, "error", "虎牙绑定二维码参数接口无响应") - return - if live_link.status != 200: - self._mark_task( - worker_db, - task, - "failed", - live_link.msg or f"虎牙绑定二维码参数获取失败: {live_link.status}", - live_link.to_log_dict(), - ) - return - - profile_nick = account_info.get("nickname") or account_info.get("username") or "" - profile_avatar = "" - profile_resp = client.get_user_profile_batch(uid=uid, cookie=cookie, target_uids=[uid]) - if profile_resp is not None and profile_resp.profiles: - profile = profile_resp.profiles[0] - profile_nick = profile.nick or profile.passport or profile_nick - profile_avatar = profile.avatar or "" - - bind_redirect_url = self._bind_redirect_url(config_info) - urls = client.build_bind_urls( - live_link.livelinkParam, - b_act_id_int, - game_auth_scene=bind_status.gameAuthScene, - nick_name=profile_nick, - face_url=profile_avatar, - redirect_url=bind_redirect_url, - ) - mini_qrcode = client.get_livelink_mini_qrcode(urls["qr_url"]) - if not mini_qrcode: - result = { - "bind_act_id": b_act_id_int, - "profile": { - "nick": profile_nick, - "avatar": profile_avatar, - }, - } - self._mark_task(worker_db, task, "failed", "绑定小程序码获取失败", result) - return - - result = { - "bind_act_id": b_act_id_int, - "mini_qrcode_image": mini_qrcode["mini_qrcode_image"], - "qrcode_token": mini_qrcode.get("qrcode_token") or "", - **bind_query_result, - "bind_phase": "waiting_scan" if mini_qrcode.get("qrcode_token") else "waiting_role", - "bind_ready_for_confirm": False, - "bind_polling": True, - "bind_redirect_url": bind_redirect_url, - **role_info, - **change_state, - "profile": { - "nick": profile_nick, - "avatar": profile_avatar, - }, - } - - account.status = "bind_qr_generated" - account.game_name = role_info["role_name"] or account.game_name - account.game_channel = self._role_channel(bind_status) or account.game_channel - account.nickname = profile_nick or account.nickname - account.updated_at = datetime.now(timezone.utc) - - # 生成二维码后自动轮询扫码/角色，避免用户必须手动点「查询角色」。 - role_name, result = self._wait_bind_role_result( - client=client, - worker_db=worker_db, - task=task, - account=account, - uid=uid, - cookie=cookie, - b_act_id_int=b_act_id_int, - result=result, - ) - if role_name: - self._mark_task(worker_db, task, "success", f"已识别角色: {role_name}，待确认绑定", result) - return - - if result.get("bind_phase") == "stopped": - self._mark_task(worker_db, task, "stopped", "任务已停止", result) - return - if result.get("bind_phase") == "qrcode_expired": - self._mark_task(worker_db, task, "failed", "绑定小程序码已失效，请重新获取", result) - return - self._mark_task(worker_db, task, "success", "已生成绑定小程序码，未检测到绑定角色", result) - - def _execute_query_game_name( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - account_info: dict, - config_info: dict, - ): - b_act_id = str(self.payload.get("bind_act_id") or config_info.get("bind_act_id") or "").strip() - if not b_act_id: - self._mark_task(worker_db, task, "failed", "请先配置虎牙绑定 bActId") - return - - b_act_id_int = self._to_int(b_act_id) - if not b_act_id_int: - self._mark_task(worker_db, task, "failed", f"虎牙绑定 bActId 无效: {b_act_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 = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - bind_status, bind_query_result = self._resolve_bind_status( - client=client, - uid=uid, - cookie=cookie, - b_act_id_int=b_act_id_int, - ) - if bind_status is None: - self._mark_task(worker_db, task, "error", "虎牙角色信息接口无响应") - return - if bind_status.status != 200: - self._mark_task( - worker_db, - task, - "failed", - bind_status.msg or f"虎牙角色信息查询失败: {bind_status.status}", - {"bind_act_id": b_act_id_int, **bind_query_result}, - ) - return - - role_info = self._bind_role_result(bind_status) - change_state = self._bind_change_state(bind_status) - result = { - "bind_act_id": b_act_id_int, - **role_info, - **change_state, - **bind_query_result, - "bind_ready_for_confirm": bool(role_info["role_name"]), - "bind_phase": "role_ready" if role_info["role_name"] else "waiting_role", - } - role_name = role_info["role_name"] - if role_name: - self._apply_role_to_account(account, bind_status, "game_queried") - self._mark_task(worker_db, task, "success", f"角色: {role_name}", result) - return - - account.status = "game_not_bound" - account.updated_at = datetime.now(timezone.utc) - self._mark_task(worker_db, task, "success", "未绑定游戏角色", result) - - def _execute_confirm_bind( - self, - worker_db: Session, - task: HuyaTask, - account: HuyaAccount, - account_info: dict, - config_info: dict, - ): - b_act_id = str(self.payload.get("bind_act_id") or config_info.get("bind_act_id") or "").strip() - if not b_act_id: - self._mark_task(worker_db, task, "failed", "请先配置虎牙绑定 bActId") - return - - b_act_id_int = self._to_int(b_act_id) - if not b_act_id_int: - self._mark_task(worker_db, task, "failed", f"虎牙绑定 bActId 无效: {b_act_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 = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) - role_status, role_query_result = self._resolve_bind_status( - client=client, - uid=uid, - cookie=cookie, - b_act_id_int=b_act_id_int, - ) - if role_status is None: - self._mark_task(worker_db, task, "error", "虎牙绑定角色查询接口无响应") - return - if role_status.status != 200: - self._mark_task( - worker_db, - task, - "failed", - role_status.msg or f"虎牙绑定角色查询失败: {role_status.status}", - {"bind_act_id": b_act_id_int, **role_query_result}, - ) - return - - role_info = self._bind_role_result(role_status) - role_name = role_info["role_name"] - if not role_name: - result = { - "bind_act_id": b_act_id_int, - "bind_confirmed": False, - **role_query_result, - } - self._mark_task(worker_db, task, "failed", "尚未识别到待确认角色，请先扫码完成绑定", result) - return - - confirm_resp = client.confirm_bind_act_account( - uid=uid, - cookie=cookie, - b_act_id=b_act_id_int, - ) - if confirm_resp is None: - self._mark_task(worker_db, task, "error", "虎牙确认绑定接口无响应") - return - if confirm_resp.status != 200: - result = { - "bind_act_id": b_act_id_int, - "bind_confirmed": False, - "confirm_result": confirm_resp.to_dict(), - "before_bind_status": role_status.to_dict(), - **role_query_result, - } - self._mark_task( - worker_db, - task, - "failed", - confirm_resp.msg or f"虎牙确认绑定失败: {confirm_resp.status}", - result, - ) - return - - refreshed_status, refreshed_query_result = self._resolve_bind_status( - client=client, - uid=uid, - cookie=cookie, - b_act_id_int=b_act_id_int, - ) - if refreshed_status is None: - result = { - "bind_act_id": b_act_id_int, - "bind_confirmed": True, - "confirm_result": confirm_resp.to_dict(), - "before_bind_status": role_status.to_dict(), - **role_info, - **role_query_result, - } - self._apply_role_to_account(account, role_status, "bind_confirmed") - self._mark_task(worker_db, task, "success", f"确认绑定: {role_name}", result) - return - if refreshed_status.status != 200: - result = { - "bind_act_id": b_act_id_int, - "bind_confirmed": True, - "confirm_result": confirm_resp.to_dict(), - "before_bind_status": role_status.to_dict(), - **role_info, - **role_query_result, - "refresh_error": refreshed_query_result, - } - self._apply_role_to_account(account, role_status, "bind_confirmed") - self._mark_task(worker_db, task, "success", f"确认绑定: {role_name}", result) - return - - refreshed_confirmed = bool( - refreshed_status.accountData.isBindAcount - and refreshed_status.accountData.isBindRole - ) - refreshed_role_info = self._bind_role_result(refreshed_status) - final_status = refreshed_status if refreshed_role_info["role_name"] else role_status - final_role_info = refreshed_role_info if refreshed_role_info["role_name"] else role_info - result = { - "bind_act_id": b_act_id_int, - "bind_confirmed": True, - "refreshed_is_bound": refreshed_confirmed, - "confirm_result": confirm_resp.to_dict(), - "before_bind_status": role_status.to_dict(), - **role_query_result, - "refresh_result": refreshed_query_result, - **final_role_info, - } - self._apply_role_to_account(account, final_status, "bind_confirmed") - role_name = final_role_info["role_name"] or role_name or "已绑定" - self._mark_task(worker_db, task, "success", f"确认绑定: {role_name}", result) +class HuyaBatchRunner( + HuyaBatchRunnerCore, BindMixin, GoodsMixin, RechargeMixin, +): + """批量执行虎牙任务（功能域 Mixin 聚合 + 批次调度）。""" def _execute_one(self, task_id: int, account_info: dict, config_info: dict, total: int): worker_db = SessionLocal() @@ -1598,67 +170,3 @@ class HuyaBatchRunner: self._push_log("result", "") finally: self.db.close() - - -class HuyaBatchRegistry: - """管理运行中的虎牙批次。""" - - def __init__(self): - self._batches: dict[str, dict] = {} - self._lock = threading.Lock() - - def _cleanup_locked(self, ttl_seconds: int = 300): - now = time.time() - expired = [ - batch_id - for batch_id, batch in self._batches.items() - if batch.get("finished") and now - float(batch.get("finished_at") or now) > ttl_seconds - ] - for batch_id in expired: - self._batches.pop(batch_id, None) - - def register( - self, - batch_id: str, - log_queue: asyncio.Queue, - loop: asyncio.AbstractEventLoop, - runner: HuyaBatchRunner, - ): - with self._lock: - self._cleanup_locked() - self._batches[batch_id] = { - "log_queue": log_queue, - "loop": loop, - "runner": runner, - "finished": False, - "finished_at": None, - } - - def get(self, batch_id: str): - with self._lock: - self._cleanup_locked() - return self._batches.get(batch_id) - - def active_ids(self) -> set[str]: - with self._lock: - self._cleanup_locked() - return { - batch_id - for batch_id, batch in self._batches.items() - if not batch.get("finished") - } - - def mark_finished(self, batch_id: str): - with self._lock: - batch = self._batches.get(batch_id) - if not batch: - return - batch["finished"] = True - batch["finished_at"] = time.time() - - def pop(self, batch_id: str): - with self._lock: - return self._batches.pop(batch_id, None) - - -huya_batch_registry = HuyaBatchRegistry() diff --git a/web/backend/services/huya_runner_bind.py b/web/backend/services/huya_runner_bind.py new file mode 100644 index 0000000..da110c6 --- /dev/null +++ b/web/backend/services/huya_runner_bind.py @@ -0,0 +1,628 @@ +"""虎牙任务执行器：角色绑定（由 huya_runner.py 按功能域拆分）。""" + +from __future__ import annotations +import time +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from core.huya import HuyaHttpClient +from ..models import HuyaAccount, HuyaTask + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .huya_runner import HuyaBatchRunner + +HUYA_BIND_ROLE_POLL_SECONDS = 180 +HUYA_BIND_ROLE_POLL_INTERVAL = 3 +HUYA_BIND_ZT_UUID = "b02faae1" +HUYA_BIND_ROOM_ID = "30596253" +from .huya_runner_core import HUYA_RECHARGE_SOURCE_ID + +class BindMixin: + """游戏角色绑定域：绑定状态机、扫码/确认绑定。""" + @staticmethod + def _role_name(bind_status) -> str: + account_data = bind_status.accountData + return account_data.gameRole.roleName or "" + + @staticmethod + def _has_bind_role(bind_status) -> bool: + return bool(bind_status and HuyaBatchRunner._role_name(bind_status)) + + @staticmethod + def _bind_role_result(bind_status) -> dict: + account_data = bind_status.accountData + game_account = account_data.gameAccount + game_role = account_data.gameRole + return { + "game_title": bind_status.gameName, + "role_name": HuyaBatchRunner._role_name(bind_status), + "change_bind_day": bind_status.changeBindDay, + "is_bind_account": account_data.isBindAcount, + "is_bind_role": account_data.isBindRole, + "is_need_act_check": account_data.isNeedActCheck, + "change_bind_time": account_data.changBindTime, + "game_account": game_account.to_dict(), + "game_role": game_role.to_dict(), + } + + @staticmethod + def _role_channel(bind_status) -> str: + game_role = bind_status.accountData.gameRole + parts = [game_role.platName, game_role.areaName] + return " / ".join(part for part in parts if part) + + @classmethod + def _bind_change_state(cls, bind_status) -> dict: + account_data = bind_status.accountData + is_bound = bool(account_data.isBindAcount and account_data.isBindRole) + change_time = int(account_data.changBindTime or 0) + now = int(datetime.now(timezone.utc).timestamp()) + can_change = not is_bound or not change_time or change_time <= now + return { + "is_bound": is_bound, + "can_change_bind": can_change, + "change_bind_time": change_time, + "change_available_at": cls._format_local_time(change_time), + "change_bind_day": int(bind_status.changeBindDay or 0), + } + + @classmethod + def _bind_ready_result(cls, bind_status) -> dict: + role_info = cls._bind_role_result(bind_status) + return { + **role_info, + **cls._bind_change_state(bind_status), + "bind_status": bind_status.to_dict(), + "bind_ready_for_confirm": bool(role_info["role_name"]), + "bind_phase": "role_ready" if role_info["role_name"] else "waiting_role", + } + + def _resolve_bind_status( + self, + client: HuyaHttpClient, + uid: int, + cookie: str, + b_act_id_int: int, + ): + """按活动页逻辑解析绑定状态，优先返回含角色的状态。""" + outer_status = client.check_user_bind_game_account( + uid=uid, + cookie=cookie, + b_act_id=b_act_id_int, + is_use_outer_act_id=1, + ) + if outer_status is None: + return None, {} + + query_result = { + "bind_status_source": "outer", + "outer_bind_status": outer_status.to_dict(), + } + if outer_status.status != 200: + return outer_status, query_result + + chosen_status = outer_status + account_data = outer_status.accountData + should_check_inner = ( + not self._has_bind_role(outer_status) + and bool(account_data.isNeedActCheck or not account_data.isBindAcount or not account_data.isBindRole) + ) + if should_check_inner: + inner_status = client.check_user_bind_game_account( + uid=uid, + cookie=cookie, + b_act_id=b_act_id_int, + is_use_outer_act_id=0, + ) + if inner_status is not None: + query_result["inner_bind_status"] = inner_status.to_dict() + if inner_status.status == 200 and self._has_bind_role(inner_status): + chosen_status = inner_status + query_result["bind_status_source"] = "inner" + + query_result["bind_status"] = chosen_status.to_dict() + return chosen_status, query_result + + def _apply_role_to_account(self, account: HuyaAccount, bind_status, status: str): + role_name = self._role_name(bind_status) + account.status = status + account.game_name = role_name or account.game_name + account.game_channel = self._role_channel(bind_status) or account.game_channel + account.updated_at = datetime.now(timezone.utc) + + @staticmethod + def _bind_redirect_url(config_info: dict) -> str: + room_pid = str(config_info.get("room_pid") or "").strip() + if not room_pid: + return "" + return ( + f"https://zt.huya.com/{HUYA_BIND_ZT_UUID}/pc/index.html" + f"?sourceId={HUYA_RECHARGE_SOURCE_ID}" + f"&pid={room_pid}" + f"&anchorUid={room_pid}" + f"&roomid={HUYA_BIND_ROOM_ID}" + ) + + def _wait_bind_role_result( + self, + client: HuyaHttpClient, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + uid: int, + cookie: str, + b_act_id_int: int, + result: dict, + ) -> tuple[str, dict]: + """生成二维码后轮询扫码状态与角色同步，直到识别到角色、超时或停止。""" + deadline = time.monotonic() + HUYA_BIND_ROLE_POLL_SECONDS + qrcode_token = str(result.get("qrcode_token") or "") + qrcode_finished = not qrcode_token + result["bind_polling"] = True + self._update_task_progress(worker_db, task, "running", "已生成绑定小程序码，等待扫码绑定", result) + + while not self._stop.is_set() and time.monotonic() < deadline: + if self._stop.wait(HUYA_BIND_ROLE_POLL_INTERVAL): + break + + if qrcode_token and not qrcode_finished: + qrcode_status = client.get_livelink_qrcode_status(qrcode_token, timeout=10.0) + if qrcode_status is not None: + result["qrcode_status"] = qrcode_status + if qrcode_status["is_expired"] or qrcode_status["is_failure"]: + result.update({ + "bind_phase": "qrcode_expired", + "bind_ready_for_confirm": False, + "bind_polling": False, + }) + self._update_task_progress( + worker_db, + task, + "running", + "绑定小程序码已失效，请重新获取", + result, + ) + return "", result + if qrcode_status["is_completed"]: + qrcode_finished = True + result["bind_phase"] = "qrcode_completed" + elif qrcode_status["is_scan"]: + result["bind_phase"] = "qrcode_scanned" + else: + result["bind_phase"] = "waiting_scan" + + bind_status, bind_query_result = self._resolve_bind_status( + client=client, + uid=uid, + cookie=cookie, + b_act_id_int=b_act_id_int, + ) + if bind_status is None: + continue + result.update(bind_query_result) + + if bind_status.status != 200: + result.update({ + "bind_phase": "role_check_failed", + "bind_status": bind_status.to_dict(), + }) + self._update_task_progress( + worker_db, + task, + "running", + bind_status.msg or "等待绑定角色同步", + result, + ) + continue + + previous_phase = result.get("bind_phase") + ready = self._bind_ready_result(bind_status) + # 角色未就绪时保留扫码阶段文案，避免状态来回跳。 + if not ready["role_name"] and previous_phase in { + "waiting_scan", + "qrcode_scanned", + "qrcode_completed", + }: + ready["bind_phase"] = previous_phase + result.update(ready) + result["bind_polling"] = True + role_name = ready["role_name"] + if role_name: + self._apply_role_to_account(account, bind_status, "game_queried") + result["bind_polling"] = False + self._update_task_progress( + worker_db, + task, + "running", + f"已识别角色: {role_name}，待确认绑定", + result, + ) + return role_name, result + + if result.get("bind_phase") == "qrcode_completed": + message = "小程序绑定已完成，等待角色同步" + elif result.get("bind_phase") == "qrcode_scanned": + message = "已扫码，等待小程序绑定完成" + else: + message = "已生成绑定小程序码，等待扫码绑定" + self._update_task_progress(worker_db, task, "running", message, result) + + result.update({ + "bind_phase": "role_timeout" if not self._stop.is_set() else "stopped", + "bind_ready_for_confirm": False, + "bind_polling": False, + }) + return "", result + + def _execute_get_bind_qr( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + account_info: dict, + config_info: dict, + ): + b_act_id = str(self.payload.get("bind_act_id") or config_info.get("bind_act_id") or "").strip() + if not b_act_id: + self._mark_task(worker_db, task, "failed", "请先配置虎牙绑定 bActId") + return + + b_act_id_int = self._to_int(b_act_id) + if not b_act_id_int: + self._mark_task(worker_db, task, "failed", f"虎牙绑定 bActId 无效: {b_act_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 = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + bind_status, bind_query_result = self._resolve_bind_status( + client=client, + uid=uid, + cookie=cookie, + b_act_id_int=b_act_id_int, + ) + if bind_status is None: + self._mark_task(worker_db, task, "error", "虎牙绑定状态接口无响应") + return + if bind_status.status != 200: + self._mark_task( + worker_db, + task, + "failed", + bind_status.msg or f"虎牙绑定状态查询失败: {bind_status.status}", + bind_query_result or bind_status.to_dict(), + ) + return + + role_info = self._bind_role_result(bind_status) + change_state = self._bind_change_state(bind_status) + if role_info["role_name"]: + self._apply_role_to_account(account, bind_status, account.status or "imported") + if not change_state["can_change_bind"]: + role_name = role_info["role_name"] or "当前角色" + available_at = change_state["change_available_at"] + result = { + "bind_act_id": b_act_id_int, + **role_info, + **change_state, + "bind_status": bind_status.to_dict(), + } + self._mark_task( + worker_db, + task, + "failed", + f"{role_name} 暂不能更换，{available_at} 后可更换", + result, + ) + return + + live_link = client.get_live_link_param( + uid=uid, + cookie=cookie, + b_act_id=b_act_id_int, + game_auth_scene=bind_status.gameAuthScene, + ) + if live_link is None: + self._mark_task(worker_db, task, "error", "虎牙绑定二维码参数接口无响应") + return + if live_link.status != 200: + self._mark_task( + worker_db, + task, + "failed", + live_link.msg or f"虎牙绑定二维码参数获取失败: {live_link.status}", + live_link.to_log_dict(), + ) + return + + profile_nick = account_info.get("nickname") or account_info.get("username") or "" + profile_avatar = "" + profile_resp = client.get_user_profile_batch(uid=uid, cookie=cookie, target_uids=[uid]) + if profile_resp is not None and profile_resp.profiles: + profile = profile_resp.profiles[0] + profile_nick = profile.nick or profile.passport or profile_nick + profile_avatar = profile.avatar or "" + + bind_redirect_url = self._bind_redirect_url(config_info) + urls = client.build_bind_urls( + live_link.livelinkParam, + b_act_id_int, + game_auth_scene=bind_status.gameAuthScene, + nick_name=profile_nick, + face_url=profile_avatar, + redirect_url=bind_redirect_url, + ) + mini_qrcode = client.get_livelink_mini_qrcode(urls["qr_url"]) + if not mini_qrcode: + result = { + "bind_act_id": b_act_id_int, + "profile": { + "nick": profile_nick, + "avatar": profile_avatar, + }, + } + self._mark_task(worker_db, task, "failed", "绑定小程序码获取失败", result) + return + + result = { + "bind_act_id": b_act_id_int, + "mini_qrcode_image": mini_qrcode["mini_qrcode_image"], + "qrcode_token": mini_qrcode.get("qrcode_token") or "", + **bind_query_result, + "bind_phase": "waiting_scan" if mini_qrcode.get("qrcode_token") else "waiting_role", + "bind_ready_for_confirm": False, + "bind_polling": True, + "bind_redirect_url": bind_redirect_url, + **role_info, + **change_state, + "profile": { + "nick": profile_nick, + "avatar": profile_avatar, + }, + } + + account.status = "bind_qr_generated" + account.game_name = role_info["role_name"] or account.game_name + account.game_channel = self._role_channel(bind_status) or account.game_channel + account.nickname = profile_nick or account.nickname + account.updated_at = datetime.now(timezone.utc) + + # 生成二维码后自动轮询扫码/角色，避免用户必须手动点「查询角色」。 + role_name, result = self._wait_bind_role_result( + client=client, + worker_db=worker_db, + task=task, + account=account, + uid=uid, + cookie=cookie, + b_act_id_int=b_act_id_int, + result=result, + ) + if role_name: + self._mark_task(worker_db, task, "success", f"已识别角色: {role_name}，待确认绑定", result) + return + + if result.get("bind_phase") == "stopped": + self._mark_task(worker_db, task, "stopped", "任务已停止", result) + return + if result.get("bind_phase") == "qrcode_expired": + self._mark_task(worker_db, task, "failed", "绑定小程序码已失效，请重新获取", result) + return + self._mark_task(worker_db, task, "success", "已生成绑定小程序码，未检测到绑定角色", result) + + def _execute_query_game_name( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + account_info: dict, + config_info: dict, + ): + b_act_id = str(self.payload.get("bind_act_id") or config_info.get("bind_act_id") or "").strip() + if not b_act_id: + self._mark_task(worker_db, task, "failed", "请先配置虎牙绑定 bActId") + return + + b_act_id_int = self._to_int(b_act_id) + if not b_act_id_int: + self._mark_task(worker_db, task, "failed", f"虎牙绑定 bActId 无效: {b_act_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 = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + bind_status, bind_query_result = self._resolve_bind_status( + client=client, + uid=uid, + cookie=cookie, + b_act_id_int=b_act_id_int, + ) + if bind_status is None: + self._mark_task(worker_db, task, "error", "虎牙角色信息接口无响应") + return + if bind_status.status != 200: + self._mark_task( + worker_db, + task, + "failed", + bind_status.msg or f"虎牙角色信息查询失败: {bind_status.status}", + {"bind_act_id": b_act_id_int, **bind_query_result}, + ) + return + + role_info = self._bind_role_result(bind_status) + change_state = self._bind_change_state(bind_status) + result = { + "bind_act_id": b_act_id_int, + **role_info, + **change_state, + **bind_query_result, + "bind_ready_for_confirm": bool(role_info["role_name"]), + "bind_phase": "role_ready" if role_info["role_name"] else "waiting_role", + } + role_name = role_info["role_name"] + if role_name: + self._apply_role_to_account(account, bind_status, "game_queried") + self._mark_task(worker_db, task, "success", f"角色: {role_name}", result) + return + + account.status = "game_not_bound" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(worker_db, task, "success", "未绑定游戏角色", result) + + def _execute_confirm_bind( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + account_info: dict, + config_info: dict, + ): + b_act_id = str(self.payload.get("bind_act_id") or config_info.get("bind_act_id") or "").strip() + if not b_act_id: + self._mark_task(worker_db, task, "failed", "请先配置虎牙绑定 bActId") + return + + b_act_id_int = self._to_int(b_act_id) + if not b_act_id_int: + self._mark_task(worker_db, task, "failed", f"虎牙绑定 bActId 无效: {b_act_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 = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + role_status, role_query_result = self._resolve_bind_status( + client=client, + uid=uid, + cookie=cookie, + b_act_id_int=b_act_id_int, + ) + if role_status is None: + self._mark_task(worker_db, task, "error", "虎牙绑定角色查询接口无响应") + return + if role_status.status != 200: + self._mark_task( + worker_db, + task, + "failed", + role_status.msg or f"虎牙绑定角色查询失败: {role_status.status}", + {"bind_act_id": b_act_id_int, **role_query_result}, + ) + return + + role_info = self._bind_role_result(role_status) + role_name = role_info["role_name"] + if not role_name: + result = { + "bind_act_id": b_act_id_int, + "bind_confirmed": False, + **role_query_result, + } + self._mark_task(worker_db, task, "failed", "尚未识别到待确认角色，请先扫码完成绑定", result) + return + + confirm_resp = client.confirm_bind_act_account( + uid=uid, + cookie=cookie, + b_act_id=b_act_id_int, + ) + if confirm_resp is None: + self._mark_task(worker_db, task, "error", "虎牙确认绑定接口无响应") + return + if confirm_resp.status != 200: + result = { + "bind_act_id": b_act_id_int, + "bind_confirmed": False, + "confirm_result": confirm_resp.to_dict(), + "before_bind_status": role_status.to_dict(), + **role_query_result, + } + self._mark_task( + worker_db, + task, + "failed", + confirm_resp.msg or f"虎牙确认绑定失败: {confirm_resp.status}", + result, + ) + return + + refreshed_status, refreshed_query_result = self._resolve_bind_status( + client=client, + uid=uid, + cookie=cookie, + b_act_id_int=b_act_id_int, + ) + if refreshed_status is None: + result = { + "bind_act_id": b_act_id_int, + "bind_confirmed": True, + "confirm_result": confirm_resp.to_dict(), + "before_bind_status": role_status.to_dict(), + **role_info, + **role_query_result, + } + self._apply_role_to_account(account, role_status, "bind_confirmed") + self._mark_task(worker_db, task, "success", f"确认绑定: {role_name}", result) + return + if refreshed_status.status != 200: + result = { + "bind_act_id": b_act_id_int, + "bind_confirmed": True, + "confirm_result": confirm_resp.to_dict(), + "before_bind_status": role_status.to_dict(), + **role_info, + **role_query_result, + "refresh_error": refreshed_query_result, + } + self._apply_role_to_account(account, role_status, "bind_confirmed") + self._mark_task(worker_db, task, "success", f"确认绑定: {role_name}", result) + return + + refreshed_confirmed = bool( + refreshed_status.accountData.isBindAcount + and refreshed_status.accountData.isBindRole + ) + refreshed_role_info = self._bind_role_result(refreshed_status) + final_status = refreshed_status if refreshed_role_info["role_name"] else role_status + final_role_info = refreshed_role_info if refreshed_role_info["role_name"] else role_info + result = { + "bind_act_id": b_act_id_int, + "bind_confirmed": True, + "refreshed_is_bound": refreshed_confirmed, + "confirm_result": confirm_resp.to_dict(), + "before_bind_status": role_status.to_dict(), + **role_query_result, + "refresh_result": refreshed_query_result, + **final_role_info, + } + self._apply_role_to_account(account, final_status, "bind_confirmed") + role_name = final_role_info["role_name"] or role_name or "已绑定" + self._mark_task(worker_db, task, "success", f"确认绑定: {role_name}", result) + diff --git a/web/backend/services/huya_runner_core.py b/web/backend/services/huya_runner_core.py new file mode 100644 index 0000000..f720c01 --- /dev/null +++ b/web/backend/services/huya_runner_core.py @@ -0,0 +1,209 @@ +"""虎牙任务执行器：公共基础（由 huya_runner.py 按功能域拆分）。""" + +from __future__ import annotations + +import asyncio +import threading +import time +from datetime import datetime, timezone +from typing import Optional + +from loguru import logger +from sqlalchemy.orm import Session +from ..models import HuyaTask +from .huya_service import cookie_value + +# 绑定跳转与充值下单共用（bind 与 recharge 两个域都引用了 sourceId） +HUYA_RECHARGE_SOURCE_ID = "yellowcarlist" +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .huya_runner import HuyaBatchRunner + +class HuyaBatchRunnerCore: + """虎牙任务执行器公共基础：批次状态、日志、任务落库。""" + def __init__( + self, + db: Session, + batch_id: str, + task_type: str, + payload: Optional[dict] = None, + log_queue: Optional[asyncio.Queue] = None, + loop: Optional[asyncio.AbstractEventLoop] = None, + concurrency: int = 3, + ): + self.db = db + self.batch_id = batch_id + self.task_type = task_type + self.payload = payload or {} + self.log_queue = log_queue + self.loop = loop + self.concurrency = max(1, min(concurrency, 10)) + self._stop = threading.Event() + self._counter_lock = threading.Lock() + self._started = 0 + + def stop(self): + self._stop.set() + + def _push_log(self, level: str, message: str): + if level == "result": + try: + huya_batch_registry.mark_finished(self.batch_id) + except NameError: + pass + if level != "result" and message: + log_func = getattr(logger, level, logger.info) + log_func(f"[huya] {message}") + if self.log_queue and self.loop: + asyncio.run_coroutine_threadsafe( + self.log_queue.put({"level": level, "message": message}), + self.loop, + ) + + @staticmethod + def _account_name(account_info: dict) -> str: + return ( + account_info.get("nickname") + or account_info.get("username") + or account_info.get("uid") + or f"#{account_info.get('account_id')}" + ) + + @staticmethod + def _to_int(value) -> int: + text = str(value or "").strip() + return int(text) if text.isdigit() else 0 + + def _resolve_uid(self, account_info: dict) -> int: + cookie = account_info.get("cookie") or "" + return ( + self._to_int(account_info.get("yyuid")) + or self._to_int(account_info.get("uid")) + or self._to_int(cookie_value(cookie, "yyuid")) + or self._to_int(cookie_value(cookie, "udb_uid")) + ) + + @staticmethod + def _format_local_time(timestamp: int) -> str: + if not timestamp: + return "" + return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S") + + @staticmethod + def _parse_scheduled_time(value) -> datetime | None: + text = str(value or "").strip() + if not text: + return None + try: + normalized = text.replace("Z", "+00:00") + dt = datetime.fromisoformat(normalized) + except ValueError: + return None + if dt.tzinfo is None: + return dt.astimezone() + return dt + + def _wait_until(self, when: datetime, uid: int) -> bool: + target = when.timestamp() + local_text = self._format_local_time(int(target)) + self._push_log("info", f"[{uid}] 定时兑换等待到 {local_text}") + while not self._stop.is_set(): + remaining = target - time.time() + if remaining <= 0: + return True + time.sleep(min(0.2, max(0.02, remaining))) + return False + + def _mark_task( + self, + worker_db: Session, + task: HuyaTask, + status: str, + message: str, + result: Optional[dict] = None, + ): + task.status = status + task.message = message + task.result = result + task.finished_at = datetime.now(timezone.utc) + worker_db.commit() + + def _update_task_progress( + self, + worker_db: Session, + task: HuyaTask, + status: str, + message: str, + result: Optional[dict] = None, + ): + task.status = status + task.message = message + if result is not None: + task.result = result + worker_db.commit() + + +class HuyaBatchRegistry: + """管理运行中的虎牙批次。""" + + def __init__(self): + self._batches: dict[str, dict] = {} + self._lock = threading.Lock() + + def _cleanup_locked(self, ttl_seconds: int = 300): + now = time.time() + expired = [ + batch_id + for batch_id, batch in self._batches.items() + if batch.get("finished") and now - float(batch.get("finished_at") or now) > ttl_seconds + ] + for batch_id in expired: + self._batches.pop(batch_id, None) + + def register( + self, + batch_id: str, + log_queue: asyncio.Queue, + loop: asyncio.AbstractEventLoop, + runner: HuyaBatchRunner, + ): + with self._lock: + self._cleanup_locked() + self._batches[batch_id] = { + "log_queue": log_queue, + "loop": loop, + "runner": runner, + "finished": False, + "finished_at": None, + } + + def get(self, batch_id: str): + with self._lock: + self._cleanup_locked() + return self._batches.get(batch_id) + + def active_ids(self) -> set[str]: + with self._lock: + self._cleanup_locked() + return { + batch_id + for batch_id, batch in self._batches.items() + if not batch.get("finished") + } + + def mark_finished(self, batch_id: str): + with self._lock: + batch = self._batches.get(batch_id) + if not batch: + return + batch["finished"] = True + batch["finished_at"] = time.time() + + def pop(self, batch_id: str): + with self._lock: + return self._batches.pop(batch_id, None) + + +huya_batch_registry = HuyaBatchRegistry() + diff --git a/web/backend/services/huya_runner_goods.py b/web/backend/services/huya_runner_goods.py new file mode 100644 index 0000000..90e560c --- /dev/null +++ b/web/backend/services/huya_runner_goods.py @@ -0,0 +1,267 @@ +"""虎牙任务执行器：积分与商城（由 huya_runner.py 按功能域拆分）。""" + +from __future__ import annotations +import time +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from core.huya import HuyaHttpClient +from ..models import HuyaAccount, HuyaGoodsSnapshot, HuyaTask + +class GoodsMixin: + """积分与商城域：积分/兑换记录/商品刷新/兑换。""" + def _execute_query_points( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + 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") + return + + cookie = account_info.get("cookie") or "" + if not cookie: + self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") + return + + client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + response = client.query_user_score(uid=uid, cookie=cookie, sid=sid_int) + if response is None: + self._mark_task(worker_db, task, "error", "虎牙积分接口无响应") + return + + 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 + + points = response.available_score + account.points = points + account.status = "points_queried" + account.updated_at = datetime.now(timezone.utc) + self._mark_task(worker_db, task, "success", f"积分: {points}", result) + + def _execute_query_exchange_records( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + 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") + return + + cookie = account_info.get("cookie") or "" + if not cookie: + self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") + return + + client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + response = client.get_user_prize_records(uid=uid, cookie=cookie, sid=sid_int) + if response is None: + self._mark_task(worker_db, task, "error", "虎牙兑换记录接口无响应") + return + + 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 + + records = result.get("records", []) + for index, item in enumerate(records, start=1): + item["index"] = index + item["exchange_time_text"] = self._format_local_time(int(item.get("exchange_time") or 0)) + if item.get("score") is not None: + item["score_text"] = f"{int(item.get('score') or 0)}积分" + + account.status = "exchange_records_queried" + account.updated_at = datetime.now(timezone.utc) + count = len(records) + message = f"兑换记录 {count} 条" if count else "暂无兑换记录" + self._mark_task(worker_db, task, "success", message, result) + + def _execute_refresh_goods( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + 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") + return + + cookie = account_info.get("cookie") or "" + if not cookie: + self._mark_task(worker_db, task, "failed", "账号 Cookie 为空") + return + + client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + 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 + + 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") + ] + now = datetime.now(timezone.utc) + worker_db.query(HuyaGoodsSnapshot).delete(synchronize_session=False) + for item in goods: + worker_db.add(HuyaGoodsSnapshot( + product_id=item["product_id"], + name=item["name"], + price=item["price"], + remain_text=item["remain_text"], + raw=item, + updated_at=now, + )) + + account.status = "goods_refreshed" + account.updated_at = now + message = f"已刷新商品 {len(goods)} 个" + self._mark_task(worker_db, task, "success", message, {**result, "goods": goods}) + + def _execute_exchange_goods( + self, + worker_db: Session, + task: HuyaTask, + account: HuyaAccount, + 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: + self._mark_task(worker_db, task, "failed", "请选择兑换商品") + 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 + + snapshot = worker_db.query(HuyaGoodsSnapshot).filter( + 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", "定时兑换时间格式无效") + return + if scheduled_at and scheduled_at.timestamp() > time.time(): + if not self._wait_until(scheduled_at, uid): + self._mark_task(worker_db, task, "stopped", "兑换任务已停止") + return + + client = HuyaHttpClient(logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")) + 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(timezone.utc).isoformat(), + "goods": snapshot.raw if snapshot else None, + }) + if response.status != 200: + self._mark_task( + worker_db, + task, + "failed", + response.msg or f"虎牙兑换失败: {response.status}", + result, + ) + return + + account.status = "goods_exchanged" + account.updated_at = datetime.now(timezone.utc) + message = response.msg or f"兑换成功: {product_name}" + self._mark_task(worker_db, task, "success", message, result) + diff --git a/web/backend/services/huya_runner_recharge.py b/web/backend/services/huya_runner_recharge.py new file mode 100644 index 0000000..9c3c12a --- /dev/null +++ b/web/backend/services/huya_runner_recharge.py @@ -0,0 +1,460 @@ +"""虎牙任务执行器：充值（由 huya_runner.py 按功能域拆分）。""" + +from __future__ import annotations +import time +from datetime import datetime, timezone + +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 + +class RechargeMixin: + """充值域：充值商品、下单与到账轮询。""" + @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 "未知")) + + @classmethod + def _is_huya_order_paid(cls, 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: HuyaHttpClient, + 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 = 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 = 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) +