""" Менеджер прокси с поддержкой SOCKS5 и HTTP. """ import asyncio import os import random from collections import defaultdict from dataclasses import dataclass, field from typing import Optional, List, Dict, Set from enum import Enum import logging from config.settings import settings from core.exceptions import ProxyTimeoutError, NoProxiesAvailableError logger = logging.getLogger(__name__) class ProxyType(str, Enum): """Типы прокси.""" SOCKS5 = "socks5" SOCKS4 = "socks4" HTTP = "http" HTTPS = "https" @dataclass class Proxy: """Прокси с поддержкой разных типов.""" ip: str port: str proxy_type: ProxyType = ProxyType.SOCKS5 login: Optional[str] = None password: Optional[str] = None @property def id(self) -> str: return f"{self.proxy_type.value}://{self.ip}:{self.port}" @property def server(self) -> str: """URL сервера для Camoufox/Playwright.""" protocol = "socks5" if self.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4] else "http" return f"{protocol}://{self.ip}:{self.port}" @property def proxy_config(self) -> dict: """Конфигурация для браузера.""" config = { 'server': self.server, } if self.login and self.password: config.update({ 'username': self.login, 'password': self.password }) return config @classmethod def from_line(cls, line: str, default_type: ProxyType = ProxyType.SOCKS5) -> Optional['Proxy']: """ Парсит строку прокси. Форматы: - ip:port (SOCKS5 по умолчанию) - ip:port:login:password - socks5://ip:port:login:password - http://ip:port:login:password """ line = line.strip() # Определяем тип прокси по префиксу proxy_type = default_type if "://" in line: protocol, rest = line.split("://", 1) protocol = protocol.lower() if protocol in ["socks5", "socks"]: proxy_type = ProxyType.SOCKS5 elif protocol == "socks4": proxy_type = ProxyType.SOCKS4 elif protocol == "http": proxy_type = ProxyType.HTTP elif protocol == "https": proxy_type = ProxyType.HTTPS line = rest # Парсим части parts = line.split(":") if len(parts) == 2: return cls(ip=parts[0], port=parts[1], proxy_type=proxy_type) elif len(parts) == 4: return cls( ip=parts[0], port=parts[1], login=parts[2], password=parts[3], proxy_type=proxy_type ) return None class ProxyManager: """ Менеджер прокси с поддержкой SOCKS5 и HTTP. """ def __init__(self, proxy_file: Optional[str] = None): self.proxy_file = proxy_file or settings.PROXY_FILE self.cooldown_time = settings.PROXY_COOLDOWN_TIME self.max_usage = settings.PROXY_MAX_USAGE_BEFORE_COOLDOWN self._proxies: List[Proxy] = [] self._lock = asyncio.Lock() self._in_use: Dict[str, asyncio.Event] = {} self._usage_count: Dict[str, int] = defaultdict(int) self._cooldown_until: Dict[str, float] = {} self._working_status: Dict[str, bool] = {} self._used_ids: Set[str] = set() self._load_proxies() def _load_proxies(self) -> None: """Загружает прокси из файла.""" if not os.path.exists(self.proxy_file): logger.warning(f"Proxy file not found: {self.proxy_file}") return try: with open(self.proxy_file, 'r', encoding='utf-8') as f: lines = [ line.strip() for line in f if line.strip() and not line.startswith('#') ] loaded = [] for line in lines: proxy = Proxy.from_line(line, default_type=ProxyType.SOCKS5) if proxy: self._proxies.append(proxy) self._working_status[proxy.id] = True loaded.append(proxy) logger.info(f"Loaded {len(loaded)} proxies from {self.proxy_file}") # Статистика по типам type_counts = defaultdict(int) for p in loaded: type_counts[p.proxy_type.value] += 1 for ptype, count in type_counts.items(): logger.info(f" {ptype}: {count}") except Exception as e: logger.error(f"Failed to load proxies: {e}", exc_info=True) def _is_available(self, proxy: Proxy) -> bool: """Проверка доступности прокси.""" proxy_id = proxy.id if proxy_id in self._in_use: return False if proxy_id in self._used_ids: return False if proxy_id in self._cooldown_until: loop = asyncio.get_running_loop() if loop.time() < self._cooldown_until[proxy_id]: return False del self._cooldown_until[proxy_id] if self._usage_count[proxy_id] >= self.max_usage: return False return self._working_status.get(proxy_id, False) async def acquire_proxy(self, timeout: float = 60) -> Optional[Proxy]: """Получает доступный прокси.""" if not self._proxies: return None loop = asyncio.get_running_loop() start_time = loop.time() attempt = 0 while True: attempt += 1 async with self._lock: available = [ p for p in self._proxies if self._is_available(p) ] if available: proxy = random.choice(available) self._in_use[proxy.id] = asyncio.Event() self._usage_count[proxy.id] += 1 self._used_ids.add(proxy.id) logger.info( f"Proxy acquired: {proxy.id} " f"(type: {proxy.proxy_type.value}, " f"used: {self._usage_count[proxy.id]}x)" ) return proxy all_used = len(self._used_ids) >= len(self._proxies) if all_used: logger.info("All proxies used once, resetting used list") self._used_ids.clear() logger.debug(f"No proxies available (attempt {attempt})") elapsed = loop.time() - start_time if elapsed >= timeout: logger.warning(f"Proxy acquire timeout ({timeout}s)") return None await asyncio.sleep(1) async def release_proxy(self, proxy: Proxy, success: bool = True) -> None: """Освобождает прокси.""" if not proxy: return proxy_id = proxy.id async with self._lock: if proxy_id in self._in_use: del self._in_use[proxy_id] if self._usage_count[proxy_id] >= self.max_usage: loop = asyncio.get_running_loop() self._cooldown_until[proxy_id] = ( loop.time() + self.cooldown_time ) self._usage_count[proxy_id] = 0 if not success: self._working_status[proxy_id] = False logger.debug(f"Proxy released: {proxy_id}") def get_stats(self) -> dict: """Статистика прокси.""" loop = asyncio.get_running_loop() now = loop.time() return { "total": len(self._proxies), "working": sum(1 for s in self._working_status.values() if s), "in_use": len(self._in_use), "available": self.available_count, "in_cooldown": sum(1 for t in self._cooldown_until.values() if now < t), } @property def count(self) -> int: return len(self._proxies) @property def available_count(self) -> int: loop = asyncio.get_running_loop() now = loop.time() working = sum(1 for s in self._working_status.values() if s) in_use = len(self._in_use) in_cooldown = sum( 1 for t in self._cooldown_until.values() if now < t ) return working - in_use - in_cooldown def has_proxies(self) -> bool: return len(self._proxies) > 0 def get_socks5_proxies(self) -> List[Proxy]: """Список SOCKS5 прокси.""" return [p for p in self._proxies if p.proxy_type == ProxyType.SOCKS5] def get_http_proxies(self) -> List[Proxy]: """Список HTTP прокси.""" return [p for p in self._proxies if p.proxy_type in [ProxyType.HTTP, ProxyType.HTTPS]]