优化代理相关, 移除旧的死代码

core/douyu/__init__.py

@@ -2,7 +2,5 @@
 
 from .login import DouyuLogin
 from .email_verifier import EmailVerifier
-from .proxy import ProxyManager
-from .whitelist import WhitelistManager
 
-__all__ = ["DouyuLogin", "EmailVerifier", "ProxyManager", "WhitelistManager"]
+__all__ = ["DouyuLogin", "EmailVerifier"]

core/douyu/login.py

@@ -76,14 +76,8 @@ class DouyuLogin:
         max_proxy_retries: int = 0,
         max_login_retries: int = 0,
         max_total_time: float = 0,
-        whitelist_uid: str = "",
-        whitelist_ukey: str = "",
-        whitelist_platform: str = "xiequ",
-        whitelist_credentials: dict = None,
         proxy_fetcher: Optional[ProxyFetcher] = None,
         stop_event: Optional[threading.Event] = None,
-        # 向后兼容：旧代码传 proxy_manager 时自动转为 proxy_fetcher
-        proxy_manager=None,
     ):
         self.account = account
         self.proxy = proxy
@@ -94,20 +88,9 @@ class DouyuLogin:
         self.stop_event = stop_event
         self.session = requests.Session()
 
-        # 初始化代理获取器
+        self.proxy_fetcher = proxy_fetcher or (
-        if proxy_fetcher:
+            ProxyFetcher(api_url=proxy_api_url) if proxy_api_url else None
-            self.proxy_fetcher = proxy_fetcher
+        )
-        elif proxy_api_url:
-            self.proxy_fetcher = ProxyFetcher(
-                api_url=proxy_api_url,
-                whitelist_platform=whitelist_platform,
-                whitelist_credentials=whitelist_credentials or (
-                    {"uid": whitelist_uid, "ukey": whitelist_ukey}
-                    if whitelist_uid and whitelist_ukey else None
-                ),
-            )
-        else:
-            self.proxy_fetcher = None
 
         self._current_proxy_url: Optional[str] = None
         self._cookie_enrich_error = ""

core/douyu/proxy.py

@@ -1,17 +1,14 @@
-"""代理管理模块兼容导出。"""
+"""代理模块兼容导出。"""
 
 from .proxy_fetcher import ProxyFetcher, get_proxy_fetcher
-from .proxy_manager import ProxyManager, get_proxy_manager
 from .proxy_parser import parse_proxy_response
 from .proxy_resolver import ProxyResolver, resolve_working_proxy
 from .proxy_verifier import verify_proxies_concurrent, verify_proxy_url
 
 __all__ = [
     "ProxyFetcher",
-    "ProxyManager",
     "ProxyResolver",
     "get_proxy_fetcher",
-    "get_proxy_manager",
     "parse_proxy_response",
     "resolve_working_proxy",
     "verify_proxies_concurrent",

core/douyu/proxy_fetcher.py

@@ -1,9 +1,9 @@
-"""简化版代理获取器（替代 ProxyManager）。
+"""简化版代理获取器。
 
 专为短效代理设计：无池、无冷却、无复用。
 每次调用 fetch_new_proxy() 从 API 取 1 个新代理，用完即弃。
 
-线程安全：白名单同步有全局锁 + IP缓存避免重复同步。
+线程安全：白名单同步有全局锁 + 标志位避免重复同步。
 """
 
 import threading
@@ -26,31 +26,30 @@ class ProxyFetcher:
     ):
         self.api_url = api_url
 
-        _wl_platform = whitelist_platform or "xiequ"
-        _wl_credentials = whitelist_credentials
-
         self._whitelist_syncer = (
-            DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials)
+            DouyuWhitelistSyncer(
-            if _wl_credentials
+                platform=whitelist_platform or "xiequ",
+                credentials=whitelist_credentials,
+            )
+            if whitelist_credentials
             else None
         )
-        # 跨线程共享：已同步白名单的IP，避免5个线程重复同步同一个IP
+        # 跨线程共享：是否已同步过白名单，避免多线程重复同步
-        self._synced_ip: Optional[str] = None
+        self._synced = False
         self._sync_lock = threading.Lock()
 
     def fetch_new_proxy(self, max_attempts: int = 3) -> Optional[str]:
         """从代理 API 获取 1 个可用代理，失败返回 None。"""
-        # 快速检查：是否已经同步过白名单
-        skip_sync = False
         if self._whitelist_syncer:
             with self._sync_lock:
-                if self._synced_ip:
+                already_synced = self._synced
-                    skip_sync = True
+        else:
+            already_synced = False
 
         resolver = ProxyResolver(
             api_url=self.api_url,
             whitelist_syncer=self._whitelist_syncer,
-            sync_local_exit_ip=bool(self._whitelist_syncer) and not skip_sync,
+            sync_local_exit_ip=bool(self._whitelist_syncer) and not already_synced,
             sync_whitelist_once=False,
         )
         result, msg = resolver.fetch_verified(
@@ -58,10 +57,9 @@ class ProxyFetcher:
             return_all=False,
         )
 
-        # 记录已同步的IP
         if self._whitelist_syncer and result:
             with self._sync_lock:
-                self._synced_ip = self._synced_ip or "done"
+                self._synced = True
 
         if isinstance(result, str):
             logger.info(f"获取新代理: {result}")

core/douyu/proxy_manager.py

@@ -1,263 +0,0 @@
-"""代理池管理。"""
-
-import threading
-import time
-from typing import Optional
-
-import requests
-from loguru import logger
-
-from .proxy_resolver import ProxyResolver
-from .proxy_whitelist import DouyuWhitelistSyncer
-
-
-class ProxyManager:
-    """代理管理器（带已验证代理池缓存，批次内共享复用）"""
-
-    # 代理失败软隔离配置
-    _COOLDOWN_BASE = 30       # 基础冷却秒数（失败1次）
-    _COOLDOWN_MAX = 120       # 最大冷却秒数
-    _MAX_FAIL_COUNT = 3       # 连续失败此次数后永久移出池
-
-    def __init__(
-        self,
-        api_url: str = "",
-        whitelist_uid: str = "",
-        whitelist_ukey: str = "",
-        whitelist_platform: str = "xiequ",
-        whitelist_credentials: dict = None,
-    ):
-        self.api_url = api_url
-        self.whitelist_uid = whitelist_uid
-        self.whitelist_ukey = whitelist_ukey
-        self.whitelist_platform = whitelist_platform
-        self.whitelist_credentials = whitelist_credentials
-        self.current_proxy: Optional[str] = None
-        self._cond = threading.Condition()
-        # 已验证可用的代理池: {proxy_url: validated_timestamp}
-        self._verified_pool: dict[str, float] = {}
-        # 正在使用中的代理（取走但未归还），避免并发账号用同一个代理
-        self._in_use: set[str] = set()
-        # 代理失败记录: {proxy_url: {"count": 失败次数, "cooldown_until": 冷却到期时间戳}}
-        self._fail_records: dict[str, dict] = {}
-        self._pool_ttl = 180
-        self._fetching = False
-
-        # 构建白名单同步器（优先使用新参数，向后兼容旧参数）
-        _wl_platform = whitelist_platform or "xiequ"
-        _wl_credentials = whitelist_credentials
-        if not _wl_credentials and whitelist_uid and whitelist_ukey:
-            _wl_platform = "xiequ"
-            _wl_credentials = {"uid": whitelist_uid, "ukey": whitelist_ukey}
-
-        self._whitelist_syncer = (
-            DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials)
-            if _wl_credentials
-            else None
-        )
-
-    def _is_cooling_down(self, proxy_url: str) -> bool:
-        """检查代理是否在冷却期内（调用前需持有锁）。"""
-        record = self._fail_records.get(proxy_url)
-        if not record:
-            return False
-        return time.time() < record.get("cooldown_until", 0)
-
-    def _pick_from_pool_locked(self) -> Optional[str]:
-        """从池中取一个未过期、未冷却、未在使用的代理（调用前需持有锁）。"""
-        now = time.time()
-        # 清理过期代理
-        expired = [proxy for proxy, ts in self._verified_pool.items() if now - ts > self._pool_ttl]
-        for proxy in expired:
-            del self._verified_pool[proxy]
-            self._in_use.discard(proxy)
-
-        # 优先选未冷却、未在用的代理
-        for proxy in self._verified_pool:
-            if proxy not in self._in_use and not self._is_cooling_down(proxy):
-                self._in_use.add(proxy)
-                self.current_proxy = proxy
-                return proxy
-
-        # 退而求其次：所有未在用的（含冷却中的），也比没有强
-        for proxy in self._verified_pool:
-            if proxy not in self._in_use:
-                self._in_use.add(proxy)
-                self.current_proxy = proxy
-                return proxy
-        return None
-
-    def get_proxy(self, max_attempts: int = 5, max_retries: int = 3) -> Optional[str]:
-        """
-        获取可用代理 IP，优先从已验证代理池复用。
-
-        线程安全：池空时只有一个线程调 API 获取并验证所有代理入池，
-        其他线程等待后复用。等待线程被唤醒后会循环尝试取代理，
-        避免多个线程竞争拿到同一个代理。
-
-        Args:
-            max_attempts: 代理 API 获取尝试次数
-            max_retries: 等待线程获取代理的重试次数（防止唤醒后池仍为空直接放弃）
-        """
-        # ── 快速路径：池中有可用代理 ──
-        with self._cond:
-            proxy = self._pick_from_pool_locked()
-            if proxy:
-                logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})")
-                return proxy
-
-        # ── 池空：需要获取新代理 ──
-        for retry in range(max_retries):
-            with self._cond:
-                # 再次检查（可能其他线程刚充盈了池）
-                proxy = self._pick_from_pool_locked()
-                if proxy:
-                    logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})")
-                    return proxy
-
-                if self._fetching:
-                    logger.debug("代理池空，等待其他线程获取...")
-                    self._cond.wait(timeout=60)
-                    # 等待后循环重试取代理（不再只尝试一次）
-                    proxy = self._pick_from_pool_locked()
-                    if proxy:
-                        logger.debug(f"等待后从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})")
-                        return proxy
-                    # 池仍为空，但可能有其他等待线程会去获取，继续重试
-                    continue
-
-                # 当前线程负责获取
-                self._fetching = True
-
-            try:
-                proxies_all, msg = self._fetch_and_verify_all(max_attempts)
-            finally:
-                with self._cond:
-                    self._fetching = False
-                    self._cond.notify_all()
-
-            if proxies_all:
-                with self._cond:
-                    now = time.time()
-                    for proxy in proxies_all:
-                        self._verified_pool[proxy] = now
-                    first = proxies_all[0]
-                    self._in_use.add(first)
-                    self.current_proxy = first
-                    logger.info(f"代理池充盈 {len(proxies_all)} 个可用代理: {first} ...")
-                    return first
-
-            logger.warning(f"获取代理失败: {msg}")
-            # 获取失败，等一小段时间后重试
-            time.sleep(2)
-
-        logger.warning(f"获取代理失败（已重试 {max_retries} 次）")
-        return None
-
-    def _fetch_and_verify_all(self, max_attempts: int) -> tuple[Optional[list[str]], str]:
-        """调代理 API 获取一批代理，并发验证。
-
-        对短效代理友好：拿到第一个可用代理就立即返回，
-        不再等所有代理验证完毕（验证期间短效代理会过期浪费）。
-        """
-        resolver = ProxyResolver(
-            api_url=self.api_url,
-            whitelist_syncer=self._whitelist_syncer,
-            sync_local_exit_ip=bool(self._whitelist_syncer),
-            sync_whitelist_once=False,
-        )
-        # return_all=False: 找到第一个可用就返回，节省短效代理时间
-        available, msg = resolver.fetch_verified(max_attempts=max_attempts, return_all=False)
-        if isinstance(available, str):
-            return [available], msg
-        if isinstance(available, list):
-            return available, msg
-        return None, msg
-
-    def mark_bad(self, proxy_url: str) -> None:
-        """
-        标记代理失败：软隔离（冷却期），而非永久移除。
-
-        - 失败 < _MAX_FAIL_COUNT 次：进入冷却期，冷却后可重新入池
-        - 连续失败 >= _MAX_FAIL_COUNT 次：永久移出池
-        """
-        with self._cond:
-            self._in_use.discard(proxy_url)
-            if self.current_proxy == proxy_url:
-                self.current_proxy = None
-
-            record = self._fail_records.get(proxy_url, {"count": 0, "cooldown_until": 0})
-            record["count"] = record.get("count", 0) + 1
-            fail_count = record["count"]
-
-            if fail_count >= self._MAX_FAIL_COUNT:
-                # 连续失败超限，永久移出
-                self._verified_pool.pop(proxy_url, None)
-                self._fail_records.pop(proxy_url, None)
-                logger.info(
-                    f"代理连续失败 {fail_count} 次，永久移出池: {proxy_url} "
-                    f"(池剩余 {len(self._verified_pool)})"
-                )
-            else:
-                # 软隔离：指数退避冷却
-                cooldown = min(self._COOLDOWN_BASE * (2 ** (fail_count - 1)), self._COOLDOWN_MAX)
-                record["cooldown_until"] = time.time() + cooldown
-                self._fail_records[proxy_url] = record
-                logger.info(
-                    f"代理失败 {fail_count} 次，冷却 {cooldown}s: {proxy_url} "
-                    f"(池剩余 {len(self._verified_pool)})"
-                )
-
-            self._cond.notify_all()
-
-    def release_proxy(self, proxy_url: str) -> None:
-        """归还代理到池（登录完成后调用，让其他账号可以复用）。"""
-        with self._cond:
-            self._in_use.discard(proxy_url)
-            # 成功归还时清除失败记录
-            self._fail_records.pop(proxy_url, None)
-            self._cond.notify_all()
-
-    def get_proxies_dict(self, proxy: str = None) -> dict:
-        """获取 requests 使用的 proxies 字典。"""
-        proxy = proxy or self.current_proxy
-        if proxy:
-            return {'http': proxy, 'https': proxy}
-        return {}
-
-    def verify_proxy(self, proxy: str = None) -> bool:
-        """验证当前代理是否可用。"""
-        proxy = proxy or self.current_proxy
-        if not proxy:
-            return False
-        try:
-            response = requests.get(
-                'https://httpbin.org/ip',
-                proxies={'http': proxy, 'https': proxy},
-                timeout=10,
-            )
-            if response.status_code == 200:
-                data = response.json()
-                logger.info(f"代理验证成功，当前IP: {data.get('origin')}")
-                return True
-            return False
-        except Exception as exc:
-            logger.error(f"代理验证失败: {exc}")
-            return False
-
-
-def get_proxy_manager(
-    api_url: str = "",
-    whitelist_uid: str = "",
-    whitelist_ukey: str = "",
-    whitelist_platform: str = "xiequ",
-    whitelist_credentials: dict = None,
-) -> ProxyManager:
-    """获取代理管理器实例。"""
-    return ProxyManager(
-        api_url,
-        whitelist_uid=whitelist_uid,
-        whitelist_ukey=whitelist_ukey,
-        whitelist_platform=whitelist_platform,
-        whitelist_credentials=whitelist_credentials,
-    )

core/douyu/proxy_resolver.py

@@ -145,30 +145,18 @@ class ProxyResolver:
 
 def resolve_working_proxy(
     api_url: str,
-    whitelist_uid: str = "",
-    whitelist_ukey: str = "",
     whitelist_platform: str = "xiequ",
     whitelist_credentials: dict = None,
     max_attempts: int = 4,
     log_func: Optional[LogFunc] = None,
 ) -> tuple[Optional[str], str]:
-    """
+    """从代理 API 获取可用代理，自动处理白名单同步。"""
-    从代理 API 获取可用代理，自动处理白名单同步。
-
-    支持：
-      - 新参数：whitelist_platform + whitelist_credentials
-      - 旧参数：whitelist_uid + whitelist_ukey（向后兼容）
-    """
-    # 构建白名单同步器
-    _wl_platform = whitelist_platform or "xiequ"
-    _wl_credentials = whitelist_credentials
-    if not _wl_credentials and whitelist_uid and whitelist_ukey:
-        _wl_platform = "xiequ"
-        _wl_credentials = {"uid": whitelist_uid, "ukey": whitelist_ukey}
-
     syncer = (
-        DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials)
+        DouyuWhitelistSyncer(
-        if _wl_credentials
+            platform=whitelist_platform or "xiequ",
+            credentials=whitelist_credentials,
+        )
+        if whitelist_credentials
         else None
     )
     resolver = ProxyResolver(

core/douyu/whitelist.py

@@ -1,89 +0,0 @@
-"""代理IP白名单管理模块（兼容层）。
-
-核心逻辑已迁移至 proxy_platforms/xiequ.py，本模块仅做向后兼容包装。
-旧代码仍可 from .whitelist import WhitelistManager 正常使用。
-"""
-
-from typing import Optional
-
-from loguru import logger
-
-from .proxy_platforms.xiequ import XiequAdapter
-from .proxy_platforms.base import (
-    BaseWhitelistAdapter,
-    _get_local_exit_ip,
-    get_exit_ip_via_proxy,
-)
-
-# 全局锁保留（基类内部使用，此处导出供旧代码引用）
-from .proxy_platforms.base import _whitelist_sync_lock
-
-
-class WhitelistManager:
-    """携趣代理IP白名单管理器（兼容层，委托给 XiequAdapter）。"""
-
-    MEMO_PREFIX = XiequAdapter.MEMO_PREFIX
-    BASE_URL = XiequAdapter.BASE_URL
-
-    def __init__(self, uid: str, ukey: str):
-        self._adapter = XiequAdapter({"uid": uid, "ukey": ukey})
-        self.uid = uid
-        self.ukey = ukey
-
-    @property
-    def memo(self) -> str:
-        return self._adapter.memo
-
-    def _build_url(self, **params) -> str:
-        """构建请求URL（兼容旧代码直接调用）。"""
-        return self._adapter._build_url(**params)
-
-    def get_whitelist_json(self) -> list[dict]:
-        """
-        获取白名单列表（JSON格式，协固原始格式）。
-
-        Returns:
-            [{"IP": "x.x.x.x", "MEMO": "备注"}, ...]
-        """
-        records = self._adapter.get_whitelist()
-        # 将统一格式映射回调固原始格式
-        return [{"IP": r["ip"], "MEMO": r["memo"]} for r in records]
-
-    def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]:
-        """添加IP到白名单。"""
-        return self._adapter.add_ip(ip, retry=retry)
-
-    def delete_ip(self, ip: str) -> bool:
-        """删除指定IP。"""
-        ok, _ = self._adapter.delete_ip(ip)
-        return ok
-
-    def get_memo_ip(self) -> Optional[str]:
-        """获取当前备注对应的IP。"""
-        records = self._adapter.get_whitelist()
-        for record in records:
-            if record.get("memo") == self.memo:
-                return record.get("ip")
-        return None
-
-    def get_memo_records(self) -> list[dict]:
-        """获取当前备注的所有记录（协固原始格式）。"""
-        records = self._adapter.get_whitelist()
-        return [
-            {"IP": r["ip"], "MEMO": r["memo"]}
-            for r in records
-            if r.get("memo") == self.memo
-        ]
-
-    def sync_ip(self, current_ip: str, keep_recent: int = 3) -> tuple[bool, str]:
-        """同步白名单IP。"""
-        return self._adapter.sync_ip(current_ip, keep_recent)
-
-    def test_connection(self) -> tuple[bool, str]:
-        """测试白名单API连接。"""
-        return self._adapter.test_connection()
-
-
-def get_local_exit_ip() -> Optional[str]:
-    """获取本机当前公网出口IP（不走代理）。"""
-    return _get_local_exit_ip()