backend/app/core/proxy_manager.py

import httpx
import asyncio
import random
import logging
import psutil
import os
from typing import List, Optional
from app.core.config import settings
from tenacity import (
    retry, stop_after_attempt, wait_fixed, retry_if_exception_type
)

import httpx
import asyncio
import random
import logging
import psutil
import os
from typing import List, Optional
from app.core.config import settings
from tenacity import (
    retry, stop_after_attempt, wait_fixed, retry_if_exception_type
)


class ProxyManager:
    """
    代理池管理类，负责代理的获取、验证、内存监控和动态调整。
    """
    _instance: Optional["ProxyManager"] = None
    _initialized: bool = False
    logger = logging.getLogger(__name__)  # 将 logger 作为类属性初始化

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(ProxyManager, cls).__new__(cls)
        return cls._instance

    def __init__(self) -> None:
        """
        初始化 ProxyManager 实例。
        """
        if not self._initialized:
            self.proxies: List[str] = []  # 存储可用代理
            self.max_size: int = settings.PROXY_POOL_MAX_SIZE  # 代理池最大容量
            self.monitor_interval: int = settings.PROXY_POOL_MONITOR_INTERVAL  # 监控间隔
            self.is_monitoring: bool = False  # 是否正在监控
            self._initialized = True
            self.logger.info(f"代理管理器初始化完成，最大容量: {self.max_size}")

    async def _fetch_proxies(self) -> List[str]:
        """
        从配置中获取代理列表。
        """
        logger.info("正在从配置中获取代理...")
        if settings.PROXY_POOL:
            return settings.PROXY_POOL
        return []

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_fixed(2),
        retry=retry_if_exception_type(Exception)
    )
    async def _verify_proxy(self, proxy: str) -> bool:
        """
        验证代理是否可用。
        """
        try:
            async with httpx.AsyncClient(
                proxies={"http://": proxy, "https://": proxy},
                timeout=5
            ) as client:
                response = await client.get("http://www.baidu.com")  # 使用百度作为测试目标
                response.raise_for_status()
                logger.debug(f"代理 {proxy} 验证成功。")
                return True
        except httpx.RequestError as e:
            logger.warning(f"代理 {proxy} 验证失败: {e}")
            # 失败的代理不返回，相当于临时禁用
            return False
        except Exception as e:
            import traceback # 导入 traceback 模块
            logger.error(f"验证代理 {proxy} 时发生未知错误: {e}\n{traceback.format_exc()}") # 记录详细堆栈信息
            return False

    async def _refresh_proxies(self) -> None:
        """
        刷新代理池，获取新代理并验证。
        """
        new_proxies: List[str] = await self._fetch_proxies()
        verified_proxies: List[str] = []
        for proxy in new_proxies:
            if await self._verify_proxy(proxy):
                verified_proxies.append(proxy)
        self.proxies = verified_proxies[:self.max_size]  # 精简容量
        logger.info(f"代理池已刷新，当前可用代理数量: {len(self.proxies)}")

    async def get_proxy(self) -> Optional[str]:
        """
        从代理池中获取一个随机代理。
        """
        if not self.proxies:
            logger.warning("代理池为空，正在尝试刷新...")
            await self._refresh_proxies()
            if not self.proxies:
                logger.error("代理池刷新后仍为空，无法提供代理。")
                return None
        return random.choice(self.proxies)

    def _get_memory_usage(self) -> float:
        """
        获取当前进程的内存使用率。
        """
        process = psutil.Process(os.getpid())
        memory_info = process.memory_info()
        # rss (Resident Set Size) 是进程实际使用的物理内存
        return memory_info.rss / (1024 * 1024)  # 返回MB

    async def _monitor_memory_and_adjust_pool(self) -> None:
        """
        监控内存使用，并根据情况动态调整代理池。
        """
        import traceback # 导入 traceback 模块
        while self.is_monitoring:
            try:
                current_memory_mb: float = self._get_memory_usage()
                logger.info(f"当前内存使用: {current_memory_mb:.2f} MB")

                # 内存压力检测和动态调整策略
                if current_memory_mb > settings.PROXY_MEMORY_THRESHOLD_MB:
                    # 内存压力大，紧急精简代理池并缩短监控间隔
                    if len(self.proxies) > self.max_size * 0.5:  # 紧急精简到一半
                        self.proxies = self.proxies[:int(self.max_size * 0.5)]
                        logger.warning(
                            "内存压力过大 "
                            f"({current_memory_mb:.2f}MB > "
                            f"{settings.PROXY_MEMORY_THRESHOLD_MB}MB)，"
                            f"已紧急精简代理池至 {len(self.proxies)} 个。"
                        )
                    self.monitor_interval = max(
                        10, settings.PROXY_POOL_MONITOR_INTERVAL // 2
                    )  # 缩短监控间隔，至少10秒
                    logger.warning(
                        "内存压力大，代理池监控间隔调整为 "
                        f"{self.monitor_interval} 秒。"
                    )
                elif len(self.proxies) > self.max_size:  # 如果代理数量超过最大容量
                    self.proxies = self.proxies[:self.max_size]
                    logger.warning(f"代理数量过多，已精简代理池至 {len(self.proxies)} 个。")
                    self.monitor_interval = settings.PROXY_POOL_MONITOR_INTERVAL  # 恢复正常监控间隔
                elif len(self.proxies) < self.max_size * 0.8:  # 如果代理数量低于最大容量的80%
                    logger.info("代理数量偏少，尝试刷新代理池以补充。")
                    await self._refresh_proxies()
                    self.monitor_interval = settings.PROXY_POOL_MONITOR_INTERVAL  # 恢复正常监控间隔
                else:
                    self.monitor_interval = (
                        settings.PROXY_POOL_MONITOR_INTERVAL
                    )  # 正常情况，保持默认监控间隔
            except Exception as e:
                logger.error(f"监控内存或调整代理池时发生异常: {e}\n{traceback.format_exc()}") # 记录详细堆栈信息
            finally:
                await asyncio.sleep(self.monitor_interval) # 确保即使发生异常也能继续等待

    async def start_monitoring(self) -> None:
        """
        启动内存监控和代理池动态调整任务。
        """
        if not self.is_monitoring:
            self.is_monitoring = True
            logger.info("启动代理池内存监控...")
            asyncio.create_task(self._monitor_memory_and_adjust_pool())
            # 首次启动时刷新代理池
            await self._refresh_proxies()

    def stop_monitoring(self) -> None:
        """
        停止内存监控任务。
        """
        self.is_monitoring = False
        logger.info("停止代理池内存监控。")


# 创建全局代理管理器实例
proxy_manager = ProxyManager()

# 导入 httpx 以便在 _verify_proxy 中使用