Yuriy Yuriev
386 lines
14 KiB
Python
386 lines
14 KiB
Python
"""
|
||
Browser service с поддержкой HTTP и sticky-прокси.
|
||
"""
|
||
|
||
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 camoufox.exceptions import InvalidIP, InvalidProxy
|
||
|
||
from config.settings import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _set_real_ip(ip: Optional[str]) -> None:
|
||
import threading
|
||
threading.current_thread()._proxy_ip = ip
|
||
|
||
|
||
def _patch_camoufox():
|
||
"""Патчит camoufox чтобы:
|
||
1. Убирать screen constraint из generate_fingerprint при ошибке
|
||
2. Возвращать реальный IP SOCKS5 прокси вместо попытки запроса через туннель
|
||
"""
|
||
import camoufox.utils as _cu
|
||
import camoufox.ip as _ci
|
||
|
||
# Патч 1: generate_fingerprint — убираем screen при ошибке
|
||
_orig_gf = _cu.generate_fingerprint
|
||
|
||
def _patched_gf(**config):
|
||
try:
|
||
return _orig_gf(**config)
|
||
except ValueError:
|
||
config.pop("screen", None)
|
||
try:
|
||
return _orig_gf(**config)
|
||
except ValueError:
|
||
minimal = {k: v for k, v in config.items() if k == "locale"}
|
||
return _orig_gf(**minimal)
|
||
|
||
_cu.generate_fingerprint = _patched_gf
|
||
|
||
# Патч 2: public_ip — возвращаем реальный IP прокси вместо HTTP-запроса
|
||
# camoufox/utils.py импортирует через "from .ip import public_ip",
|
||
# поэтому патчим именно camoufox.utils.public_ip
|
||
_orig_ip = _ci.public_ip
|
||
|
||
def _patched_ip(proxy_string: str = None) -> str:
|
||
# Вызывается из thread pool — используем thread-local атрибут
|
||
import threading
|
||
ip = getattr(threading.current_thread(), "_proxy_ip", None)
|
||
if ip:
|
||
logger.debug(f"public_ip() intercepted → returning {ip}")
|
||
return ip
|
||
logger.warning(f"public_ip() not set, proxy_string={proxy_string}")
|
||
if proxy_string is not None:
|
||
return _orig_ip(proxy_string)
|
||
return _orig_ip()
|
||
|
||
_ci.public_ip = _patched_ip
|
||
_cu.public_ip = _patched_ip
|
||
|
||
_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:
|
||
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 и sticky-прокси.
|
||
"""
|
||
|
||
_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,
|
||
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
|
||
)
|
||
|
||
if not settings.STICKY_PROXY_HOST:
|
||
return VisitResult(url=url, success=False, error="No proxy configured")
|
||
|
||
# Максимальное время одной попытки: запуск браузера + загрузка страницы + чтение + буфер
|
||
attempt_timeout = reading_time + 150
|
||
|
||
for attempt in range(3):
|
||
try:
|
||
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
|
||
|
||
async def _run_browser():
|
||
async with AsyncCamoufox(**camoufox_kwargs) as browser:
|
||
return await self._browse_page(browser, url, proxy_info, reading_time)
|
||
|
||
result = await asyncio.wait_for(_run_browser(), timeout=attempt_timeout)
|
||
|
||
is_proxy_dead = result.error and any(
|
||
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 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 asyncio.TimeoutError:
|
||
logger.warning(f"Visit attempt {attempt + 1} timed out after {attempt_timeout}s, retrying")
|
||
|
||
except (ValueError, InvalidIP, InvalidProxy) as e:
|
||
logger.warning(f"Visit attempt {attempt + 1} proxy error, retrying: {e}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Visit error: {e}", exc_info=True)
|
||
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(
|
||
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
|
||
if proxy_info == "direct" :
|
||
return VisitResult(
|
||
url=url,
|
||
initial_url=initial_url,
|
||
final_url=final_url,
|
||
redirect_occurred=redirect,
|
||
proxy_info=proxy_info,
|
||
reading_time=reading_time,
|
||
)
|
||
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=60000)
|
||
)
|
||
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:
|
||
if not reading_completed:
|
||
logger.warning(f"Browse error [{proxy_info}] {url[:50]}: {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):
|
||
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)
|