Yuriy Yuriev
471 lines
17 KiB
Python
471 lines
17 KiB
Python
"""
|
|
Менеджер прокси с поддержкой 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
|
|
|
|
# Поддержка формата login:pass@ip:port
|
|
login = password = None
|
|
if "@" in line:
|
|
creds, line = line.rsplit("@", 1)
|
|
if ":" in creds:
|
|
login, password = creds.split(":", 1)
|
|
|
|
# Парсим ip:port или ip:port:login:pass
|
|
parts = line.split(":")
|
|
|
|
if len(parts) == 2:
|
|
return cls(ip=parts[0], port=parts[1], proxy_type=proxy_type,
|
|
login=login, password=password)
|
|
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 = "https://www.google.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:
|
|
"""Загружает прокси из файла.
|
|
|
|
Форматы строк:
|
|
socks5://ip:port:login:pass — рабочий, тип известен
|
|
!socks5://ip:port:login:pass — нерабочий, пропускается
|
|
ip:port:login:pass — тип неизвестен, будет задетектирован
|
|
"""
|
|
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:
|
|
raw_lines = [l.rstrip('\n') for l in f]
|
|
|
|
loaded = []
|
|
_has_scheme = {"socks5://", "socks4://", "http://", "https://"}
|
|
for line in raw_lines:
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith('#'):
|
|
continue
|
|
|
|
# Нерабочий прокси — пропускаем, не загружаем
|
|
if stripped.startswith('!'):
|
|
logger.debug(f"Skipping disabled proxy: {stripped[1:30]}...")
|
|
continue
|
|
|
|
explicit_type = any(stripped.lower().startswith(s) for s in _has_scheme)
|
|
proxy = Proxy.from_line(stripped, 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']}"
|
|
)
|
|
loop = asyncio.get_running_loop()
|
|
await loop.run_in_executor(None, self._save_proxies_file)
|
|
|
|
def _save_proxies_file(self) -> None:
|
|
"""Перезаписывает proxies.txt с актуальными типами и пометками."""
|
|
if not os.path.exists(self.proxy_file):
|
|
return
|
|
try:
|
|
with open(self.proxy_file, 'r', encoding='utf-8') as f:
|
|
raw_lines = [l.rstrip('\n') for l in f]
|
|
|
|
new_lines = []
|
|
|
|
for line in raw_lines:
|
|
stripped = line.strip()
|
|
# Пустые строки и комментарии — без изменений
|
|
if not stripped or stripped.startswith('#'):
|
|
new_lines.append(line)
|
|
continue
|
|
|
|
# Нерабочие — оставляем как есть (уже с !)
|
|
if stripped.startswith('!'):
|
|
new_lines.append(line)
|
|
continue
|
|
|
|
proxy = Proxy.from_line(stripped, default_type=ProxyType.SOCKS5)
|
|
if not proxy:
|
|
new_lines.append(line)
|
|
continue
|
|
|
|
working = self._working_status.get(proxy.id, True)
|
|
scheme = proxy.proxy_type.value
|
|
# Формат совместимый с from_line: scheme://ip:port:login:pass
|
|
if proxy.login and proxy.password:
|
|
new_line = f"{scheme}://{proxy.ip}:{proxy.port}:{proxy.login}:{proxy.password}"
|
|
else:
|
|
new_line = f"{scheme}://{proxy.ip}:{proxy.port}"
|
|
|
|
if not working:
|
|
new_line = f"!{new_line}"
|
|
|
|
new_lines.append(new_line)
|
|
|
|
with open(self.proxy_file, 'w', encoding='utf-8') as f:
|
|
f.write('\n'.join(new_lines) + '\n')
|
|
|
|
logger.info(f"proxies.txt updated with detected types and status")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save proxies file: {e}")
|
|
|
|
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
|
|
mark_failed = False
|
|
|
|
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
|
|
# Удаляем из пула полностью
|
|
self._proxies = [p for p in self._proxies if p.id != proxy_id]
|
|
self._needs_detection.discard(proxy_id)
|
|
mark_failed = True
|
|
logger.warning(f"Proxy {proxy_id} marked as failed and removed from pool")
|
|
|
|
logger.debug(f"Proxy released: {proxy_id}")
|
|
|
|
if mark_failed:
|
|
loop = asyncio.get_running_loop()
|
|
await loop.run_in_executor(None, self._save_proxies_file)
|
|
|
|
def reload(self) -> int:
|
|
"""Перезагружает прокси из файла. Возвращает количество загруженных."""
|
|
self._proxies.clear()
|
|
self._working_status.clear()
|
|
self._in_use.clear()
|
|
self._usage_count.clear()
|
|
self._cooldown_until.clear()
|
|
self._used_ids.clear()
|
|
self._needs_detection.clear()
|
|
self._load_proxies()
|
|
logger.info(f"Proxies reloaded: {len(self._proxies)}")
|
|
return len(self._proxies)
|
|
|
|
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]] |