yml2213
73536d62c8
- 手机号列：外部 CK 的 `----手机号` 本来就落在 huya_accounts.game_phone，
列表一直没展示；现在单独成列，并纳入搜索
- 精英宝典列：显示「已开通/未开通/未查询」+ 开通时间 + 有效期
· 开通状态 = 积分中心 getRedeemNeedGoods 的 tag5（真机对照网页：1=已开通，
0=未开通，此时 tag7 会列出开通套餐），拿不准时为 NULL（显示「未查询」）
· 开通时间 = revenueWebUI.queryUserOrderList 里已支付的宝典订单 pay_time
（按 SPU 白名单 + 订单名含「精英宝典」识别，未支付订单不计，取最近一次）
· 有效期 = PlayMallUI.queryAllocation 页面配置的本赛季节点 tag10/tag11
（2026-10-15 抓包：精英宝典福利季丨Q4，2026-10-09 ~ 2026-12-31 23:59:59），
不写死索引，递归找带 tag10/tag11 的节点
- 新任务 query_all「一键查询」：勾选账号一次跑完角色绑定 + 宝典开通 + 积分，
按部分失败如实汇总（某个接口挂了不影响其它结论），并把宝典状态写回账号
- 迁移 20261015_0053：huya_accounts 新增 elite_opened / elite_open_time /
elite_expire_time（时间存 Unix 秒，与斗鱼 elite_expire_time 同约定）
- 开通商品按页面「赠送指定礼物得积分」那一排对齐成 6 款（去掉页面已不用的
旧版精英宝典 hy-5879340），列表接口只返回账号级快照并按 spu 去重
- 建 WSS 会话失败（积分中心端点静默不回包）时回退 HTTP cdnws，
不再让整批任务判失败
- 列表列宽改按比例分配，避免宽屏下被拉出奇怪的宽度；最近操作不再重复状态
- 新增 tests/test_huya_elite_query.py（10 条）；同步迁移头版本断言
真机验证：#65 已开通（10-13 20:41:45，有效期 12-31）、#70 新开通
（10-15 16:37:37）判定与订单/网页一致
381 lines
13 KiB
Python
381 lines
13 KiB
Python
"""虎牙任务执行器：公共基础（由 huya_runner.py 按功能域拆分）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import copy
|
||
import threading
|
||
import time
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
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: dict | None = None,
|
||
log_queue: asyncio.Queue | None = None,
|
||
loop: asyncio.AbstractEventLoop | None = 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
|
||
self._wss_sessions = threading.local()
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
|
||
def _open_wss_session(self, uid: int, cookie: str, kind: str):
|
||
"""在当前任务线程创建并登记 WSS 会话，统一由 finally 回收。"""
|
||
from core.huya.elite_session import HuyaEliteWssSession
|
||
|
||
session = HuyaEliteWssSession(
|
||
uid=uid,
|
||
cookie=cookie,
|
||
kind=kind,
|
||
logger=lambda message: self._push_log("info", f"[{uid}] {message}"),
|
||
)
|
||
self._session_bucket().append(session)
|
||
self._persist_session_cookie(uid, cookie, session.cookie)
|
||
return session
|
||
|
||
def _persist_session_cookie(
|
||
self, uid: int, original_cookie: str, filled_cookie: str
|
||
):
|
||
"""活动会话补齐网页字段后回写账号，避免下次任务重新生成设备态。"""
|
||
account_id = getattr(self._wss_sessions, "account_id", None)
|
||
if not account_id:
|
||
return
|
||
from ..database import SessionLocal
|
||
from .huya_service import persist_huya_session_cookie
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
if persist_huya_session_cookie(
|
||
db, int(account_id), original_cookie, filled_cookie
|
||
):
|
||
self._push_log("info", f"[{uid}] 已回写活动会话补齐后的网页 Cookie")
|
||
except Exception: # noqa: BLE001 回写失败不能打断任务
|
||
logger.debug("[huya] 回写会话 Cookie 失败")
|
||
finally:
|
||
db.close()
|
||
|
||
def _session_bucket(self) -> list[Any]:
|
||
bucket = getattr(self._wss_sessions, "items", None)
|
||
if bucket is None:
|
||
bucket = []
|
||
self._wss_sessions.items = bucket
|
||
return bucket
|
||
|
||
def _activity_client(self, uid: int, cookie: str):
|
||
"""活动通道客户端：网页设备态齐全走 WSS，否则回退 HTTP cdnws。
|
||
|
||
2026-10-15 实测（外部 CK 导入后）：只带登录态、缺 sdid/udb_deviceid/
|
||
udb_guiddata/udb_anobiztoken 的 Cookie 在 WSS 上会被服务端静默丢弃业务请求
|
||
（wsLaunch 或业务 RPC 无响应）；同一条 Cookie 走 HTTP cdnws 通道可以正常
|
||
查询角色与商品。因此这里不再直接失败，而是按能力选通道。
|
||
"""
|
||
missing = self._web_wss_cookie_missing(cookie)
|
||
if missing:
|
||
self._push_log(
|
||
"info",
|
||
f"[{uid}] Cookie 缺少网页设备态({', '.join(missing)})，改用 HTTP cdnws 通道",
|
||
)
|
||
return self._http_client()
|
||
session = self._open_wss_or_http(uid, cookie, "activity")
|
||
if session is None:
|
||
return self._http_client()
|
||
self._push_log("info", f"[{uid}] 活动链路使用 WSS 会话")
|
||
return session
|
||
|
||
def _open_wss_or_http(self, uid: int, cookie: str, kind: str):
|
||
"""建 WSS 会话；服务端不理会握手（wsLaunch 超时）时返回 None 让调用方回退 HTTP。
|
||
|
||
2026-10-15 实测：设备态字段齐全的账号也可能被积分中心端点静默丢弃
|
||
（wsLaunch 无响应），而同一账号走 HTTP cdnws 一切正常 —— 会话建不起来时
|
||
直接回退，而不是把整批任务判失败。
|
||
"""
|
||
try:
|
||
return self._open_wss_session(uid, cookie, kind)
|
||
except Exception as exc: # noqa: BLE001 - 任何建连/初始化异常都回退 HTTP
|
||
self._push_log(
|
||
"warning",
|
||
f"[{uid}] {kind} WSS 会话不可用({type(exc).__name__}: {exc})，改用 HTTP cdnws 通道",
|
||
)
|
||
return None
|
||
|
||
def _http_client(self):
|
||
"""HTTP cdnws 通道客户端（无需网页设备态，方法签名与 WSS 会话保持一致）。"""
|
||
from core.huya.http_client import HuyaHttpClient
|
||
|
||
return HuyaHttpClient(logger=lambda message: self._push_log("info", message))
|
||
|
||
@staticmethod
|
||
def _web_wss_cookie_missing(cookie: str) -> tuple[str, ...]:
|
||
"""返回 WSS 前置字段中缺失的名称，不输出任何 Cookie 值。
|
||
|
||
必填集 = 认证字段 + sdid + 匿名设备态协议可签发的字段
|
||
（udb_deviceid/udb_guiddata/udb_anobiztoken，见 core/huya/anon_device.py）。
|
||
guid/_qimei_uuid42/__yamid_new/game_did 目前无服务端签发接口，
|
||
不阻断 WSS（风控风险自担）。
|
||
"""
|
||
required = (
|
||
"udb_cred",
|
||
"udb_biztoken",
|
||
"udb_uid",
|
||
"yyuid",
|
||
"sdid",
|
||
"udb_deviceid",
|
||
"udb_guiddata",
|
||
"udb_anobiztoken",
|
||
)
|
||
return tuple(key for key in required if not cookie_value(cookie, key))
|
||
|
||
@classmethod
|
||
def _has_web_wss_cookie(cls, cookie: str) -> bool:
|
||
"""9.1 活动 WSS 需要网页设备态，不接受只有 App 认证字段的 Cookie。"""
|
||
return not cls._web_wss_cookie_missing(cookie)
|
||
|
||
def _shop_client(self, uid: int, cookie: str):
|
||
"""商城通道客户端：网页设备态齐全走 WSS，否则回退 HTTP cdnws（原因同活动通道）。"""
|
||
missing = self._web_wss_cookie_missing(cookie)
|
||
if missing:
|
||
self._push_log(
|
||
"info",
|
||
f"[{uid}] Cookie 缺少网页设备态({', '.join(missing)})，改用 HTTP cdnws 通道",
|
||
)
|
||
return self._http_client()
|
||
session = self._open_wss_or_http(uid, cookie, "shop")
|
||
if session is None:
|
||
return self._http_client()
|
||
self._push_log("info", f"[{uid}] 商城链路使用 WSS 会话")
|
||
return session
|
||
|
||
def _close_wss_sessions(self):
|
||
bucket = self._session_bucket()
|
||
for session in reversed(bucket):
|
||
try:
|
||
session.close()
|
||
except Exception: # noqa: BLE001 - cleanup must not mask task status
|
||
logger.debug("[huya] WSS 会话清理异常")
|
||
bucket.clear()
|
||
|
||
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, UTC).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: dict | None = None,
|
||
):
|
||
task.status = status
|
||
task.message = message
|
||
task.result = result
|
||
task.finished_at = datetime.now(UTC)
|
||
worker_db.commit()
|
||
self._push_task_event(task)
|
||
|
||
def _update_task_progress(
|
||
self,
|
||
worker_db: Session,
|
||
task: HuyaTask,
|
||
status: str,
|
||
message: str,
|
||
result: dict | None = None,
|
||
):
|
||
task.status = status
|
||
task.message = message
|
||
if result is not None:
|
||
task.result = result
|
||
worker_db.commit()
|
||
self._push_task_event(task)
|
||
|
||
def _push_task_event(self, task: HuyaTask) -> None:
|
||
"""向批次 WS 推送任务状态；二维码图片留给详情接口按需读取。"""
|
||
if not self.log_queue or not self.loop:
|
||
return
|
||
result = copy.deepcopy(task.result) if isinstance(task.result, dict) else None
|
||
if result and isinstance(result.get("mini_qrcode_image"), str):
|
||
result.pop("mini_qrcode_image", None)
|
||
result["has_mini_qrcode"] = True
|
||
event = {
|
||
"level": "task",
|
||
"message": "",
|
||
"task": {
|
||
"id": task.id,
|
||
"batch_id": task.batch_id,
|
||
"account_id": task.account_id,
|
||
"task_type": task.task_type,
|
||
"handbook_scope": getattr(task, "handbook_scope", "legacy") or "legacy",
|
||
"status": task.status or "",
|
||
"message": task.message or "",
|
||
"result": result,
|
||
"created_by": task.created_by,
|
||
"created_at": task.created_at.isoformat() if task.created_at else None,
|
||
"finished_at": task.finished_at.isoformat()
|
||
if task.finished_at
|
||
else None,
|
||
},
|
||
}
|
||
asyncio.run_coroutine_threadsafe(self.log_queue.put(event), self.loop)
|
||
|
||
|
||
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()
|