yml2213
bb20533c90
- 扩展号池支持 peace (和平精英/小跑道) 活动 scope，映射 xpd_balance 与 exchange_xpd_goods 任务 - 新增 POST /douyu/pool/refresh-goods 接口与前端一键刷新，支持复用库内有效 CK 自动更新商品快照 - 前端 DouyuPoolPage 升级视觉设计与栅格布局，商品选择器支持按活动类型分类联动 - 修复 DouyuTask.result 类型守卫、account_events actor_id 类型防御与 runner 过滤规则边界值处理 - 验证方式：全量 pytest (797 通过)、pyright (0 错误)、tsc -b 及 ruff format/check 全部通过
493 lines
18 KiB
Python
493 lines
18 KiB
Python
"""斗鱼任务执行器：公共基础（由 douyu_runner.py 按功能域拆分）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import operator
|
||
import threading
|
||
import time
|
||
from datetime import UTC, datetime
|
||
from typing import TYPE_CHECKING, Any, ClassVar
|
||
|
||
from loguru import logger
|
||
from sqlalchemy.orm import Session
|
||
|
||
from core.douyu import DouyuActivityClient
|
||
from core.douyu.proxy_fetcher import ProxyFetcher
|
||
|
||
from ..models import (
|
||
Account,
|
||
DouyuEsportsGoodsSnapshot,
|
||
DouyuGoodsSnapshot,
|
||
DouyuTask,
|
||
DouyuXpdGoodsSnapshot,
|
||
)
|
||
from ..models import (
|
||
ProxyConfig as ProxyConfigModel,
|
||
)
|
||
from .account_pool import release_for_task
|
||
from .douyu_service import (
|
||
DOUYU_CONFIG_FIELDS,
|
||
douyu_config_value,
|
||
douyu_task_payload,
|
||
ensure_douyu_config,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
from .douyu_runner import DouyuBatchRunner
|
||
|
||
# 支付/到账轮询（手册与充值共用）
|
||
DOUYU_PAYMENT_POLL_SECONDS = 600
|
||
|
||
#: 支付等待超时后的状态：**转人工**，不再自动轮询，租约也先不还。
|
||
MANUAL_REQUIRED_STATUS = "manual_required"
|
||
DOUYU_PAYMENT_POLL_INTERVAL = 5
|
||
|
||
# 走代理的任务类型：写操作/消耗类（兑换、锁单、支付、开通、充值、送礼）。
|
||
# 查询/刷新/绑定/扫码等读操作始终直连——它们不怕 IP 频控，走代理只会浪费代理额度。
|
||
DOUYU_PROXY_TASK_TYPES = {
|
||
# 兑换链路（IP 频控重点, 对应"精英手册火爆"）
|
||
"lock_goods",
|
||
"pay_locked_order",
|
||
"exchange_goods",
|
||
"exchange_esports_goods",
|
||
"exchange_xpd_goods",
|
||
# 消耗类写操作: 开通手册 / 充值 / 送礼
|
||
"create_elite_qr",
|
||
"create_esports_qr",
|
||
"create_gold_qr",
|
||
"donate_elite_gift",
|
||
"donate_esports_chicken_gift",
|
||
"donate_esports_firework_gift",
|
||
}
|
||
|
||
|
||
class DouyuBatchRunnerCore:
|
||
"""斗鱼任务执行器公共基础：批次状态、日志、任务落库与客户端构造。"""
|
||
|
||
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,
|
||
close_db: bool = True,
|
||
):
|
||
self.db = db
|
||
self.close_db = close_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
|
||
# 代理接入 (配置源 = 代理配置页, 与 CK 检测/虎牙注册同款):
|
||
# 仅兑换/锁单/支付/充值/送礼等写操作任务走代理, 查询/刷新/绑定类直连
|
||
self._proxy_cfg = db.query(ProxyConfigModel).first() if db else None
|
||
self._proxy_fetcher = self._create_proxy_fetcher()
|
||
self._static_proxies = self._resolve_static_proxies()
|
||
if self._static_proxies:
|
||
logger.info(
|
||
f"[douyu] 写操作任务将走静态代理: {self._static_proxies.get('https', '')}"
|
||
)
|
||
elif self._proxy_fetcher:
|
||
logger.info("[douyu] 写操作任务将按任务从代理 API 取新代理")
|
||
|
||
def _create_proxy_fetcher(self) -> ProxyFetcher | None:
|
||
"""API 代理模式: 每个任务从代理 API 取 1 个新代理 (与虎牙注册链同款)."""
|
||
cfg = self._proxy_cfg
|
||
if not cfg or not cfg.enabled or not cfg.api_url:
|
||
return None
|
||
wl_platform = getattr(cfg, "whitelist_platform", None) or "xiequ"
|
||
wl_credentials = getattr(cfg, "whitelist_credentials", None)
|
||
if (
|
||
not wl_credentials
|
||
and getattr(cfg, "whitelist_uid", "")
|
||
and getattr(cfg, "whitelist_ukey", "")
|
||
):
|
||
wl_credentials = {"uid": cfg.whitelist_uid, "ukey": cfg.whitelist_ukey}
|
||
return ProxyFetcher(
|
||
api_url=cfg.api_url,
|
||
whitelist_platform=wl_platform,
|
||
whitelist_credentials=wl_credentials if cfg.whitelist_enabled else None,
|
||
stop_event=self._stop,
|
||
)
|
||
|
||
def _resolve_static_proxies(self) -> dict[str, str] | None:
|
||
"""静态代理模式: 代理配置页手填的 http/https 地址."""
|
||
cfg = self._proxy_cfg
|
||
if not cfg or not cfg.enabled:
|
||
return None
|
||
http, https = (cfg.http or "").strip(), (cfg.https or "").strip()
|
||
if not http and not https:
|
||
return None
|
||
return {"http": http or https, "https": https or http}
|
||
|
||
def _proxies_for_task(self) -> dict[str, str] | None:
|
||
"""取本任务出网代理。
|
||
|
||
写/读分离：仅兑换/锁单/支付/开通/充值/送礼等写操作任务走代理（应对 IP 频控）；
|
||
查询/刷新/绑定/扫码等读操作任务始终直连，不消耗代理额度。
|
||
"""
|
||
if self.task_type not in DOUYU_PROXY_TASK_TYPES:
|
||
return None
|
||
if self._static_proxies:
|
||
return self._static_proxies
|
||
if self._proxy_fetcher:
|
||
try:
|
||
proxy_url = self._proxy_fetcher.fetch_new_proxy()
|
||
if proxy_url:
|
||
return {"http": proxy_url, "https": proxy_url}
|
||
self._push_log("warning", "代理 API 未返回可用代理, 本任务降级直连")
|
||
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
self._push_log("warning", f"取代理失败, 本任务降级直连: {exc}")
|
||
return None
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
|
||
def _push_log(self, level: str, message: str):
|
||
if level == "result":
|
||
try:
|
||
douyu_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"[douyu] {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: Account) -> str:
|
||
return account.nickname or account.username or account.uid or f"#{account.id}"
|
||
|
||
@staticmethod
|
||
def _to_int(value) -> int | None:
|
||
if value is None:
|
||
return None
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _format_wait_time(seconds: int | None) -> str:
|
||
if seconds is None:
|
||
return ""
|
||
seconds = max(0, int(seconds))
|
||
days, rem = divmod(seconds, 86400)
|
||
hours, rem = divmod(rem, 3600)
|
||
minutes, sec = divmod(rem, 60)
|
||
if days:
|
||
return f"{days}天{hours}小时{minutes}分"
|
||
if hours:
|
||
return f"{hours}小时{minutes}分{sec}秒"
|
||
return f"{minutes}分{sec}秒"
|
||
|
||
def _push_task_event(self, task: DouyuTask) -> None:
|
||
"""向批次 WS 推送任务状态事件（level=task），前端即时更新不依赖轮询。"""
|
||
if not self.log_queue or not self.loop:
|
||
return
|
||
try:
|
||
payload = douyu_task_payload(task)
|
||
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
logger.exception("[douyu] 推送任务状态失败: task_id={}", task.id)
|
||
return
|
||
event = {
|
||
"level": "task",
|
||
"message": "",
|
||
"task": payload,
|
||
}
|
||
asyncio.run_coroutine_threadsafe(self.log_queue.put(event), self.loop)
|
||
|
||
def _mark_task(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
status: str,
|
||
message: str,
|
||
result: dict | None = None,
|
||
) -> None:
|
||
task.status = status
|
||
task.message = message[:512]
|
||
if result is not None:
|
||
task.result = result
|
||
task.finished_at = datetime.now(UTC)
|
||
db.commit()
|
||
# 账号池：任务走到终态就**把租约还回去**（设计 §3.9 的即时释放）。
|
||
# 这一步覆盖 99% 的情况；漏掉的（进程被杀、任务行被清）由
|
||
# `account_lease_reaper` 兜底。释放失败绝不能影响任务结果本身。
|
||
self._release_pool_lease(db, task)
|
||
self._push_task_event(task)
|
||
|
||
def _release_pool_lease(self, db: Session, task: DouyuTask) -> None:
|
||
"""释放该任务对应账号的租约（不是整批 —— 同批每个账号各有一条）。"""
|
||
try:
|
||
if not task.batch_id:
|
||
return
|
||
# 转人工的任务**先不还号**：客服还在等支付/处理，账号已经挂在这单上。
|
||
# 同理，同一批次里只要还有 `manual_required` 的任务（补给链后半段会被
|
||
# 前置闸门标成 skipped），整条链的号都先留着。
|
||
# 账号不会被永久占住 —— 租约到期由 `account_lease_reaper` 兜底收回。
|
||
if task.status == MANUAL_REQUIRED_STATUS:
|
||
return
|
||
waiting_manual = (
|
||
db.query(DouyuTask.id)
|
||
.filter(
|
||
DouyuTask.batch_id == task.batch_id,
|
||
DouyuTask.status == MANUAL_REQUIRED_STATUS,
|
||
)
|
||
.first()
|
||
)
|
||
if waiting_manual:
|
||
return
|
||
release_for_task(db, task, reason=f"task_{task.status}")
|
||
except Exception as exc: # noqa: BLE001 旁路动作，绝不能带崩任务
|
||
logger.warning(f"释放账号租约失败 task={task.id}: {exc}")
|
||
|
||
# ── 补给链的两个运行期闸门（见 account_pool.build_supply_chain）──────────
|
||
|
||
#: `_skip_when` 支持的账号字段与比较符。
|
||
_SKIP_COLUMNS: ClassVar[tuple[str, ...]] = (
|
||
"points",
|
||
"gold_balance",
|
||
"exchange_balance",
|
||
"esports_points",
|
||
)
|
||
_SKIP_OPS: ClassVar[dict[str, Any]] = {
|
||
">=": operator.ge,
|
||
">": operator.gt,
|
||
"<=": operator.le,
|
||
"<": operator.lt,
|
||
"==": operator.eq,
|
||
}
|
||
|
||
def _skip_reason(self, account: Account, task: DouyuTask) -> str:
|
||
"""`_skip_when` 闸门：满足条件就跳过这一步。
|
||
|
||
补给链用它实现「余额已经够了就别再充钱」。**只在能确定时跳过**：
|
||
字段缺失、比较符不认识、值解析失败一律不跳（宁可多跑一步，
|
||
也不要因为解析不出来就静默把钱省掉/花掉）。
|
||
"""
|
||
rule = self._task_payload(task).get("_skip_when")
|
||
if not isinstance(rule, dict):
|
||
return ""
|
||
column = str(rule.get("column") or "")
|
||
if column not in self._SKIP_COLUMNS:
|
||
return ""
|
||
compare = self._SKIP_OPS.get(str(rule.get("op") or ""))
|
||
if compare is None:
|
||
return ""
|
||
current = getattr(account, column, None)
|
||
if current is None:
|
||
return ""
|
||
raw_val = rule.get("value")
|
||
if raw_val is None:
|
||
return ""
|
||
try:
|
||
target = int(raw_val)
|
||
except (TypeError, ValueError):
|
||
return ""
|
||
if compare(int(current), target):
|
||
return f"跳过：{column}={int(current)} {rule.get('op')} {target} 已满足"
|
||
return ""
|
||
|
||
def _dependency_reason(self, db: Session, task: DouyuTask) -> str:
|
||
"""`_depends_on` 闸门：前置任务没成功就不必再跑（省得连着一串失败）。"""
|
||
raw = self._task_payload(task).get("_depends_on")
|
||
if not raw:
|
||
return ""
|
||
try:
|
||
depends_on = int(raw)
|
||
except (TypeError, ValueError):
|
||
return ""
|
||
previous = db.query(DouyuTask).filter(DouyuTask.id == depends_on).first()
|
||
if previous is None:
|
||
return ""
|
||
# 前置「成功」或「被跳过」都算通过：跳过意味着那一步本来就不需要做。
|
||
if previous.status in ("success", "skipped"):
|
||
return ""
|
||
return f"前置任务 #{depends_on} 未成功（{previous.status}），跳过"
|
||
|
||
def _update_task_progress(
|
||
self,
|
||
db: Session,
|
||
task: DouyuTask,
|
||
status: str,
|
||
message: str,
|
||
result: dict | None = None,
|
||
) -> None:
|
||
task.status = status
|
||
task.message = message[:512]
|
||
if result is not None:
|
||
task.result = result
|
||
db.commit()
|
||
self._push_task_event(task)
|
||
|
||
def _upsert_goods(self, db: Session, goods: list[dict]) -> None:
|
||
now = datetime.now(UTC)
|
||
for raw in goods:
|
||
commodity_id = str(raw.get("commodityId") or raw.get("commodity_id") or "")
|
||
if not commodity_id:
|
||
continue
|
||
row = (
|
||
db.query(DouyuGoodsSnapshot)
|
||
.filter(DouyuGoodsSnapshot.commodity_id == commodity_id)
|
||
.first()
|
||
)
|
||
score = self._to_int(raw.get("score"))
|
||
if row is None:
|
||
row = DouyuGoodsSnapshot(commodity_id=commodity_id)
|
||
db.add(row)
|
||
row.name = str(raw.get("commodityName") or raw.get("name") or "")
|
||
row.score = score
|
||
row.status = str(raw.get("status") or "")
|
||
row.raw = raw
|
||
row.updated_at = now
|
||
db.commit()
|
||
|
||
def _upsert_esports_goods(self, db: Session, goods: list[dict]) -> None:
|
||
now = datetime.now(UTC)
|
||
for raw in goods:
|
||
commodity_id = str(raw.get("commodityId") or raw.get("commodity_id") or "")
|
||
if not commodity_id:
|
||
continue
|
||
row = (
|
||
db.query(DouyuEsportsGoodsSnapshot)
|
||
.filter(DouyuEsportsGoodsSnapshot.commodity_id == commodity_id)
|
||
.first()
|
||
)
|
||
if row is None:
|
||
row = DouyuEsportsGoodsSnapshot(commodity_id=commodity_id)
|
||
db.add(row)
|
||
row.name = str(raw.get("commodityName") or raw.get("name") or "")
|
||
row.score = self._to_int(raw.get("score"))
|
||
row.status = str(raw.get("status") or "")
|
||
row.raw = raw
|
||
row.updated_at = now
|
||
db.commit()
|
||
|
||
def _upsert_xpd_goods(self, db: Session, goods: list[dict]) -> None:
|
||
"""同步和平小店商品快照，移除上一次热门抢购等遗留商品。"""
|
||
now = datetime.now(UTC)
|
||
commodity_ids = {
|
||
str(raw.get("commodity_id") or raw.get("iGoodsId") or "") for raw in goods
|
||
}
|
||
commodity_ids.discard("")
|
||
query = db.query(DouyuXpdGoodsSnapshot)
|
||
if commodity_ids:
|
||
query.filter(~DouyuXpdGoodsSnapshot.commodity_id.in_(commodity_ids)).delete(
|
||
synchronize_session=False,
|
||
)
|
||
else:
|
||
query.delete(synchronize_session=False)
|
||
for raw in goods:
|
||
commodity_id = str(raw.get("commodity_id") or raw.get("iGoodsId") or "")
|
||
if not commodity_id:
|
||
continue
|
||
row = (
|
||
db.query(DouyuXpdGoodsSnapshot)
|
||
.filter(DouyuXpdGoodsSnapshot.commodity_id == commodity_id)
|
||
.first()
|
||
)
|
||
if row is None:
|
||
row = DouyuXpdGoodsSnapshot(commodity_id=commodity_id)
|
||
db.add(row)
|
||
row.name = str(raw.get("name") or raw.get("sGoodsName") or "")
|
||
row.price = self._to_int(raw.get("price") or raw.get("iPrice"))
|
||
row.org_price = self._to_int(raw.get("org_price") or raw.get("iOrgPrice"))
|
||
row.category = str(raw.get("category") or raw.get("iCategoryId") or "")
|
||
goods_left = raw.get("goods_left")
|
||
if goods_left is None:
|
||
goods_left = raw.get("iGoodsLeft")
|
||
row.goods_left = self._to_int(goods_left)
|
||
row.raw = raw
|
||
row.updated_at = now
|
||
db.commit()
|
||
|
||
def _config_info(self, db: Session) -> dict:
|
||
config = ensure_douyu_config(db)
|
||
return {
|
||
field: douyu_config_value(field, getattr(config, field, None))
|
||
for field in DOUYU_CONFIG_FIELDS
|
||
}
|
||
|
||
def _task_payload(self, task: DouyuTask) -> dict:
|
||
result = task.result if isinstance(task.result, dict) else {}
|
||
payload_raw = result.get("payload")
|
||
payload = payload_raw if isinstance(payload_raw, dict) else {}
|
||
return {**payload, **self.payload}
|
||
|
||
def _client(self, cookie: str) -> DouyuActivityClient:
|
||
"""任务 HTTP 客户端：写操作任务注入会话级代理（兑换/锁单/支付/充值/送礼），
|
||
读操作任务直连；见 DOUYU_PROXY_TASK_TYPES。"""
|
||
return DouyuActivityClient(
|
||
cookie,
|
||
logger=lambda msg: self._push_log("debug", msg),
|
||
proxies=self._proxies_for_task(),
|
||
)
|
||
|
||
def _sleep_interruptible(self, seconds: float) -> bool:
|
||
"""分段睡眠，任务停止时提前返回；返回 False 表示已被停止。"""
|
||
waited = 0.0
|
||
step = 0.5
|
||
while waited < seconds:
|
||
if self._stop.is_set():
|
||
return False
|
||
time.sleep(min(step, seconds - waited))
|
||
waited += step
|
||
return not self._stop.is_set()
|
||
|
||
|
||
class DouyuBatchRegistry:
|
||
"""管理运行中的斗鱼任务批次。"""
|
||
|
||
def __init__(self):
|
||
self._batches: dict[str, dict] = {}
|
||
|
||
def register(
|
||
self,
|
||
batch_id: str,
|
||
log_queue: asyncio.Queue,
|
||
loop: asyncio.AbstractEventLoop,
|
||
runner: DouyuBatchRunner,
|
||
):
|
||
self._batches[batch_id] = {
|
||
"log_queue": log_queue,
|
||
"loop": loop,
|
||
"runner": runner,
|
||
"finished": False,
|
||
"updated_at": time.time(),
|
||
}
|
||
|
||
def get(self, batch_id: str):
|
||
return self._batches.get(batch_id)
|
||
|
||
def pop(self, batch_id: str):
|
||
return self._batches.pop(batch_id, None)
|
||
|
||
def mark_finished(self, batch_id: str):
|
||
if batch_id in self._batches:
|
||
self._batches[batch_id]["finished"] = True
|
||
self._batches[batch_id]["updated_at"] = time.time()
|
||
|
||
def active_ids(self) -> set[str]:
|
||
return {
|
||
batch_id
|
||
for batch_id, info in self._batches.items()
|
||
if not info.get("finished")
|
||
}
|
||
|
||
|
||
douyu_batch_registry = DouyuBatchRegistry()
|