feat: 实现代理健康检查功能

添加完整的代理检测系统,包括:

后端功能:
- 新增代理检测服务模块 (app/service/proxy/)
  * ProxyCheckService: 核心检测逻辑,支持HTTP/HTTPS/SOCKS5代理
  * 内置缓存机制,避免重复检测 (10秒缓存期)
  * 支持并发检测,可配置最大并发数
- 新增4个API端点 (app/router/config_routes.py):
  * POST /api/config/proxy/check - 单个代理检测
  * POST /api/config/proxy/check-all - 批量代理检测
  * GET /api/config/proxy/cache-stats - 缓存统计
  * POST /api/config/proxy/clear-cache - 清空缓存

前端功能:
- 代理列表中每项添加状态图标和检测按钮
- 支持单个代理实时检测
- 新增检测所有代理功能按钮
- 代理检测结果模态框,显示:
  * 检测进度条和状态
  * 可用/不可用代理统计
  * 详细的检测结果列表(响应时间、错误信息)
  * 重试失败代理功能

技术特性:
- 10秒超时机制,避免长时间等待
- 异步并发检测,提升批量检测效率
- 缓存机制减少重复请求
- 完整的错误处理和用户友好提示
- 响应式UI设计,支持实时状态更新
This commit is contained in:
icesixgod
2025-07-24 19:41:03 +08:00
parent ccd4722a77
commit 9f5892a987
5 changed files with 713 additions and 0 deletions
+7
View File
@@ -0,0 +1,7 @@
"""
Proxy service module
"""
from .proxy_check_service import ProxyCheckService
__all__ = ["ProxyCheckService"]
+219
View File
@@ -0,0 +1,219 @@
"""
Proxy detection service module
"""
import asyncio
import time
from typing import Dict, List, Optional, Tuple
from urllib.parse import urlparse
import httpx
from pydantic import BaseModel
from app.log.logger import get_config_routes_logger
logger = get_config_routes_logger()
class ProxyCheckResult(BaseModel):
"""Proxy check result model"""
proxy: str
is_available: bool
response_time: Optional[float] = None
error_message: Optional[str] = None
checked_at: float
class ProxyCheckService:
"""Proxy detection service class"""
# Target URL for checking
CHECK_URL = "https://www.google.com"
# Timeout in seconds
TIMEOUT_SECONDS = 10
# Cache duration in seconds
CACHE_DURATION = 10 # 10s
def __init__(self):
self._cache: Dict[str, ProxyCheckResult] = {}
def _is_valid_proxy_format(self, proxy: str) -> bool:
"""Validate proxy format"""
try:
parsed = urlparse(proxy)
return parsed.scheme in ['http', 'https', 'socks5'] and parsed.hostname
except Exception:
return False
def _get_cached_result(self, proxy: str) -> Optional[ProxyCheckResult]:
"""Get cached check result"""
if proxy in self._cache:
result = self._cache[proxy]
# Check if cache is expired
if time.time() - result.checked_at < self.CACHE_DURATION:
logger.debug(f"Using cached proxy check result: {proxy}")
return result
else:
# Remove expired cache
del self._cache[proxy]
return None
def _cache_result(self, result: ProxyCheckResult) -> None:
"""Cache check result"""
self._cache[result.proxy] = result
async def check_single_proxy(self, proxy: str, use_cache: bool = True) -> ProxyCheckResult:
"""
Check if a single proxy is available
Args:
proxy: Proxy address in format like http://host:port or socks5://host:port
use_cache: Whether to use cached results
Returns:
ProxyCheckResult: Check result
"""
# Check cache first
if use_cache:
cached = self._get_cached_result(proxy)
if cached:
return cached
# Validate proxy format
if not self._is_valid_proxy_format(proxy):
result = ProxyCheckResult(
proxy=proxy,
is_available=False,
error_message="Invalid proxy format",
checked_at=time.time()
)
self._cache_result(result)
return result
# Perform check
start_time = time.time()
try:
logger.info(f"Starting proxy check: {proxy}")
timeout = httpx.Timeout(self.TIMEOUT_SECONDS, read=self.TIMEOUT_SECONDS)
async with httpx.AsyncClient(timeout=timeout, proxy=proxy) as client:
response = await client.head(self.CHECK_URL)
response_time = time.time() - start_time
# Check response status
is_available = response.status_code in [200, 204, 301, 302, 307, 308]
result = ProxyCheckResult(
proxy=proxy,
is_available=is_available,
response_time=round(response_time, 3),
error_message=None if is_available else f"HTTP {response.status_code}",
checked_at=time.time()
)
logger.info(f"Proxy check completed: {proxy}, available: {is_available}, response_time: {response_time:.3f}s")
except asyncio.TimeoutError:
result = ProxyCheckResult(
proxy=proxy,
is_available=False,
error_message="Connection timeout",
checked_at=time.time()
)
logger.warning(f"Proxy check timeout: {proxy}")
except Exception as e:
result = ProxyCheckResult(
proxy=proxy,
is_available=False,
error_message=str(e),
checked_at=time.time()
)
logger.error(f"Proxy check failed: {proxy}, error: {str(e)}")
# Cache result
self._cache_result(result)
return result
async def check_multiple_proxies(
self,
proxies: List[str],
use_cache: bool = True,
max_concurrent: int = 5
) -> List[ProxyCheckResult]:
"""
Check multiple proxies concurrently
Args:
proxies: List of proxy addresses
use_cache: Whether to use cached results
max_concurrent: Maximum concurrent check count
Returns:
List[ProxyCheckResult]: List of check results
"""
if not proxies:
return []
logger.info(f"Starting batch proxy check for {len(proxies)} proxies")
# Use semaphore to limit concurrency
semaphore = asyncio.Semaphore(max_concurrent)
async def check_with_semaphore(proxy: str) -> ProxyCheckResult:
async with semaphore:
return await self.check_single_proxy(proxy, use_cache)
# Execute checks concurrently
tasks = [check_with_semaphore(proxy) for proxy in proxies]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Handle exception results
final_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"Proxy check task exception: {proxies[i]}, error: {str(result)}")
final_results.append(ProxyCheckResult(
proxy=proxies[i],
is_available=False,
error_message=f"Check task exception: {str(result)}",
checked_at=time.time()
))
else:
final_results.append(result)
available_count = sum(1 for r in final_results if r.is_available)
logger.info(f"Batch proxy check completed: {available_count}/{len(proxies)} proxies available")
return final_results
def get_cache_stats(self) -> Dict[str, int]:
"""Get cache statistics"""
current_time = time.time()
valid_cache_count = sum(
1 for result in self._cache.values()
if current_time - result.checked_at < self.CACHE_DURATION
)
return {
"total_cached": len(self._cache),
"valid_cached": valid_cache_count,
"expired_cached": len(self._cache) - valid_cache_count
}
def clear_cache(self) -> None:
"""Clear all cache"""
self._cache.clear()
logger.info("Proxy check cache cleared")
# Global instance
_proxy_check_service: Optional[ProxyCheckService] = None
def get_proxy_check_service() -> ProxyCheckService:
"""Get proxy check service instance"""
global _proxy_check_service
if _proxy_check_service is None:
_proxy_check_service = ProxyCheckService()
return _proxy_check_service