yml2213
a7da949d9c
1. ROUNDCUBE_URL: os.getenv返回空字符串时不回退默认值，改为 or 默认值 2. ProxyFetcher: 缓存已同步白名单IP，后续线程跳过重复同步
87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""简化版代理获取器（替代 ProxyManager）。
|
|
|
|
专为短效代理设计：无池、无冷却、无复用。
|
|
每次调用 fetch_new_proxy() 从 API 取 1 个新代理，用完即弃。
|
|
|
|
线程安全：白名单同步有全局锁 + IP缓存避免重复同步。
|
|
"""
|
|
|
|
import threading
|
|
from typing import Optional
|
|
|
|
from loguru import logger
|
|
|
|
from .proxy_resolver import ProxyResolver
|
|
from .proxy_whitelist import DouyuWhitelistSyncer
|
|
|
|
|
|
class ProxyFetcher:
|
|
"""每次从代理 API 取 1 个新代理，无缓存无复用。"""
|
|
|
|
def __init__(
|
|
self,
|
|
api_url: str,
|
|
whitelist_platform: str = "xiequ",
|
|
whitelist_credentials: dict = None,
|
|
):
|
|
self.api_url = api_url
|
|
|
|
_wl_platform = whitelist_platform or "xiequ"
|
|
_wl_credentials = whitelist_credentials
|
|
|
|
self._whitelist_syncer = (
|
|
DouyuWhitelistSyncer(platform=_wl_platform, credentials=_wl_credentials)
|
|
if _wl_credentials
|
|
else None
|
|
)
|
|
# 跨线程共享：已同步白名单的IP，避免5个线程重复同步同一个IP
|
|
self._synced_ip: Optional[str] = None
|
|
self._sync_lock = threading.Lock()
|
|
|
|
def fetch_new_proxy(self, max_attempts: int = 3) -> Optional[str]:
|
|
"""从代理 API 获取 1 个可用代理，失败返回 None。"""
|
|
# 快速检查：是否已经同步过白名单
|
|
skip_sync = False
|
|
if self._whitelist_syncer:
|
|
with self._sync_lock:
|
|
if self._synced_ip:
|
|
skip_sync = True
|
|
|
|
resolver = ProxyResolver(
|
|
api_url=self.api_url,
|
|
whitelist_syncer=self._whitelist_syncer,
|
|
sync_local_exit_ip=bool(self._whitelist_syncer) and not skip_sync,
|
|
sync_whitelist_once=False,
|
|
)
|
|
result, msg = resolver.fetch_verified(
|
|
max_attempts=max_attempts,
|
|
return_all=False,
|
|
)
|
|
|
|
# 记录已同步的IP
|
|
if self._whitelist_syncer and result:
|
|
with self._sync_lock:
|
|
self._synced_ip = self._synced_ip or "done"
|
|
|
|
if isinstance(result, str):
|
|
logger.info(f"获取新代理: {result}")
|
|
return result
|
|
if isinstance(result, list) and result:
|
|
logger.info(f"获取新代理: {result[0]}")
|
|
return result[0]
|
|
logger.warning(f"获取代理失败: {msg}")
|
|
return None
|
|
|
|
|
|
def get_proxy_fetcher(
|
|
api_url: str,
|
|
whitelist_platform: str = "xiequ",
|
|
whitelist_credentials: dict = None,
|
|
) -> ProxyFetcher:
|
|
"""获取代理获取器实例。"""
|
|
return ProxyFetcher(
|
|
api_url,
|
|
whitelist_platform=whitelist_platform,
|
|
whitelist_credentials=whitelist_credentials,
|
|
)
|