246 lines
8.5 KiB
Python
246 lines
8.5 KiB
Python
"""代理管理模块"""
|
||
|
||
import re
|
||
import time
|
||
import requests
|
||
from typing import Optional
|
||
from loguru import logger
|
||
|
||
|
||
class ProxyManager:
|
||
"""代理管理器"""
|
||
|
||
def __init__(self, api_url: str = ""):
|
||
self.api_url = api_url
|
||
self.current_proxy: Optional[str] = None
|
||
|
||
def get_proxy(self) -> Optional[str]:
|
||
"""
|
||
从代理API获取代理IP，支持API返回多个IP（多行格式），
|
||
逐一验证返回第一个可用的。
|
||
|
||
Returns:
|
||
代理URL，格式: http://ip:port
|
||
"""
|
||
try:
|
||
logger.info("获取代理IP...")
|
||
response = requests.get(self.api_url, timeout=10)
|
||
response.raise_for_status()
|
||
|
||
text = response.text.strip()
|
||
logger.debug(f"代理API响应: {text}")
|
||
|
||
proxy_urls, _ = parse_proxy_response(text)
|
||
|
||
if not proxy_urls:
|
||
logger.warning(f"无法解析代理地址: {text}")
|
||
return None
|
||
|
||
if len(proxy_urls) == 1:
|
||
proxy = proxy_urls[0]
|
||
self.current_proxy = proxy
|
||
logger.info(f"获取到代理: {proxy}")
|
||
return proxy
|
||
|
||
# 多个代理逐一验证，返回第一个可用的
|
||
logger.info(f"获取到 {len(proxy_urls)} 个代理，逐一验证")
|
||
for proxy in proxy_urls:
|
||
if self.verify_proxy(proxy):
|
||
self.current_proxy = proxy
|
||
logger.success(f"可用代理: {proxy}")
|
||
return proxy
|
||
logger.warning(f"代理 {proxy} 不可用，尝试下一个")
|
||
|
||
logger.warning("所有代理均不可用")
|
||
self.current_proxy = None
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取代理失败: {e}")
|
||
return None
|
||
|
||
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 e:
|
||
logger.error(f"代理验证失败: {e}")
|
||
return False
|
||
|
||
|
||
def get_proxy_manager(api_url: str = "") -> ProxyManager:
|
||
"""获取代理管理器实例（每次传入 api_url 时创建新实例，避免全局状态污染）"""
|
||
return ProxyManager(api_url)
|
||
|
||
|
||
def parse_proxy_response(text: str) -> tuple[list[str], Optional[str]]:
|
||
"""
|
||
解析代理API响应，支持返回多个代理地址。
|
||
|
||
Returns:
|
||
(proxy_urls, whitelist_ip)
|
||
- proxy_urls: 解析到的所有代理地址列表（http://ip:port）
|
||
- whitelist_ip: 需要添加到白名单的IP（当API返回白名单错误时），无错误时为 None
|
||
"""
|
||
text = text.strip()
|
||
|
||
# 白名单错误优先检测
|
||
if '添加白名单' in text or '白名单' in text:
|
||
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', text)
|
||
if ip_match:
|
||
return [], ip_match.group(1)
|
||
|
||
# 解析所有 ip:port，支持多行格式
|
||
matches = re.findall(r'(\d+\.\d+\.\d+\.\d+):(\d+)', text)
|
||
proxies = []
|
||
for ip, port in matches:
|
||
# 排除白名单提示中误匹配的
|
||
proxy = f"http://{ip}:{port}"
|
||
proxies.append(proxy)
|
||
|
||
if proxies:
|
||
return proxies, None
|
||
|
||
return [], None
|
||
|
||
|
||
def verify_proxy_url(proxy_url: str, timeout: tuple = (4, 6)) -> tuple[bool, str]:
|
||
"""
|
||
验证代理是否可用。
|
||
|
||
验证优先级：
|
||
1. 斗鱼主站（最相关，能访问斗鱼才是最终目的）
|
||
2. myip（快速 IP 验证）
|
||
3. 百度 IP 查询（备用）
|
||
|
||
Returns:
|
||
(是否可用, 消息)
|
||
"""
|
||
proxies = {'http': proxy_url, 'https': proxy_url}
|
||
targets = [
|
||
('https://www.douyu.com', '斗鱼主站'),
|
||
('https://myip.ipip.net', 'IP验证'),
|
||
('https://qifu-api.baidubce.com/ip/local/geo/v1/district', '百度IP查询'),
|
||
]
|
||
|
||
for url, label in targets:
|
||
try:
|
||
response = requests.get(
|
||
url, proxies=proxies, timeout=timeout,
|
||
headers={'User-Agent': 'Mozilla/5.0'},
|
||
)
|
||
response.raise_for_status()
|
||
return True, f'代理可用 → {label}'
|
||
except Exception as e:
|
||
err_msg = str(e)
|
||
if 'Tunnel connection failed' in err_msg or '503' in err_msg:
|
||
detail = '代理拒绝连接（白名单可能未生效）'
|
||
elif 'timed out' in err_msg.lower():
|
||
detail = '连接超时'
|
||
else:
|
||
detail = type(e).__name__
|
||
logger.debug(f"代理验证 {label} 失败: {detail}")
|
||
continue
|
||
|
||
return False, '代理验证失败（所有目标不可达）'
|
||
|
||
|
||
def resolve_working_proxy(
|
||
api_url: str,
|
||
whitelist_uid: str = "",
|
||
whitelist_ukey: str = "",
|
||
max_attempts: int = 3,
|
||
log_func=None,
|
||
) -> tuple[Optional[str], str]:
|
||
"""
|
||
从代理API获取可用代理，自动处理白名单同步。
|
||
支持API返回多个代理IP，逐一验证直到找到可用的。
|
||
|
||
Args:
|
||
api_url: 代理API地址
|
||
whitelist_uid: 白名单UID（启用白名单时传入）
|
||
whitelist_ukey: 白名单UKEY
|
||
max_attempts: 最大获取尝试次数（默认3次）
|
||
log_func: 日志回调函数 (level, message)
|
||
|
||
Returns:
|
||
(代理URL, 消息)
|
||
"""
|
||
def log(level, msg):
|
||
if log_func:
|
||
log_func(level, msg)
|
||
else:
|
||
getattr(logger, level if level in ('info', 'warning', 'error', 'success') else 'info', logger.info)(msg)
|
||
|
||
synced_whitelist = False
|
||
last_error = ""
|
||
|
||
for attempt in range(1, max_attempts + 1):
|
||
# 重试之间增加退避延迟，避免代理API返回同一个不可用IP
|
||
if attempt > 1:
|
||
delay = min(attempt, 3)
|
||
log('info', f'等待 {delay}s 后重试...')
|
||
time.sleep(delay)
|
||
|
||
log('info', f'代理预检 {attempt}/{max_attempts}: 正在获取代理')
|
||
try:
|
||
response = requests.get(api_url, timeout=10)
|
||
response.raise_for_status()
|
||
text = response.text.strip()
|
||
|
||
proxy_urls, whitelist_ip = parse_proxy_response(text)
|
||
|
||
if proxy_urls:
|
||
log('info', f'获取到 {len(proxy_urls)} 个代理，逐一验证')
|
||
for proxy_url in proxy_urls:
|
||
ok, msg = verify_proxy_url(proxy_url)
|
||
if ok:
|
||
log('success', f'代理预检成功: {proxy_url}')
|
||
return proxy_url, msg
|
||
log('warning', f'代理 {proxy_url} 不可用: {msg}')
|
||
last_error = f'共 {len(proxy_urls)} 个代理均不可用'
|
||
log('warning', f'代理预检 {attempt}/{max_attempts}: {last_error}')
|
||
continue
|
||
|
||
# 代理API返回白名单错误
|
||
if whitelist_ip and not synced_whitelist and whitelist_uid and whitelist_ukey:
|
||
log('warning', f'代理需要白名单IP: {whitelist_ip}，自动同步...')
|
||
from core.douyu.whitelist import WhitelistManager
|
||
manager = WhitelistManager(whitelist_uid, whitelist_ukey)
|
||
ok, sync_msg = manager.sync_ip(whitelist_ip)
|
||
log('success' if ok else 'error', f'白名单同步: {sync_msg}')
|
||
if ok:
|
||
synced_whitelist = True
|
||
log('info', '白名单已更新，等待2秒后重试...')
|
||
time.sleep(2)
|
||
continue
|
||
return None, f'白名单同步失败: {sync_msg}'
|
||
|
||
last_error = f'代理API响应无法解析'
|
||
log('warning', f'代理预检 {attempt}/{max_attempts}: {last_error}: {text[:80]}')
|
||
|
||
except Exception as exc:
|
||
last_error = f'代理API请求失败: {exc}'
|
||
log('warning', f'代理预检 {attempt}/{max_attempts}: {last_error}')
|
||
|
||
return None, f'代理预检失败（{max_attempts}次尝试均失败）: {last_error}'
|