203 lines
7.3 KiB
Python
203 lines
7.3 KiB
Python
"""代理获取、白名单同步、可用性验证的统一流程。"""
|
|
|
|
import threading
|
|
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,
|
|
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)
|
|
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 self._is_stopped():
|
|
return None, "任务已停止"
|
|
if attempt > 1:
|
|
delay = min(attempt - 1, 2)
|
|
if self.log_func:
|
|
self._log("info", f"等待 {delay}s 后重试...")
|
|
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:
|
|
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 = 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
|