diff --git a/core/douyu/login.py b/core/douyu/login.py index 2c0820a..eff5893 100644 --- a/core/douyu/login.py +++ b/core/douyu/login.py @@ -66,6 +66,7 @@ class DouyuLogin: "state=https%3A%2F%2Fwww.douyu.com%2Fdirectory" ) REQUEST_TIMEOUT = (10, 30) + MAX_STATIC_RETRY = 3 # 静态代理失败上限（无法切换代理时的最大重试次数） def __init__( self, @@ -73,7 +74,6 @@ class DouyuLogin: proxy: Optional[str | Mapping[str, str]] = None, proxy_api_url: Optional[str] = None, timeout: tuple[float, float] = REQUEST_TIMEOUT, - max_proxy_retries: int = 0, max_login_retries: int = 0, max_total_time: float = 0, proxy_fetcher: Optional[ProxyFetcher] = None, @@ -82,7 +82,6 @@ class DouyuLogin: self.account = account self.proxy = proxy self.timeout = timeout - self.max_proxy_retries = max_proxy_retries # 0=无限切换直到成功 self.max_login_retries = max_login_retries # 0=无限重试直到成功 self.max_total_time = max_total_time # 0=不限制单账号登录总时长 self.stop_event = stop_event @@ -94,6 +93,7 @@ class DouyuLogin: self._current_proxy_url: Optional[str] = None self._cookie_enrich_error = "" + self._static_retry_count = 0 self._setup_session() def _is_stopped(self) -> bool: @@ -183,50 +183,41 @@ class DouyuLogin: parsed = urlsplit(url) return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) - def _request(self, method: str, url: str, max_retries: int = 2, **kwargs) -> requests.Response: + def _request(self, method: str, url: str, **kwargs) -> requests.Response: """ 统一发送请求，附带分段超时和更明确的错误信息。 - 代理连接失败时直接抛异常，由 login 整体重试换新代理。 + 代理失败/超时直接抛异常，由 login 整体重试换新代理（短效代理失效后重试同一个无意义）。 """ timeout = kwargs.pop('timeout', self.timeout) safe_url = self._safe_url(url) - for attempt in range(max_retries): - self._ensure_not_stopped() - started = time.monotonic() + self._ensure_not_stopped() + started = time.monotonic() - try: - response = self.session.request(method, url, timeout=timeout, **kwargs) - elapsed = time.monotonic() - started - logger.debug( - f"{method.upper()} {safe_url} -> {response.status_code} " - f"({elapsed:.2f}s)" - ) - return response - except requests.Timeout as exc: - elapsed = time.monotonic() - started - raise TimeoutError( - f"{method.upper()} {safe_url} 超时，耗时 {elapsed:.1f}s，" - f"timeout={timeout}" - ) from exc - except requests.ConnectionError as exc: - elapsed = time.monotonic() - started - err_str = str(exc) - is_proxy_err = "proxy" in err_str.lower() or "Proxy" in type(exc).__name__ - if is_proxy_err and attempt < max_retries - 1: - logger.warning(f"代理连接失败，尝试 {attempt + 1}/{max_retries}: {self._truncate_error(err_str)}") - self._sleep_interruptible(1) - continue - raise ConnectionError( - f"{method.upper()} {safe_url} 连接失败: {self._truncate_error(err_str)}" - ) from exc - except requests.RequestException as exc: - elapsed = time.monotonic() - started - raise ConnectionError( - f"{method.upper()} {safe_url} 请求失败，耗时 {elapsed:.1f}s: {exc}" - ) from exc - - raise ConnectionError(f"{method.upper()} {safe_url} 连接失败，已重试 {max_retries} 次") + try: + response = self.session.request(method, url, timeout=timeout, **kwargs) + elapsed = time.monotonic() - started + logger.debug( + f"{method.upper()} {safe_url} -> {response.status_code} " + f"({elapsed:.2f}s)" + ) + return response + except requests.Timeout as exc: + elapsed = time.monotonic() - started + raise TimeoutError( + f"{method.upper()} {safe_url} 超时，耗时 {elapsed:.1f}s，" + f"timeout={timeout}" + ) from exc + except requests.ConnectionError as exc: + err_str = str(exc) + raise ConnectionError( + f"{method.upper()} {safe_url} 连接失败: {self._truncate_error(err_str)}" + ) from exc + except requests.RequestException as exc: + elapsed = time.monotonic() - started + raise ConnectionError( + f"{method.upper()} {safe_url} 请求失败，耗时 {elapsed:.1f}s: {exc}" + ) from exc def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict: """请求 JSON 接口，并在响应异常时输出可定位的信息。""" @@ -274,7 +265,11 @@ class DouyuLogin: else: logger.info(f"登录整体重试 {attempt} (无限重试)，换新代理从头开始") # 重试前：取新代理 + 重置 session - self._prepare_retry() + if not self._prepare_retry(): + return LoginResult( + success=False, + message=f"静态代理连续失败 {self.MAX_STATIC_RETRY} 次，无法切换代理", + ) try: # 1️⃣ 第一次登录（获取极验参数） @@ -330,14 +325,31 @@ class DouyuLogin: # 所有重试耗尽或超时 return LoginResult(success=False, message=str(e)) - def _prepare_retry(self) -> None: - """重试前准备：取新代理、重置 session cookies。""" + def _prepare_retry(self) -> bool: + """重试前准备：取新代理、重置 session cookies。 + + Returns: + True - 可以继续重试 + False - 静态代理已连续失败 MAX_STATIC_RETRY 次，无法切换代理，应放弃 + """ self.session.cookies.clear() if self.proxy_fetcher: new_proxy = self.proxy_fetcher.fetch_new_proxy() if new_proxy: self._apply_proxy(new_proxy) logger.info(f"重试换新代理: {new_proxy}") + return True + if self.proxy: + self._static_retry_count += 1 + if self._static_retry_count > self.MAX_STATIC_RETRY: + logger.error( + f"静态代理连续失败 {self.MAX_STATIC_RETRY} 次，无法切换代理" + ) + return False + logger.warning( + f"静态代理失败重试 {self._static_retry_count}/{self.MAX_STATIC_RETRY}" + ) + return True def _first_login(self) -> Tuple[str, str, str, dict]: """ diff --git a/core/douyu/proxy_fetcher.py b/core/douyu/proxy_fetcher.py index abd0d0e..0c8f779 100644 --- a/core/douyu/proxy_fetcher.py +++ b/core/douyu/proxy_fetcher.py @@ -8,6 +8,8 @@ 时由 ProxyResolver 内部被动同步兜底。 """ +import threading +import time from typing import Optional from loguru import logger @@ -19,13 +21,18 @@ from .proxy_whitelist import DouyuWhitelistSyncer class ProxyFetcher: """每次从代理 API 取 1 个新代理，无缓存无复用。""" + # 代理 API 最小调用间隔（秒）——防止高并发 + 多轮重试把代理 API 打爆 + MIN_FETCH_INTERVAL = 1.5 + def __init__( self, api_url: str, whitelist_platform: str = "xiequ", whitelist_credentials: dict = None, + stop_event: Optional[threading.Event] = None, ): self.api_url = api_url + self.stop_event = stop_event self._whitelist_syncer = ( DouyuWhitelistSyncer( @@ -36,6 +43,10 @@ class ProxyFetcher: else None ) + # 节流：跨线程共享的"上次取代理时刻" + self._last_fetch_at = 0.0 + self._fetch_lock = threading.Lock() + def warmup_whitelist(self) -> tuple[bool, str]: """批次启动前调用一次：把当前本地出口 IP 同步到白名单。 @@ -60,12 +71,31 @@ class ProxyFetcher: 不主动检测/同步本地出口 IP；如果代理 API 返回"白名单错误"， 由 ProxyResolver 内部被动同步兜底。 + + 进程内节流：高并发场景下任意两次调用至少间隔 MIN_FETCH_INTERVAL 秒， + 避免代理 API 被限频。 """ + with self._fetch_lock: + elapsed = time.monotonic() - self._last_fetch_at + if elapsed < self.MIN_FETCH_INTERVAL: + wait = self.MIN_FETCH_INTERVAL - elapsed + logger.debug(f"代理 API 节流等待 {wait:.2f}s") + if self.stop_event: + if self.stop_event.wait(wait): + return None + else: + time.sleep(wait) + self._last_fetch_at = time.monotonic() + + if self.stop_event and self.stop_event.is_set(): + return None + resolver = ProxyResolver( api_url=self.api_url, whitelist_syncer=self._whitelist_syncer, sync_local_exit_ip=False, sync_whitelist_once=False, + stop_event=self.stop_event, ) result, msg = resolver.fetch_verified( max_attempts=max_attempts, diff --git a/core/douyu/proxy_resolver.py b/core/douyu/proxy_resolver.py index 967f3eb..027cea8 100644 --- a/core/douyu/proxy_resolver.py +++ b/core/douyu/proxy_resolver.py @@ -1,5 +1,6 @@ """代理获取、白名单同步、可用性验证的统一流程。""" +import threading import time from typing import Callable, Optional, Protocol @@ -33,15 +34,28 @@ class ProxyResolver: log_func: Optional[LogFunc] = None, sync_local_exit_ip: bool = False, sync_whitelist_once: bool = True, + stop_event: Optional[threading.Event] = None, ): self.api_url = api_url self.whitelist_syncer = whitelist_syncer self.log_func = log_func self.sync_local_exit_ip = sync_local_exit_ip self.sync_whitelist_once = sync_whitelist_once + self.stop_event = stop_event self._last_synced_ip: Optional[str] = None self._has_synced_whitelist = False + def _is_stopped(self) -> bool: + return bool(self.stop_event and self.stop_event.is_set()) + + def _wait_or_stopped(self, seconds: float) -> bool: + """等待 seconds 秒，期间收到停止信号返回 True。""" + if self.stop_event: + return self.stop_event.wait(seconds) + if seconds > 0: + time.sleep(seconds) + return False + def _log(self, level: str, message: str) -> None: if self.log_func: self.log_func(level, message) @@ -92,13 +106,18 @@ class ProxyResolver: last_error = "" for attempt in range(1, max_attempts + 1): + if self._is_stopped(): + return None, '任务已停止' if attempt > 1: delay = min(attempt - 1, 2) if self.log_func: self._log('info', f'等待 {delay}s 后重试...') - time.sleep(delay) + if self._wait_or_stopped(delay): + return None, '任务已停止' self._sync_local_exit_ip_if_needed(attempt) + if self._is_stopped(): + return None, '任务已停止' self._log('info', f'代理预检 {attempt}/{max_attempts}: 正在获取代理') try: diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index 05758eb..be91bf8 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -54,7 +54,6 @@ async def create_batch( account_ids=valid_ids, created_by=current.id, creator_permissions=get_user_permissions(current), - max_proxy_retries=req.max_proxy_retries, max_login_retries=req.max_login_retries, max_total_time=req.max_total_time, proxy_config=proxy, diff --git a/web/backend/schemas.py b/web/backend/schemas.py index e3e1643..51f1418 100644 --- a/web/backend/schemas.py +++ b/web/backend/schemas.py @@ -124,7 +124,6 @@ class AccountOut(BaseModel): # ---- 登录任务 ---- class LoginBatchRequest(BaseModel): account_ids: list[int] - max_proxy_retries: int = 0 # 代理切换次数，0=无限切换直到成功 max_login_retries: int = 0 # 登录整体重试次数，0=无限重试直到成功 max_total_time: float = 0 # 单账号登录总时长上限，0=不限制 concurrency: int = 3 # 并发数，1-10 diff --git a/web/backend/services/login_service.py b/web/backend/services/login_service.py index c8e5378..8d32233 100644 --- a/web/backend/services/login_service.py +++ b/web/backend/services/login_service.py @@ -25,7 +25,6 @@ class LoginBatchRunner: account_ids: list[int], created_by: int, creator_permissions: list[str], - max_proxy_retries: int = 0, max_login_retries: int = 0, max_total_time: float = 0, proxy_config: Optional[ProxyConfigModel] = None, @@ -37,7 +36,6 @@ class LoginBatchRunner: self.account_ids = account_ids self.created_by = created_by self.creator_permissions = creator_permissions - self.max_proxy_retries = max_proxy_retries # 限制最大重试次数，避免无限重试导致资源耗尽 # 0 表示使用默认值 20，其他值保持原样 self.max_login_retries = max_login_retries if max_login_retries > 0 else 20 @@ -68,6 +66,7 @@ class LoginBatchRunner: api_url=proxy_config.api_url, whitelist_platform=wl_platform, whitelist_credentials=wl_credentials, + stop_event=self._stop, ) def stop(self): @@ -151,7 +150,6 @@ class LoginBatchRunner: loginer = DouyuLogin( account, proxy=proxy_dict, - max_proxy_retries=self.max_proxy_retries, max_login_retries=self.max_login_retries, max_total_time=self.max_total_time, proxy_fetcher=self._shared_proxy_fetcher,