Yuriy Yuriev
353 lines
13 KiB
Python
353 lines
13 KiB
Python
"""
|
||
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__)
|
||
|
||
|
||
def _locale_for_ip(ip: str) -> str:
|
||
"""Возвращает BCP47 locale без script-тега по реальному IP прокси."""
|
||
try:
|
||
from camoufox.utils import get_geolocation
|
||
geo = get_geolocation(ip)
|
||
lang = geo.locale.language
|
||
region = geo.locale.region
|
||
if lang and region:
|
||
return f"{lang}-{region}"
|
||
except Exception:
|
||
pass
|
||
return "en-US"
|
||
|
||
|
||
@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
|
||
)
|
||
|
||
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)
|
||
|
||
socks5_locale = None
|
||
if proxy and proxy.proxy_type in [ProxyType.SOCKS5, ProxyType.SOCKS4]:
|
||
proxy_config = await self.socks5_pool.get_proxy_config(proxy)
|
||
use_socks5 = True
|
||
# При SOCKS5 camoufox видит 127.0.0.1 → geoip берёт IP сервера → ru-Cyrl-RU
|
||
# Делаем lookup по реальному IP прокси сами и передаём чистый locale
|
||
real_ip = proxy.id.split("://")[-1].split(":")[0]
|
||
socks5_locale = _locale_for_ip(real_ip)
|
||
logger.info(f"Using SOCKS5 tunnel for: {proxy.id} locale={socks5_locale}")
|
||
elif proxy:
|
||
proxy_config = proxy.proxy_config
|
||
logger.info(f"Using HTTP proxy: {proxy.id}")
|
||
else:
|
||
proxy_config = None
|
||
logger.info("Direct connection")
|
||
|
||
if proxy_config:
|
||
camoufox_kwargs = dict(
|
||
headless=True,
|
||
geoip=True,
|
||
humanize=True,
|
||
exclude_addons=[DefaultAddons.UBO],
|
||
proxy=proxy_config,
|
||
)
|
||
if socks5_locale:
|
||
camoufox_kwargs["locale"] = socks5_locale
|
||
async with AsyncCamoufox(**camoufox_kwargs) as browser:
|
||
result = await self._browse_page(
|
||
browser, url,
|
||
proxy.id if proxy else "direct",
|
||
reading_time
|
||
)
|
||
else:
|
||
async with AsyncCamoufox(
|
||
headless=True,
|
||
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 ValueError as e:
|
||
logger.warning(f"Visit attempt {attempt + 1} fingerprint error, retrying with new proxy: {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=False)
|
||
if use_socks5:
|
||
await self.socks5_pool.release(proxy)
|
||
return VisitResult(url=url, success=False, error=str(e))
|
||
|
||
return VisitResult(url=url, success=False, error="All attempts failed (fingerprint error)")
|
||
|
||
async def _browse_page(
|
||
self, browser, url: str, proxy_info: str, reading_time: int
|
||
) -> VisitResult:
|
||
"""Выполняет просмотр страницы."""
|
||
page = await browser.new_page()
|
||
initial_url = url
|
||
final_url = url
|
||
redirect = False
|
||
reading_completed = False
|
||
|
||
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
|
||
|
||
try:
|
||
final_url, redirect = await self._wait_redirect(page, initial_url)
|
||
except Exception as e:
|
||
logger.warning(f"Redirect wait error (ignored): {e}")
|
||
final_url = page.url
|
||
|
||
await self._simulate_reading(page, reading_time)
|
||
reading_completed = True
|
||
|
||
screenshot_path = None
|
||
if settings.SCREENSHOTS_DIR:
|
||
try:
|
||
screenshot_path = await self._take_screenshot(page, final_url)
|
||
except Exception:
|
||
pass
|
||
|
||
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 asyncio.CancelledError:
|
||
raise
|
||
except Exception as e:
|
||
return VisitResult(
|
||
url=url,
|
||
initial_url=initial_url,
|
||
final_url=final_url,
|
||
redirect_occurred=redirect,
|
||
proxy_info=proxy_info,
|
||
reading_time=reading_time,
|
||
success=reading_completed,
|
||
error=str(e) if not reading_completed else None
|
||
)
|
||
finally:
|
||
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 Exception:
|
||
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 Exception:
|
||
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": True,
|
||
"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")
|