From 3b3a4bebcb50087260f2a79fe6c2b7bcb7a69242 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 30 Aug 2026 11:28:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(douyu):=20=E4=BB=A3=E7=90=86=E5=86=99?= =?UTF-8?q?=E8=AF=BB=E5=88=86=E7=A6=BB=20=E2=80=94=20=E4=BB=85=E5=85=91?= =?UTF-8?q?=E6=8D=A2/=E9=94=81=E5=8D=95/=E6=94=AF=E4=BB=98/=E5=BC=80?= =?UTF-8?q?=E9=80=9A/=E5=85=85=E5=80=BC/=E9=80=81=E7=A4=BC=E8=B5=B0?= =?UTF-8?q?=E4=BB=A3=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前代理配置页开启后所有斗鱼任务(含查积分/刷列表等只读任务)统一走代理: - API 模式每任务 fetch 新代理, 查询任务白白消耗代理额度 - 查询任务失败面被代理抖动放大, 但查询本不怕 IP 频控 新策略 (DOUYU_PROXY_TASK_TYPES 白名单): - 写操作/消耗类: lock/pay/exchange_goods, exchange_esports/xpd_goods, create_elite/esports/gold_qr, donate_* 走代理 (静态模式或 API 每任务新代理) - 读操作类: query_*/refresh_*/get_*_bind_qr 等始终直连, 不消耗额度 - 判型用批次 task_type (同批次同类型, 与任务一致); 总开关仍为代理配置页 测试: 新增读任务直连/白名单覆盖断言, 三模式行为断言保留, 101 单测通过 --- tests/test_douyu_runner_proxy.py | 47 +++++++++++++++++----- web/backend/services/douyu_runner_core.py | 48 +++++++++++++++++------ 2 files changed, 74 insertions(+), 21 deletions(-) diff --git a/tests/test_douyu_runner_proxy.py b/tests/test_douyu_runner_proxy.py index 22c64c5..8910b13 100644 --- a/tests/test_douyu_runner_proxy.py +++ b/tests/test_douyu_runner_proxy.py @@ -1,4 +1,4 @@ -"""斗鱼任务代理接入三模式行为测试（静态 / API / 关闭）。""" +"""斗鱼任务代理接入三模式行为测试（静态 / API / 关闭）+ 写读分离。""" import os import unittest @@ -9,7 +9,7 @@ os.environ.setdefault("APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODl from core.douyu.activity_client import DouyuActivityClient from web.backend.services.douyu_runner import DouyuBatchRunner -from web.backend.services.douyu_runner_core import DouyuBatchRunnerCore +from web.backend.services.douyu_runner_core import DouyuBatchRunnerCore, DOUYU_PROXY_TASK_TYPES from web.backend.models import ProxyConfig @@ -32,17 +32,17 @@ def _cfg(**overrides) -> ProxyConfig: return cfg -def _runner(db_cfg: ProxyConfig): +def _runner(db_cfg: ProxyConfig, task_type: str = "query_points"): db = Mock() db.query.return_value.first.return_value = db_cfg runner = DouyuBatchRunner.__new__(DouyuBatchRunner) - DouyuBatchRunnerCore.__init__(runner, db, "batch", "query_points") + DouyuBatchRunnerCore.__init__(runner, db, "batch", task_type) return runner class DouyuProxyModeTests(unittest.TestCase): def test_disabled_mode_goes_direct(self): - runner = _runner(_cfg(enabled=False)) + runner = _runner(_cfg(enabled=False), task_type="exchange_goods") self.assertIsNone(runner._proxies_for_task()) self.assertIsNone(runner._proxy_fetcher) client = runner._client("acf_uid=100") @@ -50,8 +50,8 @@ class DouyuProxyModeTests(unittest.TestCase): # 未开启代理时 execute 客户端不带 proxies 参数 self.assertIsInstance(client, DouyuActivityClient) - def test_static_mode_uses_configured_proxies(self): - runner = _runner(_cfg(enabled=True, http="http://10.0.0.1:8080")) + def test_static_mode_uses_configured_proxies_for_write_tasks(self): + runner = _runner(_cfg(enabled=True, http="http://10.0.0.1:8080"), task_type="exchange_goods") proxies = runner._proxies_for_task() self.assertEqual(proxies, {"http": "http://10.0.0.1:8080", "https": "http://10.0.0.1:8080"}) client = runner._client("acf_uid=100") @@ -59,9 +59,9 @@ class DouyuProxyModeTests(unittest.TestCase): # 每次任务都取同一静态代理 self.assertEqual(runner._proxies_for_task(), proxies) - def test_api_mode_fetches_new_proxy_per_task(self): + def test_api_mode_fetches_new_proxy_per_write_task(self): cfg = _cfg(enabled=True, api_url="https://proxy-api.example/fetch") - runner = _runner(cfg) + runner = _runner(cfg, task_type="exchange_goods") self.assertIsNotNone(runner._proxy_fetcher) fake = Mock() fake.fetch_new_proxy.return_value = "http://10.0.0.2:3128" @@ -78,7 +78,7 @@ class DouyuProxyModeTests(unittest.TestCase): self.assertEqual(fake.fetch_new_proxy.call_count, 3) def test_api_mode_failure_falls_back_to_direct(self): - runner = _runner(_cfg(enabled=True, api_url="https://proxy-api.example/fetch")) + runner = _runner(_cfg(enabled=True, api_url="https://proxy-api.example/fetch"), task_type="exchange_goods") runner._proxy_fetcher = Mock() runner._proxy_fetcher.fetch_new_proxy.return_value = None runner._push_log = Mock() @@ -88,6 +88,33 @@ class DouyuProxyModeTests(unittest.TestCase): self.assertEqual(client.session.proxies, {}) runner._push_log.assert_called() + def test_read_tasks_stay_direct_even_with_proxy_enabled(self): + """写读分离：查询/刷新/绑定等读任务即使代理开启也直连，不消耗代理额度。""" + runner = _runner(_cfg(enabled=True, http="http://10.0.0.1:8080"), task_type="query_points") + self.assertIsNotNone(runner._static_proxies) + self.assertIsNone(runner._proxies_for_task()) + client = runner._client("acf_uid=100") + self.assertEqual(client.session.proxies, {}) + + # API 模式同理：读任务不触发 fetch + runner = _runner(_cfg(enabled=True, api_url="https://proxy-api.example/fetch"), task_type="query_exchange_records") + runner._proxy_fetcher = Mock() + runner._proxy_fetcher.fetch_new_proxy.side_effect = AssertionError("读任务不应取代理") + self.assertIsNone(runner._proxies_for_task()) + + def test_proxy_task_types_covers_all_write_tasks(self): + """白名单集合应覆盖全部会消耗/下单的写操作任务，且不含查询类。""" + self.assertIn("exchange_goods", DOUYU_PROXY_TASK_TYPES) + self.assertIn("lock_goods", DOUYU_PROXY_TASK_TYPES) + self.assertIn("pay_locked_order", DOUYU_PROXY_TASK_TYPES) + self.assertIn("exchange_esports_goods", DOUYU_PROXY_TASK_TYPES) + self.assertIn("exchange_xpd_goods", DOUYU_PROXY_TASK_TYPES) + self.assertIn("create_gold_qr", DOUYU_PROXY_TASK_TYPES) + self.assertIn("donate_elite_gift", DOUYU_PROXY_TASK_TYPES) + self.assertNotIn("query_points", DOUYU_PROXY_TASK_TYPES) + self.assertNotIn("refresh_goods", DOUYU_PROXY_TASK_TYPES) + self.assertNotIn("get_bind_qr", DOUYU_PROXY_TASK_TYPES) + def test_activity_client_accepts_proxies_at_session_level(self): client = DouyuActivityClient( "acf_uid=100", diff --git a/web/backend/services/douyu_runner_core.py b/web/backend/services/douyu_runner_core.py index be06d90..f9562c1 100644 --- a/web/backend/services/douyu_runner_core.py +++ b/web/backend/services/douyu_runner_core.py @@ -6,7 +6,7 @@ import asyncio import threading import time from datetime import datetime, timezone -from typing import Optional +from typing import Optional, TYPE_CHECKING from loguru import logger from sqlalchemy.orm import Session @@ -23,16 +23,35 @@ from ..models import ( ) 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 +# 支付/到账轮询（手册与充值共用） +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, @@ -54,14 +73,14 @@ class DouyuBatchRunnerCore: 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', '')}") + logger.info(f"[douyu] 写操作任务将走静态代理: {self._static_proxies.get('https', '')}") elif self._proxy_fetcher: - logger.info("[douyu] 任务将按任务从代理 API 取新代理") + logger.info("[douyu] 写操作任务将按任务从代理 API 取新代理") def _create_proxy_fetcher(self) -> ProxyFetcher | None: """API 代理模式: 每个任务从代理 API 取 1 个新代理 (与虎牙注册链同款).""" @@ -90,7 +109,13 @@ class DouyuBatchRunnerCore: return {"http": http or https, "https": https or http} def _proxies_for_task(self) -> dict[str, str] | None: - """取本任务出网代理: 静态优先; API 模式每任务取新代理, 失败降级直连并告警.""" + """取本任务出网代理。 + + 写/读分离：仅兑换/锁单/支付/开通/充值/送礼等写操作任务走代理（应对 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: @@ -287,7 +312,8 @@ class DouyuBatchRunnerCore: return {**payload, **self.payload} def _client(self, cookie: str) -> DouyuActivityClient: - """任务 HTTP 客户端: 会话级代理 (覆盖精英手册/和平小店/电竞等全部请求).""" + """任务 HTTP 客户端：写操作任务注入会话级代理（兑换/锁单/支付/充值/送礼）， + 读操作任务直连；见 DOUYU_PROXY_TASK_TYPES。""" return DouyuActivityClient( cookie, logger=lambda msg: self._push_log("debug", msg),