yml2213
2ccc2a9336
问题：get_proxy每次返回池中第一个代理，导致并发账号用同一个代理， 容易触发斗鱼/极验风控。 修复： - 增加_in_use集合，get_proxy取代理时标记为使用中，避免重复分配 - 代理池空但所有代理都在使用中时，允许复用（兜底） - 新增release_proxy方法，登录完成后归还代理到池 - login.py的finally块中调用release_proxy，确保成功/失败都归还 - mark_bad同时从_in_use移除
471 lines
19 KiB
Python
471 lines
19 KiB
Python
"""代理管理模块"""
|
||
|
||
import json
|
||
import re
|
||
import threading
|
||
import time
|
||
import requests
|
||
from typing import Optional
|
||
from loguru import logger
|
||
|
||
|
||
class ProxyManager:
|
||
"""代理管理器（带已验证代理池缓存，批次内共享复用）"""
|
||
|
||
def __init__(self, api_url: str = "", whitelist_uid: str = "", whitelist_ukey: str = ""):
|
||
self.api_url = api_url
|
||
self.whitelist_uid = whitelist_uid
|
||
self.whitelist_ukey = whitelist_ukey
|
||
self.current_proxy: Optional[str] = None
|
||
self._cond = threading.Condition()
|
||
# 已验证可用的代理池: {proxy_url: validated_timestamp}
|
||
self._verified_pool: dict[str, float] = {}
|
||
# 正在使用中的代理（取走但未归还），避免并发账号用同一个代理
|
||
self._in_use: set[str] = set()
|
||
self._pool_ttl = 90 # 代理验证后90秒内可复用
|
||
self._fetching = False # 是否有线程正在获取代理
|
||
|
||
def _pick_from_pool_locked(self) -> Optional[str]:
|
||
"""从池中取一个未过期且未在使用的代理（调用前需持有锁）"""
|
||
now = time.time()
|
||
# 清理过期代理
|
||
expired = [p for p, t in self._verified_pool.items() if now - t > self._pool_ttl]
|
||
for p in expired:
|
||
del self._verified_pool[p]
|
||
self._in_use.discard(p)
|
||
# 取一个不在使用中的
|
||
for proxy in self._verified_pool:
|
||
if proxy not in self._in_use:
|
||
self._in_use.add(proxy)
|
||
self.current_proxy = proxy
|
||
return proxy
|
||
# 所有代理都在使用中，但池非空（并发数>代理数），允许复用第一个
|
||
for proxy in self._verified_pool:
|
||
self.current_proxy = proxy
|
||
return proxy
|
||
return None
|
||
|
||
def get_proxy(self, max_attempts: int = 5) -> Optional[str]:
|
||
"""
|
||
获取可用代理IP，优先从已验证代理池复用。
|
||
线程安全：池空时只有一个线程调API获取并验证所有代理入池，其他线程等待后复用。
|
||
|
||
Returns:
|
||
代理URL，格式: http://ip:port
|
||
"""
|
||
with self._cond:
|
||
# 1. 优先从池中取未过期的
|
||
proxy = self._pick_from_pool_locked()
|
||
if proxy:
|
||
logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})")
|
||
return proxy
|
||
# 2. 已有线程在获取，等待结果
|
||
if self._fetching:
|
||
logger.debug("代理池空，等待其他线程获取...")
|
||
self._cond.wait(timeout=60)
|
||
proxy = self._pick_from_pool_locked()
|
||
if proxy:
|
||
logger.debug(f"等待后从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})")
|
||
return proxy
|
||
return None
|
||
# 3. 自己去获取
|
||
self._fetching = True
|
||
|
||
# 释放锁后执行耗时的API调用+验证
|
||
try:
|
||
proxies_all, msg = self._fetch_and_verify_all(max_attempts)
|
||
finally:
|
||
with self._cond:
|
||
self._fetching = False
|
||
self._cond.notify_all()
|
||
|
||
if proxies_all:
|
||
with self._cond:
|
||
now = time.time()
|
||
for p in proxies_all:
|
||
self._verified_pool[p] = now
|
||
first = proxies_all[0]
|
||
self.current_proxy = first
|
||
logger.info(f"代理池充盈 {len(proxies_all)} 个可用代理: {first} ...")
|
||
return first
|
||
logger.warning(f"获取代理失败: {msg}")
|
||
return None
|
||
|
||
def _fetch_and_verify_all(self, max_attempts: int) -> tuple[Optional[list[str]], str]:
|
||
"""调代理API获取一批代理，并发验证所有可用代理，自动处理白名单同步。"""
|
||
last_error = ""
|
||
last_synced_ip: Optional[str] = None
|
||
|
||
for attempt in range(1, max_attempts + 1):
|
||
if attempt > 1:
|
||
time.sleep(min(attempt - 1, 2))
|
||
|
||
# 每次尝试前主动同步白名单（出口IP可能漂移，需重新同步）
|
||
if self.whitelist_uid and self.whitelist_ukey:
|
||
from core.douyu.whitelist import get_local_exit_ip, WhitelistManager
|
||
local_ip = get_local_exit_ip()
|
||
if local_ip and local_ip != last_synced_ip:
|
||
logger.info(f"[尝试 {attempt}] 检测出口IP: {local_ip}，同步白名单...")
|
||
manager = WhitelistManager(self.whitelist_uid, self.whitelist_ukey)
|
||
ok, sync_msg = manager.sync_ip(local_ip)
|
||
if ok:
|
||
last_synced_ip = local_ip
|
||
logger.info(f"白名单同步成功: {sync_msg}")
|
||
# 等待白名单生效
|
||
time.sleep(2)
|
||
else:
|
||
logger.warning(f"白名单同步失败: {sync_msg}")
|
||
elif local_ip == last_synced_ip:
|
||
logger.debug(f"[尝试 {attempt}] 出口IP未变: {local_ip}")
|
||
|
||
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:
|
||
logger.info(f"获取到 {len(proxy_urls)} 个代理，并发验证所有")
|
||
available, msg = verify_proxies_concurrent(proxy_urls, return_all=True)
|
||
if available:
|
||
return available, msg
|
||
last_error = msg
|
||
logger.warning(f"代理预检 {attempt}/{max_attempts}: {last_error}")
|
||
continue
|
||
|
||
# 代理API返回白名单错误
|
||
if whitelist_ip and self.whitelist_uid and self.whitelist_ukey:
|
||
logger.warning(f"代理需要白名单IP: {whitelist_ip}，自动同步...")
|
||
from core.douyu.whitelist import WhitelistManager
|
||
manager = WhitelistManager(self.whitelist_uid, self.whitelist_ukey)
|
||
ok, sync_msg = manager.sync_ip(whitelist_ip)
|
||
if ok:
|
||
last_synced_ip = whitelist_ip
|
||
logger.info("白名单已更新，等待2秒后重试...")
|
||
time.sleep(2)
|
||
continue
|
||
return None, f'白名单同步失败: {sync_msg}'
|
||
|
||
last_error = '代理API响应无法解析'
|
||
logger.warning(f"代理预检 {attempt}/{max_attempts}: {last_error}: {text[:80]}")
|
||
|
||
except Exception as exc:
|
||
last_error = f'代理API请求失败: {exc}'
|
||
logger.warning(f"代理预检 {attempt}/{max_attempts}: {last_error}")
|
||
|
||
return None, f'代理预检失败（{max_attempts}次尝试均失败）: {last_error}'
|
||
|
||
def mark_bad(self, proxy_url: str) -> None:
|
||
"""标记代理为不可用，从池中移除（极验失败/代理连接失败时调用）"""
|
||
with self._cond:
|
||
removed = self._verified_pool.pop(proxy_url, None)
|
||
self._in_use.discard(proxy_url)
|
||
if self.current_proxy == proxy_url:
|
||
self.current_proxy = None
|
||
if removed:
|
||
logger.info(f"代理标记为不可用并移出池: {proxy_url} (池剩余 {len(self._verified_pool)})")
|
||
self._cond.notify_all()
|
||
|
||
def release_proxy(self, proxy_url: str) -> None:
|
||
"""归还代理到池（登录完成后调用，让其他账号可以复用）"""
|
||
with self._cond:
|
||
self._in_use.discard(proxy_url)
|
||
self._cond.notify_all()
|
||
|
||
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 = "",
|
||
whitelist_uid: str = "",
|
||
whitelist_ukey: str = "",
|
||
) -> ProxyManager:
|
||
"""获取代理管理器实例"""
|
||
return ProxyManager(api_url, whitelist_uid=whitelist_uid, whitelist_ukey=whitelist_ukey)
|
||
|
||
|
||
def parse_proxy_response(text: str) -> tuple[list[str], Optional[str]]:
|
||
"""
|
||
解析代理API响应，支持 JSON 格式与旧版纯文本格式。
|
||
|
||
JSON 格式示例：
|
||
正常: {"code":0,"success":"true","msg":"","data":[{"IP":"1.2.3.4","Port":5791,...}]}
|
||
白名单错误: {"code":-1,"success":"true","msg":"51.请先添加白名单：39.144.109.21","data":""}
|
||
|
||
旧版纯文本格式（兼容）：
|
||
正常: 1.2.3.4:5791\\n5.6.7.8:8080
|
||
白名单错误: 请先添加白名单：39.144.109.21
|
||
|
||
Returns:
|
||
(proxy_urls, whitelist_ip)
|
||
- proxy_urls: 解析到的所有代理地址列表（http://ip:port）
|
||
- whitelist_ip: 需要添加到白名单的IP（当API返回白名单错误时），无错误时为 None
|
||
"""
|
||
text = (text or "").strip()
|
||
if not text:
|
||
return [], None
|
||
|
||
# 优先尝试 JSON 解析
|
||
try:
|
||
data = json.loads(text)
|
||
if isinstance(data, dict):
|
||
# 白名单错误：code != 0 且 msg 含白名单提示
|
||
code = data.get("code")
|
||
msg = data.get("msg", "") or ""
|
||
if code != 0 and ("白名单" in msg or "添加白名单" in msg):
|
||
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', msg)
|
||
if ip_match:
|
||
return [], ip_match.group(1)
|
||
|
||
# 正常返回：从 data 数组提取 IP/Port
|
||
data_field = data.get("data")
|
||
proxies: list[str] = []
|
||
if isinstance(data_field, list):
|
||
for item in data_field:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
ip = item.get("IP") or item.get("ip")
|
||
port = item.get("Port") or item.get("port")
|
||
if ip and port:
|
||
proxies.append(f"http://{ip}:{port}")
|
||
if proxies:
|
||
return proxies, None
|
||
|
||
# data 为空但 code==0，可能代理暂时不可用
|
||
if code == 0 and not proxies:
|
||
return [], None
|
||
except (json.JSONDecodeError, ValueError):
|
||
# 不是 JSON，回退到文本解析
|
||
pass
|
||
|
||
# 旧版文本格式：白名单错误优先检测
|
||
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 = [f"http://{ip}:{port}" for ip, port in matches]
|
||
if proxies:
|
||
return proxies, None
|
||
|
||
return [], None
|
||
|
||
|
||
def verify_proxy_url(proxy_url: str, timeout: tuple = (5, 8)) -> tuple[bool, str]:
|
||
"""
|
||
验证代理是否可用，只验证斗鱼主站（登录的最终目标）。
|
||
|
||
Returns:
|
||
(是否可用, 消息)
|
||
"""
|
||
from utils.http_logger import log_http
|
||
import time as _time
|
||
|
||
proxies = {'http': proxy_url, 'https': proxy_url}
|
||
started = _time.monotonic()
|
||
try:
|
||
response = requests.get(
|
||
'https://www.douyu.com',
|
||
proxies=proxies,
|
||
timeout=timeout,
|
||
headers={'User-Agent': 'Mozilla/5.0'},
|
||
)
|
||
elapsed = _time.monotonic() - started
|
||
response.raise_for_status()
|
||
log_http(
|
||
category="proxy_verify", method="GET", url="https://www.douyu.com",
|
||
status_code=response.status_code, response_body=f"代理可用: {proxy_url}",
|
||
duration=elapsed, proxy=proxy_url, tag="proxy_verify",
|
||
)
|
||
return True, '代理可用 → 斗鱼主站'
|
||
except Exception as e:
|
||
elapsed = _time.monotonic() - started
|
||
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"代理验证失败 [{proxy_url}]: {detail}")
|
||
log_http(
|
||
category="proxy_verify", method="GET", url="https://www.douyu.com",
|
||
duration=elapsed, error=f"[{proxy_url}] {detail}: {err_msg[:200]}",
|
||
proxy=proxy_url, tag="proxy_verify",
|
||
)
|
||
return False, detail
|
||
|
||
|
||
def verify_proxies_concurrent(proxy_urls: list[str], timeout: tuple = (5, 8), max_workers: int = 5, return_all: bool = False) -> tuple[Optional[str | list[str]], str]:
|
||
"""
|
||
并发验证多个代理URL。
|
||
|
||
Args:
|
||
proxy_urls: 代理URL列表
|
||
timeout: 验证超时
|
||
max_workers: 最大并发数
|
||
return_all: True 时返回所有可用代理列表；False（默认）返回第一个可用的
|
||
|
||
Returns:
|
||
return_all=False: (可用代理URL或None, 消息)
|
||
return_all=True: (可用代理URL列表或None, 消息)
|
||
"""
|
||
if not proxy_urls:
|
||
return None, '无代理可验证'
|
||
|
||
if len(proxy_urls) == 1:
|
||
ok, msg = verify_proxy_url(proxy_urls[0], timeout)
|
||
if ok:
|
||
return (proxy_urls if return_all else proxy_urls[0]), msg
|
||
return None, msg
|
||
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
|
||
if return_all:
|
||
# 收集所有可用代理（不取消任何任务）
|
||
available: list[str] = []
|
||
with ThreadPoolExecutor(max_workers=min(max_workers, len(proxy_urls))) as executor:
|
||
future_map = {
|
||
executor.submit(verify_proxy_url, p, timeout): p
|
||
for p in proxy_urls
|
||
}
|
||
for future in as_completed(future_map):
|
||
try:
|
||
ok, _ = future.result()
|
||
if ok:
|
||
available.append(future_map[future])
|
||
except Exception:
|
||
continue
|
||
if available:
|
||
logger.success(f"并发验证找到 {len(available)}/{len(proxy_urls)} 个可用代理")
|
||
return available, f'找到 {len(available)} 个可用代理'
|
||
return None, f'共 {len(proxy_urls)} 个代理均不可用'
|
||
|
||
# 默认：返回第一个可用的，取消其余
|
||
with ThreadPoolExecutor(max_workers=min(max_workers, len(proxy_urls))) as executor:
|
||
future_map = {
|
||
executor.submit(verify_proxy_url, p, timeout): p
|
||
for p in proxy_urls
|
||
}
|
||
for future in as_completed(future_map):
|
||
proxy_url = future_map[future]
|
||
try:
|
||
ok, msg = future.result()
|
||
if ok:
|
||
logger.success(f"并发验证找到可用代理: {proxy_url}")
|
||
for f in future_map:
|
||
if f != future:
|
||
f.cancel()
|
||
return proxy_url, msg
|
||
except Exception:
|
||
continue
|
||
|
||
return None, f'共 {len(proxy_urls)} 个代理均不可用'
|
||
|
||
|
||
def resolve_working_proxy(
|
||
api_url: str,
|
||
whitelist_uid: str = "",
|
||
whitelist_ukey: str = "",
|
||
max_attempts: int = 4,
|
||
log_func=None,
|
||
) -> tuple[Optional[str], str]:
|
||
"""
|
||
从代理API获取可用代理，自动处理白名单同步。
|
||
支持API返回多个代理IP，逐一验证直到找到可用的。
|
||
|
||
Args:
|
||
api_url: 代理API地址
|
||
whitelist_uid: 白名单UID（启用白名单时传入）
|
||
whitelist_ukey: 白名单UKEY
|
||
max_attempts: 最大获取尝试次数（默认4次）
|
||
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 - 1, 2)
|
||
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)} 个代理，并发验证')
|
||
proxy_url, msg = verify_proxies_concurrent(proxy_urls)
|
||
if proxy_url:
|
||
log('success', f'代理预检成功: {proxy_url}')
|
||
return proxy_url, msg
|
||
last_error = msg
|
||
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}'
|