393 lines
14 KiB
Python
393 lines
14 KiB
Python
"""斗鱼任务执行器：公共基础（由 douyu_runner.py 按功能域拆分）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import threading
|
||
import time
|
||
from datetime import UTC, datetime
|
||
from typing import TYPE_CHECKING
|
||
|
||
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 .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
|
||
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,
|
||
):
|
||
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
|
||
# 代理接入 (配置源 = 代理配置页, 与 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()
|
||
self._push_task_event(task)
|
||
|
||
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()
|