173 lines
6.3 KiB
Python
173 lines
6.3 KiB
Python
"""代理获取、白名单同步、可用性验证的统一流程。"""
|
|
|
|
import time
|
|
from typing import Callable, Optional, Protocol
|
|
|
|
import requests
|
|
from loguru import logger
|
|
|
|
from .proxy_parser import parse_proxy_response
|
|
from .proxy_verifier import verify_proxies_concurrent
|
|
from .proxy_whitelist import DouyuWhitelistSyncer
|
|
|
|
LogFunc = Callable[[str, str], None]
|
|
|
|
|
|
class WhitelistSyncer(Protocol):
|
|
"""代理解析流程需要的白名单能力。"""
|
|
|
|
def sync_ip(self, ip: str) -> tuple[bool, str]:
|
|
...
|
|
|
|
def get_local_exit_ip(self) -> Optional[str]:
|
|
...
|
|
|
|
|
|
class ProxyResolver:
|
|
"""从代理 API 获取并验证代理，按需同步白名单。"""
|
|
|
|
def __init__(
|
|
self,
|
|
api_url: str,
|
|
whitelist_syncer: Optional[WhitelistSyncer] = None,
|
|
log_func: Optional[LogFunc] = None,
|
|
sync_local_exit_ip: bool = False,
|
|
sync_whitelist_once: bool = True,
|
|
):
|
|
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._last_synced_ip: Optional[str] = None
|
|
self._has_synced_whitelist = False
|
|
|
|
def _log(self, level: str, message: str) -> None:
|
|
if self.log_func:
|
|
self.log_func(level, message)
|
|
return
|
|
log_method = getattr(
|
|
logger,
|
|
level if level in ('debug', 'info', 'warning', 'error', 'success') else 'info',
|
|
logger.info,
|
|
)
|
|
log_method(message)
|
|
|
|
def _sync_ip(self, ip: str) -> tuple[bool, str]:
|
|
if not self.whitelist_syncer:
|
|
return False, '未配置白名单 UID/UKEY'
|
|
ok, sync_msg = self.whitelist_syncer.sync_ip(ip)
|
|
if ok:
|
|
self._last_synced_ip = ip
|
|
self._has_synced_whitelist = True
|
|
return ok, sync_msg
|
|
|
|
def _sync_local_exit_ip_if_needed(self, attempt: int) -> None:
|
|
if not self.sync_local_exit_ip or not self.whitelist_syncer:
|
|
return
|
|
|
|
local_ip = self.whitelist_syncer.get_local_exit_ip()
|
|
if local_ip and local_ip != self._last_synced_ip:
|
|
self._log('info', f"[尝试 {attempt}] 检测出口IP: {local_ip}，同步白名单...")
|
|
ok, sync_msg = self._sync_ip(local_ip)
|
|
if ok:
|
|
self._log('info', f"白名单同步成功: {sync_msg}")
|
|
else:
|
|
self._log('warning', f"白名单同步失败: {sync_msg}")
|
|
elif local_ip == self._last_synced_ip:
|
|
self._log('debug', f"[尝试 {attempt}] 出口IP未变: {local_ip}")
|
|
|
|
def fetch_verified(
|
|
self,
|
|
max_attempts: int = 4,
|
|
return_all: bool = False,
|
|
) -> tuple[Optional[str | list[str]], str]:
|
|
"""
|
|
获取并验证代理。
|
|
|
|
Args:
|
|
max_attempts: 最大尝试次数
|
|
return_all: True 返回所有可用代理，False 返回第一个可用代理
|
|
"""
|
|
last_error = ""
|
|
|
|
for attempt in range(1, max_attempts + 1):
|
|
if attempt > 1:
|
|
delay = min(attempt - 1, 2)
|
|
if self.log_func:
|
|
self._log('info', f'等待 {delay}s 后重试...')
|
|
time.sleep(delay)
|
|
|
|
self._sync_local_exit_ip_if_needed(attempt)
|
|
self._log('info', f'代理预检 {attempt}/{max_attempts}: 正在获取代理')
|
|
|
|
try:
|
|
response = requests.get(self.api_url, timeout=10)
|
|
response.raise_for_status()
|
|
text = response.text.strip()
|
|
|
|
proxy_urls, whitelist_ip = parse_proxy_response(text)
|
|
|
|
if proxy_urls:
|
|
self._log('info', f'获取到 {len(proxy_urls)} 个代理，并发验证')
|
|
available, msg = verify_proxies_concurrent(proxy_urls, return_all=return_all)
|
|
if available:
|
|
if not return_all:
|
|
self._log('success', f'代理预检成功: {available}')
|
|
return available, msg
|
|
last_error = msg
|
|
self._log('warning', f"代理预检 {attempt}/{max_attempts}: {last_error}")
|
|
continue
|
|
|
|
if whitelist_ip and self.whitelist_syncer:
|
|
if self.sync_whitelist_once and self._has_synced_whitelist:
|
|
last_error = f'白名单已同步但代理API仍返回白名单错误: {whitelist_ip}'
|
|
self._log('warning', last_error)
|
|
continue
|
|
|
|
self._log('warning', f'代理需要白名单IP: {whitelist_ip}，自动同步...')
|
|
ok, sync_msg = self._sync_ip(whitelist_ip)
|
|
self._log('success' if ok else 'error', f'白名单同步: {sync_msg}')
|
|
if ok:
|
|
self._log('info', '白名单已更新，立即重试...')
|
|
continue
|
|
return None, f'白名单同步失败: {sync_msg}'
|
|
|
|
last_error = '代理API响应无法解析'
|
|
self._log('warning', f"代理预检 {attempt}/{max_attempts}: {last_error}: {text[:80]}")
|
|
|
|
except Exception as exc:
|
|
last_error = f'代理API请求失败: {exc}'
|
|
self._log('warning', f"代理预检 {attempt}/{max_attempts}: {last_error}")
|
|
|
|
return None, f'代理预检失败（{max_attempts}次尝试均失败）: {last_error}'
|
|
|
|
|
|
def resolve_working_proxy(
|
|
api_url: str,
|
|
whitelist_platform: str = "xiequ",
|
|
whitelist_credentials: dict = None,
|
|
max_attempts: int = 4,
|
|
log_func: Optional[LogFunc] = None,
|
|
) -> tuple[Optional[str], str]:
|
|
"""从代理 API 获取可用代理，自动处理白名单同步。"""
|
|
syncer = (
|
|
DouyuWhitelistSyncer(
|
|
platform=whitelist_platform or "xiequ",
|
|
credentials=whitelist_credentials,
|
|
)
|
|
if whitelist_credentials
|
|
else None
|
|
)
|
|
resolver = ProxyResolver(
|
|
api_url=api_url,
|
|
whitelist_syncer=syncer,
|
|
log_func=log_func,
|
|
sync_local_exit_ip=False,
|
|
sync_whitelist_once=True,
|
|
)
|
|
proxy, msg = resolver.fetch_verified(max_attempts=max_attempts, return_all=False)
|
|
if isinstance(proxy, list):
|
|
return (proxy[0] if proxy else None), msg
|
|
return proxy, msg
|