yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

继续优化登录逻辑

Browse Source
This commit is contained in:
yml2213
2026-06-25 08:40:00 +08:00
parent a62e7c192f
commit 0b4fe1daa2
6 changed files with 105 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/douyu/login.py
+54 -42
View File
@@ -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]:
"""
core/douyu/proxy_fetcher.py
+30
View File
@@ -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,
core/douyu/proxy_resolver.py
+20 -1
View File
@@ -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:
web/backend/routers/login.py
-1
View File
@@ -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,
web/backend/schemas.py
-1
View File
@@ -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
web/backend/services/login_service.py
+1 -3
View File
@@ -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,