diff --git a/core/douyu/activity_client.py b/core/douyu/activity_client.py index dfd37b0..a6c243e 100644 --- a/core/douyu/activity_client.py +++ b/core/douyu/activity_client.py @@ -81,11 +81,15 @@ class DouyuActivityClient: *, logger: Callable[[str], None] | None = None, timeout: tuple[float, float] = (8, 20), + proxies: dict[str, str] | None = None, ): self.cookie = normalize_douyu_cookie(cookie) self.timeout = timeout self.logger = logger or (lambda _msg: None) self.session = requests.Session() + # trust_env=False: 不读环境变量代理, 只用显式传入的 proxies (代理配置页控制) + if proxies: + self.session.proxies.update(proxies) self.session.trust_env = False self.session.headers.update({ "User-Agent": PC_UA, diff --git a/web/backend/services/douyu_runner_core.py b/web/backend/services/douyu_runner_core.py index a14f099..be06d90 100644 --- a/web/backend/services/douyu_runner_core.py +++ b/web/backend/services/douyu_runner_core.py @@ -12,12 +12,24 @@ from loguru import logger from sqlalchemy.orm import Session from core.douyu import DouyuActivityClient -from ..models import Account, DouyuEsportsGoodsSnapshot, DouyuGoodsSnapshot, DouyuTask, DouyuXpdGoodsSnapshot +from core.douyu.proxy_fetcher import ProxyFetcher +from ..models import ( + Account, + DouyuEsportsGoodsSnapshot, + DouyuGoodsSnapshot, + DouyuTask, + DouyuXpdGoodsSnapshot, + ProxyConfig as ProxyConfigModel, +) from .douyu_service import DOUYU_CONFIG_FIELDS, douyu_config_value, ensure_douyu_config, douyu_task_payload # 支付/到账轮询(手册与充值共用) DOUYU_PAYMENT_POLL_SECONDS = 600 DOUYU_PAYMENT_POLL_INTERVAL = 5 +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .douyu_runner import DouyuBatchRunner class DouyuBatchRunnerCore: """斗鱼任务执行器公共基础:批次状态、日志、任务落库与客户端构造。""" @@ -41,6 +53,55 @@ class DouyuBatchRunnerCore: self._stop = threading.Event() self._counter_lock = threading.Lock() self._started = 0 + # 代理接入 (配置源 = 代理配置页, 与 CK 检测/虎牙注册同款): + # 开启后斗鱼任务(精英手册兑换/和平小店/电竞等)出网统一走代理, 应对 IP 频控 + 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: + """取本任务出网代理: 静态优先; API 模式每任务取新代理, 失败降级直连并告警.""" + 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: + self._push_log("warning", f"取代理失败, 本任务降级直连: {exc}") + return None def stop(self): self._stop.set() @@ -226,7 +287,12 @@ class DouyuBatchRunnerCore: return {**payload, **self.payload} def _client(self, cookie: str) -> DouyuActivityClient: - return DouyuActivityClient(cookie, logger=lambda msg: self._push_log("debug", msg)) + """任务 HTTP 客户端: 会话级代理 (覆盖精英手册/和平小店/电竞等全部请求).""" + return DouyuActivityClient( + cookie, + logger=lambda msg: self._push_log("debug", msg), + proxies=self._proxies_for_task(), + ) def _sleep_interruptible(self, seconds: float) -> bool: """分段睡眠,任务停止时提前返回;返回 False 表示已被停止。"""