""" Менеджер прокси с поддержкой 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 def _check_proxy_sync( scheme: str, ip: str, port: str, login: Optional[str], password: Optional[str], timeout: float, test_url: str, ) -> bool: """Синхронная проверка прокси через requests + PySocks.""" import requests if login and password: proxy_url = f"{scheme}://{login}:{password}@{ip}:{port}" else: proxy_url = f"{scheme}://{ip}:{port}" proxies = {"http": proxy_url, "https": proxy_url} try: r = requests.get(test_url, proxies=proxies, timeout=timeout) return r.status_code in (200, 204) except Exception: return False async def detect_proxy_type( ip: str, port: str, login: Optional[str] = None, password: Optional[str] = None, timeout: float = 5.0, test_url: str = "http://www.gstatic.com/generate_204" ) -> Optional[ProxyType]: """ Определяет тип прокси подключением. Пробует SOCKS5 → SOCKS4 → HTTP. Возвращает первый рабочий или None. Учётные данные передаются в URL прокси (socks5://user:pass@ip:port). """ candidates = [ (ProxyType.SOCKS5, "socks5"), (ProxyType.SOCKS4, "socks4"), (ProxyType.HTTP, "http"), ] loop = asyncio.get_event_loop() for proxy_type, scheme in candidates: ok = await loop.run_in_executor( None, _check_proxy_sync, scheme, ip, port, login, password, timeout, test_url ) if ok: logger.info(f"Proxy {ip}:{port} detected as {proxy_type.value}") return proxy_type logger.warning(f"Proxy {ip}:{port} — type not detected") 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._needs_detection: 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 = [] _has_scheme = {"socks5://", "socks4://", "http://", "https://"} for line in lines: explicit_type = any(line.lower().startswith(s) for s in _has_scheme) proxy = Proxy.from_line(line, default_type=ProxyType.SOCKS5) if proxy: self._proxies.append(proxy) self._working_status[proxy.id] = True loaded.append(proxy) if not explicit_type: self._needs_detection.add(proxy.id) 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) async def detect_types(self, concurrency: int = 10) -> None: """Определяет типы прокси у которых тип не задан явно.""" targets = [p for p in self._proxies if p.id in self._needs_detection] if not targets: return logger.info(f"Detecting proxy types for {len(targets)} proxies...") sem = asyncio.Semaphore(concurrency) detected = {"socks5": 0, "socks4": 0, "http": 0, "failed": 0} async def _detect_one(proxy: Proxy): async with sem: old_id = proxy.id ptype = await detect_proxy_type(proxy.ip, proxy.port, proxy.login, proxy.password) if ptype and ptype != proxy.proxy_type: self._working_status.pop(old_id, None) self._needs_detection.discard(old_id) proxy.proxy_type = ptype self._working_status[proxy.id] = True detected[ptype.value] += 1 elif ptype is None: self._working_status[proxy.id] = False detected["failed"] += 1 else: detected[ptype.value] += 1 await asyncio.gather(*[_detect_one(p) for p in targets]) self._needs_detection.clear() logger.info( f"Proxy detection done — " f"socks5:{detected['socks5']} socks4:{detected['socks4']} " f"http:{detected['http']} failed:{detected['failed']}" ) 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]]