diff --git a/core/douyu/__init__.py b/core/douyu/__init__.py index 9b6506a..5194f20 100644 --- a/core/douyu/__init__.py +++ b/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"] diff --git a/core/douyu/login.py b/core/douyu/login.py index 85b4d82..2c0820a 100644 --- a/core/douyu/login.py +++ b/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() - # 初始化代理获取器 - if proxy_fetcher: - 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.proxy_fetcher = proxy_fetcher or ( + ProxyFetcher(api_url=proxy_api_url) if proxy_api_url else None + ) self._current_proxy_url: Optional[str] = None self._cookie_enrich_error = "" diff --git a/core/douyu/proxy.py b/core/douyu/proxy.py index c3521f9..eaa25a2 100644 --- a/core/douyu/proxy.py +++ b/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", diff --git a/core/douyu/proxy_fetcher.py b/core/douyu/proxy_fetcher.py index 017d7b3..3195147 100644 --- a/core/douyu/proxy_fetcher.py +++ b/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) - if _wl_credentials + DouyuWhitelistSyncer( + 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: - skip_sync = True + already_synced = self._synced + 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}") diff --git a/core/douyu/proxy_manager.py b/core/douyu/proxy_manager.py deleted file mode 100644 index 99566a4..0000000 --- a/core/douyu/proxy_manager.py +++ /dev/null @@ -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, - ) diff --git a/core/douyu/proxy_resolver.py b/core/douyu/proxy_resolver.py index ee1d626..967f3eb 100644 --- a/core/douyu/proxy_resolver.py +++ b/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 获取可用代理，自动处理白名单同步。 - - 支持： - - 新参数：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} - + """从代理 API 获取可用代理，自动处理白名单同步。""" syncer = ( - DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials) - if _wl_credentials + DouyuWhitelistSyncer( + platform=whitelist_platform or "xiequ", + credentials=whitelist_credentials, + ) + if whitelist_credentials else None ) resolver = ProxyResolver( diff --git a/core/douyu/whitelist.py b/core/douyu/whitelist.py deleted file mode 100644 index 1bd4832..0000000 --- a/core/douyu/whitelist.py +++ /dev/null @@ -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()