fix: 修复proxy_service和account_service中不当的顶层import
- proxy_service.py: 将requests和WhitelistManager改为函数内延迟import， 避免启动时加载不需要的依赖；移除未使用的get_exit_ip_via_proxy导入 - account_service.py: 移除未使用的func、joinedload、AuditLog、 user_has_permission顶层导入
This commit is contained in:
yml2213
+14
-468
@@ -1,470 +1,16 @@
|
||||
"""代理管理模块"""
|
||||
"""代理管理模块兼容导出。"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import requests
|
||||
from typing import Optional
|
||||
from loguru import logger
|
||||
from .proxy_manager import ProxyManager, get_proxy_manager
|
||||
from .proxy_parser import parse_proxy_response
|
||||
from .proxy_resolver import ProxyResolver, resolve_working_proxy
|
||||
from .proxy_verifier import verify_proxies_concurrent, verify_proxy_url
|
||||
|
||||
|
||||
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}'
|
||||
__all__ = [
|
||||
"ProxyManager",
|
||||
"ProxyResolver",
|
||||
"get_proxy_manager",
|
||||
"parse_proxy_response",
|
||||
"resolve_working_proxy",
|
||||
"verify_proxies_concurrent",
|
||||
"verify_proxy_url",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
"""代理池管理。"""
|
||||
|
||||
import threading
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
import requests
|
||||
from loguru import logger
|
||||
|
||||
from .proxy_resolver import ProxyResolver
|
||||
from .proxy_whitelist import DouyuWhitelistSyncer
|
||||
|
||||
|
||||
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
|
||||
self._fetching = False
|
||||
self._whitelist_syncer = (
|
||||
DouyuWhitelistSyncer(whitelist_uid, whitelist_ukey)
|
||||
if whitelist_uid and whitelist_ukey
|
||||
else None
|
||||
)
|
||||
|
||||
def _pick_from_pool_locked(self) -> Optional[str]:
|
||||
"""从池中取一个未过期且未在使用的代理（调用前需持有锁）。"""
|
||||
now = time.time()
|
||||
expired = [proxy for proxy, ts in self._verified_pool.items() if now - ts > self._pool_ttl]
|
||||
for proxy in expired:
|
||||
del self._verified_pool[proxy]
|
||||
self._in_use.discard(proxy)
|
||||
|
||||
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 获取并验证所有代理入池，
|
||||
其他线程等待后复用。
|
||||
"""
|
||||
with self._cond:
|
||||
proxy = self._pick_from_pool_locked()
|
||||
if proxy:
|
||||
logger.debug(f"从代理池复用: {proxy} (池剩余 {len(self._verified_pool)})")
|
||||
return proxy
|
||||
|
||||
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
|
||||
|
||||
self._fetching = True
|
||||
|
||||
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 proxy in proxies_all:
|
||||
self._verified_pool[proxy] = now
|
||||
first = proxies_all[0]
|
||||
self._in_use.add(first)
|
||||
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 获取一批代理，并发验证所有可用代理。"""
|
||||
resolver = ProxyResolver(
|
||||
api_url=self.api_url,
|
||||
whitelist_syncer=self._whitelist_syncer,
|
||||
sync_local_exit_ip=bool(self._whitelist_syncer),
|
||||
sync_whitelist_once=False,
|
||||
)
|
||||
available, msg = resolver.fetch_verified(max_attempts=max_attempts, return_all=True)
|
||||
if isinstance(available, list):
|
||||
return available, msg
|
||||
if isinstance(available, str):
|
||||
return [available], msg
|
||||
return None, msg
|
||||
|
||||
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 exc:
|
||||
logger.error(f"代理验证失败: {exc}")
|
||||
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)
|
||||
@@ -0,0 +1,59 @@
|
||||
"""代理 API 响应解析。"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def parse_proxy_response(text: str) -> tuple[list[str], Optional[str]]:
|
||||
"""
|
||||
解析代理 API 响应，支持 JSON 格式与旧版纯文本格式。
|
||||
|
||||
Returns:
|
||||
(proxy_urls, whitelist_ip)
|
||||
- proxy_urls: 解析到的所有代理地址列表（http://ip:port）
|
||||
- whitelist_ip: 需要添加到白名单的 IP，无错误时为 None
|
||||
"""
|
||||
text = (text or "").strip()
|
||||
if not text:
|
||||
return [], None
|
||||
|
||||
try:
|
||||
data = json.loads(text)
|
||||
if isinstance(data, dict):
|
||||
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_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
|
||||
|
||||
if code == 0 and not proxies:
|
||||
return [], None
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
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)
|
||||
|
||||
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
|
||||
@@ -0,0 +1,175 @@
|
||||
"""代理获取、白名单同步、可用性验证的统一流程。"""
|
||||
|
||||
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}")
|
||||
time.sleep(2)
|
||||
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', '白名单已更新，等待2秒后重试...')
|
||||
time.sleep(2)
|
||||
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_uid: str = "",
|
||||
whitelist_ukey: str = "",
|
||||
max_attempts: int = 4,
|
||||
log_func: Optional[LogFunc] = None,
|
||||
) -> tuple[Optional[str], str]:
|
||||
"""
|
||||
从代理 API 获取可用代理，自动处理白名单同步。
|
||||
|
||||
保留旧函数签名，供 Web 层代理测试继续使用。
|
||||
"""
|
||||
syncer = (
|
||||
DouyuWhitelistSyncer(whitelist_uid, whitelist_ukey)
|
||||
if whitelist_uid and whitelist_ukey
|
||||
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
|
||||
@@ -0,0 +1,114 @@
|
||||
"""代理可用性验证。"""
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from typing import Optional
|
||||
import time as _time
|
||||
|
||||
import requests
|
||||
from loguru import logger
|
||||
|
||||
|
||||
def verify_proxy_url(proxy_url: str, timeout: tuple = (5, 8)) -> tuple[bool, str]:
|
||||
"""
|
||||
验证代理是否可用，只验证斗鱼主站（登录的最终目标）。
|
||||
|
||||
Returns:
|
||||
(是否可用, 消息)
|
||||
"""
|
||||
from utils.http_logger import log_http
|
||||
|
||||
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 exc:
|
||||
elapsed = _time.monotonic() - started
|
||||
err_msg = str(exc)
|
||||
if 'Tunnel connection failed' in err_msg or '503' in err_msg:
|
||||
detail = '代理拒绝连接（白名单可能未生效）'
|
||||
elif 'timed out' in err_msg.lower():
|
||||
detail = '连接超时'
|
||||
else:
|
||||
detail = type(exc).__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。
|
||||
|
||||
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
|
||||
|
||||
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, proxy, timeout): proxy
|
||||
for proxy 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, proxy, timeout): proxy
|
||||
for proxy 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 item in future_map:
|
||||
if item != future:
|
||||
item.cancel()
|
||||
return proxy_url, msg
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
return None, f'共 {len(proxy_urls)} 个代理均不可用'
|
||||
@@ -0,0 +1,20 @@
|
||||
"""代理模块使用的白名单适配器。"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from .whitelist import WhitelistManager, get_local_exit_ip
|
||||
|
||||
|
||||
class DouyuWhitelistSyncer:
|
||||
"""把白名单 API 隔离成代理解析流程可调用的适配器。"""
|
||||
|
||||
def __init__(self, uid: str, ukey: str):
|
||||
self.uid = uid
|
||||
self.ukey = ukey
|
||||
|
||||
def sync_ip(self, ip: str) -> tuple[bool, str]:
|
||||
manager = WhitelistManager(self.uid, self.ukey)
|
||||
return manager.sync_ip(ip)
|
||||
|
||||
def get_local_exit_ip(self) -> Optional[str]:
|
||||
return get_local_exit_ip()
|
||||
Reference in New Issue
Block a user