""" Browser service с поддержкой SOCKS5 через локальный HTTP туннель. """ import asyncio import random import logging from pathlib import Path from datetime import datetime from typing import Optional from dataclasses import dataclass from camoufox import DefaultAddons from camoufox.async_api import AsyncCamoufox 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__) @dataclass class VisitResult: """Результат посещения страницы.""" url: str initial_url: Optional[str] = None final_url: Optional[str] = None redirect_occurred: bool = False proxy_info: str = "direct" reading_time: int = 0 screenshot_path: Optional[str] = None success: bool = False error: Optional[str] = None class BrowserService: """ Сервис для посещения страниц через Camoufox. Поддерживает HTTP, SOCKS5 (через туннель) и прямое соединение. """ def __init__(self, proxy_manager: ProxyManager): self.proxy_manager = proxy_manager self.socks5_pool = Socks5ProxyPool(idle_timeout=300) self.screenshots_dir = Path(settings.SCREENSHOTS_DIR) self.screenshots_dir.mkdir(exist_ok=True) async def visit_page( self, url: str, reading_time: Optional[int] = None ) -> VisitResult: """Посещает страницу.""" if reading_time is None: reading_time = random.randint( settings.DEFAULT_MIN_READING, settings.DEFAULT_MAX_READING ) proxy = None use_socks5 = False try: # Получаем прокси if self.proxy_manager.has_proxies(): proxy = await self.proxy_manager.acquire_proxy(timeout=30) # Определяем конфигурацию if proxy and proxy.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4]: # SOCKS5 через локальный HTTP туннель proxy_config = await self.socks5_pool.get_proxy_config(proxy) use_socks5 = True logger.info(f"Using SOCKS5 tunnel for: {proxy.id}") elif proxy: # HTTP прокси напрямую proxy_config = proxy.proxy_config logger.info(f"Using HTTP proxy: {proxy.id}") else: # Прямое соединение proxy_config = None logger.info("Direct connection") # Запускаем браузер if proxy_config: async with AsyncCamoufox( headless=False, geoip=True, humanize=True, exclude_addons=[DefaultAddons.UBO], proxy=proxy_config ) as browser: result = await self._browse_page( browser, url, proxy.id if proxy else "direct", reading_time ) else: async with AsyncCamoufox( headless=False, geoip=False, humanize=True, locale="ru-RU", exclude_addons=[DefaultAddons.UBO] ) as browser: result = await self._browse_page( browser, url, "direct", reading_time ) # Освобождаем if proxy: await self.proxy_manager.release_proxy(proxy, result.success) if use_socks5: await self.socks5_pool.release(proxy) return result except Exception as e: logger.error(f"Visit error: {e}", exc_info=True) if proxy: await self.proxy_manager.release_proxy(proxy, success=False) if use_socks5: await self.socks5_pool.release(proxy) return VisitResult(url=url, success=False, error=str(e)) async def _browse_page( self, browser, url: str, proxy_info: str, reading_time: int ) -> VisitResult: """Выполняет просмотр страницы.""" page = await browser.new_page() try: await page.set_viewport_size({ "width": settings.VIEWPORT_WIDTH, "height": settings.VIEWPORT_HEIGHT }) goto_task = asyncio.create_task( page.goto(url, wait_until="commit", timeout=30000) ) await self._move_mouse(page, goto_task) await goto_task initial_url = page.url final_url, redirect = await self._wait_redirect(page, initial_url) await self._simulate_reading(page, reading_time) screenshot_path = None if settings.SCREENSHOTS_DIR: screenshot_path = await self._take_screenshot(page, final_url) return VisitResult( url=url, initial_url=initial_url, final_url=final_url, redirect_occurred=redirect, proxy_info=proxy_info, reading_time=reading_time, screenshot_path=screenshot_path, success=True ) except Exception as e: return VisitResult(url=url, reading_time=reading_time, success=False, error=str(e)) finally: if page.url == url: logger.error() await page.close() async def _move_mouse(self, page, task): """Двигает мышь.""" w, h = settings.VIEWPORT_WIDTH, settings.VIEWPORT_HEIGHT try: while not task.done(): await page.mouse.move( random.randint(100, w-100), random.randint(100, h-100) ) await asyncio.sleep(random.uniform(0.1, 0.3)) except: pass async def _wait_redirect(self, page, initial_url): """Ждет редирект.""" loop = asyncio.get_running_loop() start = loop.time() redirect = False current = initial_url while loop.time() - start < settings.MAX_REDIRECT_WAIT: await asyncio.sleep(2) new_url = page.url if new_url != current: current = new_url redirect = True task = asyncio.create_task( page.wait_for_load_state("load", timeout=30000) ) await self._move_mouse(page, task) await task break await self._random_action(page) return page.url, redirect async def _simulate_reading(self, page, duration): """Симулирует чтение.""" loop = asyncio.get_running_loop() start = loop.time() while loop.time() - start < duration: await self._random_action(page) async def _random_action(self, page): """Случайное действие.""" try: action = random.choice(['move', 'scroll', 'pause']) if action == 'move': x = random.randint(100, settings.VIEWPORT_WIDTH-100) y = random.randint(100, settings.VIEWPORT_HEIGHT-100) await page.mouse.move(x, y) await asyncio.sleep(random.uniform(0.3, 1.0)) elif action == 'scroll': await page.evaluate(f"window.scrollBy(0, {random.randint(100, 300)})") await asyncio.sleep(random.uniform(0.3, 0.8)) else: await asyncio.sleep(random.uniform(1, 3)) except: await asyncio.sleep(0.5) async def _take_screenshot(self, page, url): """Скриншот.""" try: ts = datetime.now().strftime("%Y%m%d_%H%M%S") domain = url.replace("https://", "").replace("http://", "").split("/")[0][:30] fp = self.screenshots_dir / f"{ts}_{domain}.png" await page.screenshot(path=str(fp), full_page=True) return str(fp) except Exception as e: logger.error(f"Screenshot: {e}") 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}") async def visit_page_from_twitch(self, url: str, channel: str, reading_time: int = None) -> VisitResult: if reading_time is None: reading_time = random.randint(settings.DEFAULT_MIN_READING, settings.DEFAULT_MAX_READING) proxy = None for attempt in range(2): # Две попытки с разными прокси try: if self.proxy_manager.has_proxies(): proxy = await self.proxy_manager.acquire_proxy(timeout=90) if proxy and proxy.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4]: proxy_config = await self.socks5_pool.get_proxy_config(proxy) elif proxy: proxy_config = proxy.proxy_config else: proxy_config = None browser_kwargs = { "headless": False, "geoip": True, "humanize": True, "exclude_addons": [DefaultAddons.UBO], } if proxy_config: browser_kwargs["proxy"] = proxy_config async with AsyncCamoufox(**browser_kwargs) as browser: result = await self._browse_page( browser, url, proxy.id if proxy else "direct", reading_time ) if proxy: await self.proxy_manager.release_proxy(proxy, result.success) if result.success: return result except Exception as e: logger.warning(f"Attempt {attempt + 1} failed: {e}") if proxy: await self.proxy_manager.release_proxy(proxy, success=False) await self.socks5_pool.release(proxy) proxy = None if attempt < 1: await asyncio.sleep(2) return VisitResult(url=url, success=False, error="All attempts failed") async def _browse_twitch_ref(self, browser, url: str, channel: str, proxy_info: str, reading_time: int) -> VisitResult: """Посещает страницу как переход с Twitch канала.""" page = await browser.new_page() twitch_url = f"https://www.twitch.tv/{channel}" try: await page.set_viewport_size({"width": settings.VIEWPORT_WIDTH, "height": settings.VIEWPORT_HEIGHT}) # Заголовки имитирующие переход с Twitch await page.set_extra_http_headers({ "Referer": twitch_url, "Origin": "https://www.twitch.tv", }) logger.info(f"📺 From Twitch: {channel} → {url}") # Переход с Referer goto = asyncio.create_task(page.goto(url, referer=twitch_url, wait_until="commit", timeout=30000)) await self._move_mouse(page, goto) await goto initial_url = page.url final_url, redirect = await self._wait_redirect(page, initial_url) await self._simulate_reading(page, reading_time) return VisitResult( url=url, initial_url=initial_url, final_url=final_url, redirect_occurred=redirect, proxy_info=proxy_info, reading_time=reading_time, success=True ) except Exception as e: return VisitResult(url=url, success=False, error=str(e)) finally: await page.close() async def _browse_from_twitch( self, browser, url: str, channel: str, proxy_info: str, reading_time: int ) -> VisitResult: """ Эмулирует переход с Twitch: 1. Открывает страницу стримера 2. Прокручивает чат 3. Переходит по ссылке """ page = await browser.new_page() twitch_url = f"https://www.twitch.tv/{channel}" try: await page.set_viewport_size({ "width": settings.VIEWPORT_WIDTH, "height": settings.VIEWPORT_HEIGHT }) # === Шаг 1: Заходим на Twitch === logger.info(f"📺 Opening Twitch: {twitch_url}") goto_task = asyncio.create_task( page.goto(twitch_url, wait_until="commit", timeout=30000) ) await self._move_mouse(page, goto_task) await goto_task # Ждем загрузки await asyncio.sleep(random.uniform(2, 4)) # === Шаг 2: Имитируем просмотр стрима === logger.info("👀 Watching stream...") # Прокручиваем страницу как будто смотрим for _ in range(random.randint(2, 4)): scroll = random.randint(200, 500) await page.evaluate(f"window.scrollBy(0, {scroll})") await asyncio.sleep(random.uniform(0.5, 1.5)) # Двигаем мышь (как будто читаем чат) w, h = settings.VIEWPORT_WIDTH, settings.VIEWPORT_HEIGHT for _ in range(random.randint(3, 6)): x = random.randint(100, w - 100) y = random.randint(100, h - 100) await page.mouse.move(x, y) await asyncio.sleep(random.uniform(0.3, 0.8)) # === Шаг 3: Переходим по ссылке === logger.info(f"🔗 Clicking link: {url}") # Создаем новую вкладку для перехода (как target="_blank") # Или просто переходим с Referer await page.evaluate(f""" window.open('{url}', '_blank'); """) # Ждем открытия новой вкладки await asyncio.sleep(2) # Получаем новую вкладку pages = await browser.pages() if len(pages) > 1: new_page = pages[-1] await new_page.bring_to_front() else: # Если вкладка не открылась - переходим в этой же new_page = page await new_page.goto(url, referer=twitch_url, wait_until="commit") initial_url = new_page.url logger.info(f"📍 Landed: {initial_url}") # === Шаг 4: Ждем редирект === final_url, redirect = await self._wait_redirect(new_page, initial_url) # === Шаг 5: Читаем страницу === await self._simulate_reading(new_page, reading_time) # Закрываем новую вкладку если она отдельная if new_page != page: await new_page.close() # Возвращаемся на Twitch и закрываем await page.close() return VisitResult( url=url, initial_url=initial_url, final_url=final_url, redirect_occurred=redirect, proxy_info=proxy_info, reading_time=reading_time, success=True ) except Exception as e: logger.error(f"Browse from Twitch error: {e}") try: await page.close() except: pass return VisitResult(url=url, reading_time=reading_time, success=False, error=str(e))