add stiky

2026-05-29 21:47:40 +07:00
parent 61ff90e51b
commit b02b7518cf
6 changed files with 109 additions and 220 deletions
@@ -1,5 +1,5 @@
 """
-Browser service с поддержкой SOCKS5 через локальный HTTP туннель.
+Browser service с поддержкой HTTP и sticky-прокси.
 """

 import asyncio
@@ -12,11 +12,9 @@ from dataclasses import dataclass

 from camoufox import DefaultAddons
 from camoufox.async_api import AsyncCamoufox
-from camoufox.exceptions import InvalidIP
+from camoufox.exceptions import InvalidIP, InvalidProxy

 from config.settings import settings
-from managers.proxy_manager import ProxyManager, Proxy, ProxyType
-from services.socks5_to_http_proxy import Socks5ProxyPool

 logger = logging.getLogger(__name__)

@@ -73,6 +71,44 @@ def _patch_camoufox():
 _patch_camoufox()


+def _make_sticky_proxy_config() -> tuple:
+    """Возвращает (proxy_config, proxy_url_with_auth, proxy_info) для sticky-сессии."""
+    import re
+    session_id = random.randint(100000, 999999)
+    # Заменяем session-XXXXXX в username, если уже есть — иначе добавляем
+    base = settings.STICKY_PROXY_USER
+    if re.search(r'session-\d+', base):
+        username = re.sub(r'session-\d+', f'session-{session_id}', base)
+    else:
+        username = f"{base}-session-{session_id}"
+    config = {
+        "server": f"http://{settings.STICKY_PROXY_HOST}:{settings.STICKY_PROXY_PORT}",
+        "username": username,
+        "password": settings.STICKY_PROXY_PASS,
+    }
+    proxy_url = (
+        f"http://{username}:{settings.STICKY_PROXY_PASS}"
+        f"@{settings.STICKY_PROXY_HOST}:{settings.STICKY_PROXY_PORT}"
+    )
+    return config, proxy_url, f"sticky-{session_id}@{settings.STICKY_PROXY_HOST}"
+
+
+async def _resolve_sticky_ip(proxy_url: str) -> Optional[str]:
+    """Определяет реальный IP sticky-сессии через сам прокси."""
+    import aiohttp
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                "https://api.ipify.org",
+                proxy=proxy_url,
+                timeout=aiohttp.ClientTimeout(total=10),
+            ) as resp:
+                return (await resp.text()).strip()
+    except Exception as e:
+        logger.warning(f"Sticky proxy IP resolve failed: {e}")
+        return None
+
+
 def _locale_for_ip(ip: str) -> str:
    """Возвращает BCP47 locale без script-тега по реальному IP прокси."""
    try:
@@ -104,14 +140,15 @@ class VisitResult:
 class BrowserService:
    """
    Сервис для посещения страниц через Camoufox.
-    Поддерживает HTTP, SOCKS5 (через туннель) и прямое соединение.
+    Поддерживает HTTP и sticky-прокси.
    """

-    def __init__(self, proxy_manager: ProxyManager):
-        self.proxy_manager = proxy_manager
-        self.socks5_pool = Socks5ProxyPool(idle_timeout=300)
+    _PROXY_FAIL_THRESHOLD = 5  # пауза задачи после N подряд неудачных визитов
+
+    def __init__(self):
        self.screenshots_dir = Path(settings.SCREENSHOTS_DIR)
        self.screenshots_dir.mkdir(exist_ok=True)
+        self._proxy_fail_streak: int = 0

    async def visit_page(
        self,
@@ -125,101 +162,61 @@ class BrowserService:
                settings.DEFAULT_MAX_READING
            )

+        if not settings.STICKY_PROXY_HOST:
+            return VisitResult(url=url, success=False, error="No proxy configured")
+
        for attempt in range(3):
-            proxy = None
-            use_socks5 = False
-
            try:
-                if self.proxy_manager.has_proxies():
-                    proxy = await self.proxy_manager.acquire_proxy(timeout=30)
+                proxy_config, proxy_url, proxy_info = _make_sticky_proxy_config()
+                real_ip = await _resolve_sticky_ip(proxy_url)
+                if real_ip is None:
+                    logger.warning(f"Sticky proxy IP resolve failed (attempt {attempt + 1}), retrying with new session")
+                    continue
+                _set_real_ip(real_ip)
+                proxy_locale = _locale_for_ip(real_ip)
+                logger.info(f"Using sticky proxy: {proxy_info} ip={real_ip}")
+                camoufox_kwargs = dict(
+                    headless=True,
+                    geoip=True,
+                    humanize=True,
+                    exclude_addons=[DefaultAddons.UBO],
+                    proxy=proxy_config,
+                )
+                if proxy_locale:
+                    camoufox_kwargs["locale"] = proxy_locale

-                proxy_locale = None
+                async with AsyncCamoufox(**camoufox_kwargs) as browser:
+                    result = await self._browse_page(browser, url, proxy_info, reading_time)

-                try:
-                    if proxy and proxy.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4]:
-                        proxy_config = await self.socks5_pool.get_proxy_config(proxy)
-                        use_socks5 = True
-                        real_ip = proxy.id.split("://")[-1].split(":")[0]
-                        _set_real_ip(real_ip)
-                        proxy_locale = _locale_for_ip(real_ip)
-                        has_auth = bool(proxy.login and proxy.password)
-                        logger.info(f"Using SOCKS5 tunnel for: {proxy.id} auth={'yes' if has_auth else 'NO'} locale={proxy_locale}")
-                    elif proxy:
-                        proxy_config = proxy.proxy_config
-                        real_ip = proxy.id.split("://")[-1].split(":")[0]
-                        _set_real_ip(real_ip)
-                        proxy_locale = _locale_for_ip(real_ip)
-                        logger.info(f"Using HTTP proxy: {proxy.id} locale={proxy_locale}")
-                    else:
-                        proxy_config = None
-                        _set_real_ip(None)
-                        logger.info("Direct connection")
-                except asyncio.CancelledError:
-                    # Задача отменена — прокси не виноват, освобождаем без штрафа
-                    if proxy:
-                        await self.proxy_manager.release_proxy(proxy, success=True)
-                    raise
-
-                if proxy_config:
-                    camoufox_kwargs = dict(
-                        headless=True,
-                        geoip=True,
-                        humanize=True,
-                        exclude_addons=[DefaultAddons.UBO],
-                        proxy=proxy_config,
-                    )
-                    if proxy_locale:
-                        camoufox_kwargs["locale"] = proxy_locale
-                    async with AsyncCamoufox(**camoufox_kwargs) as browser:
-                        result = await self._browse_page(
-                            browser, url,
-                            proxy.id if proxy else "direct",
-                            reading_time
-                        )
-                else:
-                    return VisitResult(url=url, success=False, error="No proxy available")
-
-                # Определяем тип ошибки: ошибка соединения (прокси сломан) vs ошибка загрузки (прокси ok)
                is_proxy_dead = result.error and any(
-                    k in result.error for k in ("502", "Bad Gateway", "SOCKS", "NS_ERROR_PROXY")
+                    k in result.error for k in ("502", "Bad Gateway", "NS_ERROR_PROXY")
                )
                is_load_error = result.error and not is_proxy_dead and any(
                    k in result.error for k in ("Timeout", "ERR_", "NS_ERROR_", "Connection")
                )

-                if proxy:
-                    # Таймаут/ошибка загрузки — прокси не виноват, освобождаем без штрафа
-                    release_success = True if is_load_error else result.success
-                    await self.proxy_manager.release_proxy(proxy, release_success)
-                    if use_socks5:
-                        await self.socks5_pool.release(proxy)
-
                if is_proxy_dead:
                    logger.warning(f"Visit attempt {attempt + 1} proxy connection failed, retrying with new proxy")
                    continue
                elif is_load_error:
-                    # Ошибка загрузки — не меняем прокси, просто возвращаем результат
-                    # (сайт мог быть временно недоступен, прокси ок)
                    logger.warning(f"Visit attempt {attempt + 1} page load error: {result.error or 'unknown'}")

+                self._proxy_fail_streak = 0
                return result

-            except (ValueError, InvalidIP) as e:
+            except (ValueError, InvalidIP, InvalidProxy) as e:
                logger.warning(f"Visit attempt {attempt + 1} proxy error, retrying: {e}")
-                if proxy:
-                    await self.proxy_manager.release_proxy(proxy, success=False)
-                    if use_socks5:
-                        await self.socks5_pool.release(proxy)

            except Exception as e:
                logger.error(f"Visit error: {e}", exc_info=True)
-                if proxy:
-                    # Ошибка кода/браузера — прокси не виноват
-                    await self.proxy_manager.release_proxy(proxy, success=True)
-                    if use_socks5:
-                        await self.socks5_pool.release(proxy)
                return VisitResult(url=url, success=False, error=str(e))

+        self._proxy_fail_streak += 1
+        if self._proxy_fail_streak >= self._PROXY_FAIL_THRESHOLD:
+            logger.error(f"Proxy fail streak {self._proxy_fail_streak} — signalling PROXY_UNAVAILABLE")
+            self._proxy_fail_streak = 0
+            return VisitResult(url=url, success=False, error="PROXY_UNAVAILABLE")
+
        return VisitResult(url=url, success=False, error="All attempts failed (fingerprint error)")

    async def _browse_page(
@@ -373,13 +370,7 @@ class BrowserService:
            return None

    async def cleanup(self):
-        """Очистка с таймаутом."""
-        try:
-            await asyncio.wait_for(self.socks5_pool.stop_all(), timeout=10)
-        except asyncio.TimeoutError:
-            logger.warning("SOCKS5 pool cleanup timeout")
-        except Exception as e:
-            logger.error(f"Cleanup error: {e}")
+        pass

    async def visit_page_from_twitch(self, url: str, channel: str, reading_time: int = None) -> VisitResult:
        return await self.visit_page(url, reading_time)
@@ -142,7 +142,6 @@ class TwitchGQLChatClient:
        on_drain_fail: Callable[[], Awaitable[None]] = None,
    ) -> dict:
        stats = {"links_found": 0, "messages": 0, "reconnects": 0}
-        start_time = datetime.now()
        last_message_time = datetime.now()
        drain_errors = 0

@@ -154,12 +153,18 @@ class TwitchGQLChatClient:
        if not await self.connect():
            raise ConnectionError(f"IRC WS connect failed for #{self.channel}")

-        while duration == 0 or (datetime.now() - start_time).seconds < duration:
+        # start_time после паузы и подключения — пауза не считается в длительность
+        start_time = datetime.now()
+
+        while duration == 0 or (datetime.now() - start_time).total_seconds() < duration:
            if is_active and not is_active():
                logger.info(f"⏸️ #{self.channel} пауза...")
                await self._cleanup()
+                pause_begin = datetime.now()
                while is_active and not is_active():
                    await asyncio.sleep(2)
+                # Исключаем время паузы из отсчёта длительности
+                start_time += datetime.now() - pause_begin
                logger.info(f"▶️ #{self.channel} resume, переподключение...")
                if not await self.connect():
                    drain_errors += 1
@@ -173,7 +173,6 @@ class TwitchIRCClient:
    ) -> dict:
        stats = {"links_found": 0, "messages": 0, "reconnects": 0}

-        start_time = datetime.now()
        last_message_time = datetime.now()
        drain_errors = 0

@@ -182,20 +181,24 @@ class TwitchIRCClient:
            logger.info(f"⏸️ #{self.channel} waiting for resume before connecting...")
            while is_active and not is_active():
                await asyncio.sleep(2)
-                if not is_active and not is_active():
-                    return stats
            logger.info(f"▶️ #{self.channel} resumed, connecting to IRC")

        if not await self.connect():
            raise ConnectionError(f"IRC connect failed for #{self.channel}")

-        while duration == 0 or (datetime.now() - start_time).seconds < duration:
+        # start_time после паузы и подключения — пауза не считается в длительность
+        start_time = datetime.now()
+
+        while duration == 0 or (datetime.now() - start_time).total_seconds() < duration:
            # === ПРОВЕРКА АКТИВНОСТИ ===
            if is_active and not is_active():
                logger.info(f"⏸️ #{self.channel} paused, waiting for resume...")
                await self.disconnect()
+                pause_begin = datetime.now()
                while is_active and not is_active():
                    await asyncio.sleep(2)
+                # Исключаем время паузы из отсчёта длительности
+                start_time += datetime.now() - pause_begin
                logger.info(f"▶️ #{self.channel} resumed, reconnecting")
                await self.connect()
                last_message_time = datetime.now()