yml2213
dd56de9bd4
- proxy_service.py: 将requests和WhitelistManager改为函数内延迟import， 避免启动时加载不需要的依赖；移除未使用的get_exit_ip_via_proxy导入 - account_service.py: 移除未使用的func、joinedload、AuditLog、 user_has_permission顶层导入
166 lines
6.1 KiB
Python
166 lines
6.1 KiB
Python
"""代理池管理。"""
|
|
|
|
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)
|