fix(browser): local camoufox + HTTP wrapper, drop broken playwright WS server (#905) #909
5 changed files with 418 additions and 392 deletions
|
|
@ -207,21 +207,24 @@ class Settings(BaseSettings):
|
|||
# ENV: YANDEX_COOKIES_FILE.
|
||||
yandex_cookies_file: str | None = None
|
||||
|
||||
# ── #884/#905: BrowserFetcher — connect-mode к tradein-browser контейнеру ──
|
||||
# ── #884/#905: BrowserFetcher — HTTP-клиент к tradein-browser контейнеру ────
|
||||
# scraper_fetch_mode: "curl_cffi" (дефолт, текущее поведение) или "browser"
|
||||
# (playwright.firefox.connect к tradein-browser WS-серверу). Phase 1+.
|
||||
# (HTTP POST к tradein-browser /fetch эндпоинту). Phase 1+.
|
||||
# Пока дефолт "curl_cffi" — никаких изменений в рантайме.
|
||||
# ENV: SCRAPER_FETCH_MODE.
|
||||
scraper_fetch_mode: Literal["curl_cffi", "browser"] = "curl_cffi"
|
||||
# WS-эндпоинт tradein-browser сервиса. В Docker-сети — имя сервиса из compose.
|
||||
# ENV: BROWSER_WS_ENDPOINT.
|
||||
browser_ws_endpoint: str = "ws://tradein-browser:3000/playwright"
|
||||
# Сколько страниц обработать в одном browser-сеансе перед переподключением
|
||||
# (ограничение утечек памяти на стороне браузер-контейнера). ENV: BROWSER_RECYCLE_PAGES.
|
||||
# HTTP-эндпоинт tradein-browser сервиса. В Docker-сети — имя сервиса из compose.
|
||||
# ENV: BROWSER_HTTP_ENDPOINT.
|
||||
browser_http_endpoint: str = "http://tradein-browser:3000"
|
||||
# Сколько страниц обработать в одном browser-сеансе перед перезапуском браузера
|
||||
# (ограничение утечек памяти). Читается сервером из env BROWSER_RECYCLE_PAGES.
|
||||
# Оставлено для справки / compat. ENV: BROWSER_RECYCLE_PAGES.
|
||||
browser_recycle_pages: int = 15
|
||||
# Таймаут навигации (page.goto) в мс. ENV: BROWSER_NAV_TIMEOUT_MS.
|
||||
# Таймаут навигации (page.goto) в мс. Читается сервером из env BROWSER_NAV_TIMEOUT_MS.
|
||||
# ENV: BROWSER_NAV_TIMEOUT_MS.
|
||||
browser_nav_timeout_ms: int = 60000
|
||||
# Ожидание после DOMContentLoaded для JS-гидрации в мс. ENV: BROWSER_WAIT_MS.
|
||||
# Ожидание после DOMContentLoaded для JS-гидрации в мс. Читается сервером.
|
||||
# ENV: BROWSER_WAIT_MS.
|
||||
browser_wait_ms: int = 2500
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,33 +1,38 @@
|
|||
"""browser_fetcher.py — Playwright connect-mode клиент для browser-mode scraping (#884/#905).
|
||||
"""browser_fetcher.py — HTTP-клиент к tradein-browser сервису (#884/#905).
|
||||
|
||||
Провайдер-агностичная основа: не затрагивает существующую avito/cian-логику.
|
||||
Активируется только когда settings.scraper_fetch_mode == "browser" (дефолт
|
||||
"curl_cffi" — поведение идентично старому до явного переключения).
|
||||
Тонкий клиент над tradein-browser контейнером, который запускает AsyncCamoufox
|
||||
локально и экспонирует HTTP API (POST /fetch).
|
||||
|
||||
Архитектура (Phase 0.5, #905):
|
||||
- Camoufox Firefox запущен в отдельном контейнере ``tradein-browser`` как
|
||||
Playwright WS-сервер (camoufox.server.launch_server).
|
||||
- Этот модуль подключается к нему через ``playwright.async_api.firefox.connect``
|
||||
используя эндпоинт из settings.browser_ws_endpoint.
|
||||
- Proxy/fingerprint/locale/geoip настраиваются на стороне browser-сервера;
|
||||
этот модуль управляет только жизненным циклом соединения и страниц.
|
||||
Публичный интерфейс не изменился:
|
||||
|
||||
Особенности:
|
||||
- Ленивые импорты playwright: не ломают app-импорт если не установлен.
|
||||
- Page recycle: каждые settings.browser_recycle_pages страниц закрываем и
|
||||
переподключаемся (ограничение утечек памяти на стороне браузер-контейнера).
|
||||
- Crash-recovery: TargetClosedError / connection-closed → переподключение → один retry.
|
||||
async with BrowserFetcher() as fetcher:
|
||||
html = await fetcher.fetch("https://example.com")
|
||||
|
||||
Внутреннее устройство: httpx.AsyncClient + POST к settings.browser_http_endpoint.
|
||||
Recycle, crash-recovery и управление браузером живут на стороне сервера (server.py).
|
||||
При HTTPError / ConnectError делает одну повторную попытку после короткой паузы,
|
||||
затем пробрасывает исключение.
|
||||
|
||||
Архитектура выбрана потому, что playwright WS-сервер (launch_server) несовместим
|
||||
с playwright >=1.45: ``browserServerImpl.js`` отсутствует → MODULE_NOT_FOUND.
|
||||
Локальный AsyncCamoufox + HTTP — работающая альтернатива.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_RETRY_SLEEP_S: float = 1.0
|
||||
_HTTP_TIMEOUT_S: float = 120.0 # навигация медленная → щедрый таймаут
|
||||
|
||||
|
||||
class BrowserFetcher:
|
||||
"""Async context manager: Playwright-клиент подключения к tradein-browser WS-серверу.
|
||||
"""Async context manager: HTTP-клиент к tradein-browser HTTP-сервису.
|
||||
|
||||
Использование::
|
||||
|
||||
|
|
@ -36,164 +41,64 @@ class BrowserFetcher:
|
|||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._browser: object | None = None
|
||||
self._pw: object | None = None
|
||||
self._page_count: int = 0
|
||||
# Lazy-loaded настройки чтобы не импортировать settings на верхнем уровне
|
||||
self._recycle_after: int | None = None
|
||||
self._nav_timeout_ms: int | None = None
|
||||
self._wait_ms: int | None = None
|
||||
self._ws_endpoint: str | None = None
|
||||
self._client: httpx.AsyncClient | None = None
|
||||
self._endpoint: str | None = None
|
||||
|
||||
# ── lifecycle ──────────────────────────────────────────────────────────────
|
||||
|
||||
async def __aenter__(self) -> BrowserFetcher:
|
||||
from app.core.config import settings
|
||||
|
||||
self._recycle_after = settings.browser_recycle_pages
|
||||
self._nav_timeout_ms = settings.browser_nav_timeout_ms
|
||||
self._wait_ms = settings.browser_wait_ms
|
||||
self._ws_endpoint = settings.browser_ws_endpoint
|
||||
await self._connect()
|
||||
self._endpoint = settings.browser_http_endpoint
|
||||
self._client = httpx.AsyncClient(timeout=_HTTP_TIMEOUT_S)
|
||||
logger.info("BrowserFetcher: клиент создан, endpoint=%s", self._endpoint)
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *_: object) -> None:
|
||||
await self._close()
|
||||
if self._client is not None:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
logger.debug("BrowserFetcher: клиент закрыт")
|
||||
|
||||
# ── public API ─────────────────────────────────────────────────────────────
|
||||
|
||||
async def fetch(self, url: str) -> str:
|
||||
"""Открывает новую страницу, переходит по URL, ждёт JS, возвращает HTML.
|
||||
"""Запрашивает HTML страницы через tradein-browser HTTP-сервис.
|
||||
|
||||
При TargetClosedError или connection-closed (crash/перезапуск браузер-контейнера)
|
||||
переподключается и делает одну повторную попытку. Остальные исключения
|
||||
всплывают к вызывающему коду.
|
||||
При HTTPError или ConnectError делает одну повторную попытку после
|
||||
короткой паузы. Остальные исключения всплывают к вызывающему коду.
|
||||
|
||||
Returns:
|
||||
Полный HTML-контент страницы (innerHTML documentElement).
|
||||
Полный HTML-контент страницы.
|
||||
"""
|
||||
assert self._client is not None, "BrowserFetcher: используй как async context manager"
|
||||
assert self._endpoint is not None, "BrowserFetcher: endpoint не задан"
|
||||
|
||||
try:
|
||||
return await self._fetch_once(url)
|
||||
except Exception as exc:
|
||||
if _is_target_closed(exc) or _is_crash_exception(exc):
|
||||
logger.warning(
|
||||
"BrowserFetcher: соединение с браузером потеряно (%s),"
|
||||
" переподключение + retry: %s",
|
||||
type(exc).__name__,
|
||||
url,
|
||||
)
|
||||
await self._reconnect()
|
||||
return await self._fetch_once(url)
|
||||
raise
|
||||
return await self._post_fetch(url)
|
||||
except (httpx.HTTPError, httpx.TransportError) as exc:
|
||||
logger.warning(
|
||||
"BrowserFetcher: ошибка запроса (%s), retry через %.1fs: %s",
|
||||
type(exc).__name__,
|
||||
_RETRY_SLEEP_S,
|
||||
url,
|
||||
)
|
||||
await asyncio.sleep(_RETRY_SLEEP_S)
|
||||
return await self._post_fetch(url)
|
||||
|
||||
# ── internal ───────────────────────────────────────────────────────────────
|
||||
|
||||
async def _fetch_once(self, url: str) -> str:
|
||||
"""Одна попытка: new_page → goto → wait → content → close."""
|
||||
assert self._browser is not None, "BrowserFetcher: нет активного соединения"
|
||||
async def _post_fetch(self, url: str) -> str:
|
||||
"""Один HTTP POST к /fetch эндпоинту сервиса."""
|
||||
assert self._client is not None
|
||||
assert self._endpoint is not None
|
||||
|
||||
browser = self._browser # type: ignore[assignment]
|
||||
|
||||
page = await browser.new_page() # type: ignore[attr-defined]
|
||||
try:
|
||||
await page.goto(url, timeout=self._nav_timeout_ms, wait_until="domcontentloaded")
|
||||
if self._wait_ms and self._wait_ms > 0:
|
||||
await page.wait_for_timeout(self._wait_ms)
|
||||
html: str = await page.content()
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
self._page_count += 1
|
||||
logger.debug(
|
||||
"BrowserFetcher: fetch OK url=%r pages_since_connect=%d", url, self._page_count
|
||||
resp = await self._client.post(
|
||||
f"{self._endpoint}/fetch",
|
||||
json={"url": url},
|
||||
)
|
||||
|
||||
# Recycle: переподключаемся каждые N страниц
|
||||
assert self._recycle_after is not None
|
||||
if self._page_count >= self._recycle_after:
|
||||
logger.info(
|
||||
"BrowserFetcher: recycle threshold (%d) достигнут, переподключение",
|
||||
self._recycle_after,
|
||||
)
|
||||
await self._reconnect()
|
||||
|
||||
resp.raise_for_status()
|
||||
data: dict[str, str] = resp.json()
|
||||
html = data["html"]
|
||||
logger.debug("BrowserFetcher: fetch OK url=%r html_len=%d", url, len(html))
|
||||
return html
|
||||
|
||||
async def _connect(self) -> None:
|
||||
"""Запускает Playwright и подключается к WS-серверу tradein-browser."""
|
||||
try:
|
||||
from playwright.async_api import async_playwright
|
||||
except ImportError as exc:
|
||||
raise RuntimeError(
|
||||
"playwright не установлен. Добавь 'playwright>=1.45' в pyproject.toml."
|
||||
) from exc
|
||||
|
||||
assert self._ws_endpoint, "BrowserFetcher: browser_ws_endpoint не задан"
|
||||
|
||||
self._pw = await async_playwright().start()
|
||||
pw = self._pw # type: ignore[assignment]
|
||||
self._browser = await pw.firefox.connect( # type: ignore[attr-defined]
|
||||
self._ws_endpoint,
|
||||
timeout=30000, # 30s на установку соединения
|
||||
)
|
||||
self._page_count = 0
|
||||
logger.info(
|
||||
"BrowserFetcher: подключился к %s",
|
||||
self._ws_endpoint,
|
||||
)
|
||||
|
||||
async def _close(self) -> None:
|
||||
"""Закрывает соединение с браузером и останавливает Playwright."""
|
||||
browser = self._browser
|
||||
if browser is not None:
|
||||
try:
|
||||
await browser.close() # type: ignore[attr-defined]
|
||||
logger.info("BrowserFetcher: соединение закрыто")
|
||||
except Exception as exc:
|
||||
logger.warning("BrowserFetcher: ошибка при закрытии соединения: %s", exc)
|
||||
finally:
|
||||
self._browser = None
|
||||
|
||||
pw = self._pw
|
||||
if pw is not None:
|
||||
try:
|
||||
await pw.stop() # type: ignore[attr-defined]
|
||||
except Exception as exc:
|
||||
logger.warning("BrowserFetcher: ошибка при остановке Playwright: %s", exc)
|
||||
finally:
|
||||
self._pw = None
|
||||
|
||||
self._page_count = 0
|
||||
|
||||
async def _reconnect(self) -> None:
|
||||
"""Закрывает текущее соединение и устанавливает новое."""
|
||||
await self._close()
|
||||
await self._connect()
|
||||
|
||||
|
||||
# ── helpers ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _is_target_closed(exc: BaseException) -> bool:
|
||||
"""Проверяет является ли исключение playwright TargetClosedError.
|
||||
|
||||
Проверка по имени класса (не import) — playwright может не быть в dev-окружении.
|
||||
"""
|
||||
return type(exc).__name__ == "TargetClosedError"
|
||||
|
||||
|
||||
def _is_crash_exception(exc: BaseException) -> bool:
|
||||
"""Проверяет является ли исключение признаком краша / разрыва соединения.
|
||||
|
||||
Ловим широкий набор playwright crash/disconnect сигналов по имени/сообщению.
|
||||
"""
|
||||
cls_name = type(exc).__name__
|
||||
if cls_name in ("BrowserClosedError", "ConnectionClosedError"):
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return (
|
||||
"browser has been closed" in msg
|
||||
or "target closed" in msg
|
||||
or "connection closed" in msg
|
||||
or ("websocket" in msg and "closed" in msg)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,12 +1,13 @@
|
|||
"""Tests for BrowserFetcher (#884/#905, connect-mode).
|
||||
"""Tests for BrowserFetcher (#884/#905, HTTP-клиент).
|
||||
|
||||
Полностью замоканы: не требуют установленного Firefox, camoufox или реального
|
||||
WS-сервера. Покрывают:
|
||||
- fetch(): new_page → goto → wait_for_timeout → content → close
|
||||
- recycle: счётчик страниц → reconnect при достижении порога
|
||||
- crash-recovery: TargetClosedError → reconnect → retry
|
||||
- page counter сбрасывается после reconnect
|
||||
- non-crash exception пробрасывается без перехвата
|
||||
Полностью замоканы через monkeypatch на httpx.AsyncClient.post:
|
||||
не требуют установленного Firefox, camoufox или реального HTTP-сервера.
|
||||
|
||||
Покрывают:
|
||||
- fetch(): POST к правильному endpoint + разбор {"html": "..."} из ответа
|
||||
- HTTPError триггерит один retry, затем поднимает исключение
|
||||
- URL строится из settings.browser_http_endpoint
|
||||
- импорт модуля не ломается без playwright/camoufox
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -15,6 +16,7 @@ import os
|
|||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
|
@ -24,240 +26,181 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:
|
|||
|
||||
|
||||
def _mock_settings(
|
||||
proxy_url: str | None = None,
|
||||
recycle_pages: int = 15,
|
||||
nav_timeout_ms: int = 60000,
|
||||
wait_ms: int = 2500,
|
||||
ws_endpoint: str = "ws://fake-browser:3000/playwright",
|
||||
browser_http_endpoint: str = "http://fake-browser:3000",
|
||||
) -> SimpleNamespace:
|
||||
return SimpleNamespace(
|
||||
scraper_proxy_url=proxy_url,
|
||||
browser_recycle_pages=recycle_pages,
|
||||
browser_nav_timeout_ms=nav_timeout_ms,
|
||||
browser_wait_ms=wait_ms,
|
||||
browser_ws_endpoint=ws_endpoint,
|
||||
return SimpleNamespace(browser_http_endpoint=browser_http_endpoint)
|
||||
|
||||
|
||||
def _make_ok_response(html: str = "<html>ok</html>") -> MagicMock:
|
||||
"""Создаёт мок httpx.Response с JSON {"html": html} и статусом 200."""
|
||||
resp = MagicMock(spec=httpx.Response)
|
||||
resp.status_code = 200
|
||||
resp.json = MagicMock(return_value={"html": html})
|
||||
resp.raise_for_status = MagicMock() # нет исключения
|
||||
return resp
|
||||
|
||||
|
||||
def _make_error_response(status: int = 500) -> MagicMock:
|
||||
"""Создаёт мок httpx.Response который кидает HTTPStatusError при raise_for_status."""
|
||||
resp = MagicMock(spec=httpx.Response)
|
||||
resp.status_code = status
|
||||
resp.raise_for_status = MagicMock(
|
||||
side_effect=httpx.HTTPStatusError(
|
||||
f"Server error {status}",
|
||||
request=MagicMock(),
|
||||
response=resp,
|
||||
)
|
||||
)
|
||||
return resp
|
||||
|
||||
|
||||
def _make_mock_page(html: str = "<html>ok</html>") -> MagicMock:
|
||||
"""Создаёт мок страницы playwright с нужными async-методами."""
|
||||
page = MagicMock()
|
||||
page.goto = AsyncMock()
|
||||
page.wait_for_timeout = AsyncMock()
|
||||
page.content = AsyncMock(return_value=html)
|
||||
page.close = AsyncMock()
|
||||
return page
|
||||
# ── импорт не ломается ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_playwright_stack(
|
||||
page: MagicMock | None = None,
|
||||
html: str = "<html>ok</html>",
|
||||
) -> tuple[MagicMock, MagicMock, MagicMock]:
|
||||
"""Создаёт полную mock-цепочку playwright: pw_cm → pw → browser → page.
|
||||
|
||||
Returns:
|
||||
(pw_cm, pw_mock, browser_mock)
|
||||
"""
|
||||
if page is None:
|
||||
page = _make_mock_page(html=html)
|
||||
|
||||
browser_mock = MagicMock()
|
||||
browser_mock.new_page = AsyncMock(return_value=page)
|
||||
browser_mock.close = AsyncMock()
|
||||
|
||||
firefox_mock = MagicMock()
|
||||
firefox_mock.connect = AsyncMock(return_value=browser_mock)
|
||||
|
||||
pw_mock = MagicMock()
|
||||
pw_mock.firefox = firefox_mock
|
||||
pw_mock.stop = AsyncMock()
|
||||
|
||||
async def _start() -> MagicMock:
|
||||
return pw_mock
|
||||
|
||||
pw_cm = MagicMock()
|
||||
pw_cm.start = _start
|
||||
|
||||
return pw_cm, pw_mock, browser_mock
|
||||
def test_import_browser_fetcher() -> None:
|
||||
"""import app.services.scrapers.browser_fetcher не поднимает исключений."""
|
||||
import app.services.scrapers.browser_fetcher # noqa: F401
|
||||
|
||||
|
||||
# ── fetch(): new_page → goto → content → close ────────────────────────────────
|
||||
# ── fetch(): возвращает HTML из JSON-ответа ────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_calls_page_lifecycle():
|
||||
"""fetch() вызывает new_page, goto, wait_for_timeout, content, close."""
|
||||
async def test_fetch_returns_html_from_json_response() -> None:
|
||||
"""fetch() разбирает {'html': ...} из 200-ответа и возвращает строку."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
expected_html = "<html><body>test</body></html>"
|
||||
page = _make_mock_page(html=expected_html)
|
||||
pw_cm, _pw, browser = _make_playwright_stack(page=page)
|
||||
ms = _mock_settings(recycle_pages=15, nav_timeout_ms=60000, wait_ms=2500)
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
with patch("playwright.async_api.async_playwright", return_value=pw_cm):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
result = await fetcher.fetch("https://example.com")
|
||||
|
||||
assert result == expected_html
|
||||
browser.new_page.assert_called_once()
|
||||
page.goto.assert_called_once_with(
|
||||
"https://example.com", timeout=60000, wait_until="domcontentloaded"
|
||||
)
|
||||
page.wait_for_timeout.assert_called_once_with(2500)
|
||||
page.content.assert_called_once()
|
||||
page.close.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_returns_html_string():
|
||||
"""fetch() возвращает строку HTML от page.content()."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
html = "<html><title>Avito</title></html>"
|
||||
pw_cm, _pw, _browser = _make_playwright_stack(html=html)
|
||||
expected_html = "<html><body>avito</body></html>"
|
||||
ok_resp = _make_ok_response(html=expected_html)
|
||||
ms = _mock_settings()
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
with patch("playwright.async_api.async_playwright", return_value=pw_cm):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
result = await fetcher.fetch("https://avito.ru/test")
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert fetcher._client is not None
|
||||
fetcher._client.post = AsyncMock(return_value=ok_resp) # type: ignore[method-assign]
|
||||
result = await fetcher.fetch("https://avito.ru/test")
|
||||
|
||||
assert result == html
|
||||
assert result == expected_html
|
||||
assert isinstance(result, str)
|
||||
|
||||
|
||||
# ── recycle: reconnect at threshold ───────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_recycle_triggers_reconnect_at_threshold():
|
||||
"""После recycle_pages страниц выполняется переподключение (_reconnect)."""
|
||||
async def test_fetch_posts_to_correct_endpoint() -> None:
|
||||
"""fetch() делает POST к {browser_http_endpoint}/fetch с {"url": ...}."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
recycle_after = 3
|
||||
pw_cm, _pw, _browser = _make_playwright_stack()
|
||||
ms = _mock_settings(recycle_pages=recycle_after, wait_ms=0)
|
||||
|
||||
reconnect_count = 0
|
||||
original_reconnect = BrowserFetcher._reconnect
|
||||
|
||||
async def counting_reconnect(self: BrowserFetcher) -> None:
|
||||
nonlocal reconnect_count
|
||||
reconnect_count += 1
|
||||
await original_reconnect(self)
|
||||
endpoint = "http://fake-browser:3000"
|
||||
ok_resp = _make_ok_response()
|
||||
ms = _mock_settings(browser_http_endpoint=endpoint)
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
with patch("playwright.async_api.async_playwright", return_value=pw_cm):
|
||||
with patch.object(BrowserFetcher, "_reconnect", counting_reconnect):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert reconnect_count == 0
|
||||
for _ in range(recycle_after):
|
||||
await fetcher.fetch("https://example.com")
|
||||
# Третья страница триггерит recycle
|
||||
assert reconnect_count == 1
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert fetcher._client is not None
|
||||
post_mock = AsyncMock(return_value=ok_resp)
|
||||
fetcher._client.post = post_mock # type: ignore[method-assign]
|
||||
await fetcher.fetch("https://example.com/page")
|
||||
|
||||
post_mock.assert_called_once_with(
|
||||
f"{endpoint}/fetch",
|
||||
json={"url": "https://example.com/page"},
|
||||
)
|
||||
|
||||
|
||||
# ── retry при HTTPError ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_page_counter_resets_after_reconnect():
|
||||
"""После recycle _page_count сбрасывается в 0."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
recycle_after = 2
|
||||
pw_cm, _pw, _browser = _make_playwright_stack()
|
||||
ms = _mock_settings(recycle_pages=recycle_after, wait_ms=0)
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
with patch("playwright.async_api.async_playwright", return_value=pw_cm):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
for _ in range(recycle_after):
|
||||
await fetcher.fetch("https://example.com")
|
||||
# После recycle счётчик сброшен
|
||||
assert fetcher._page_count == 0
|
||||
|
||||
|
||||
# ── crash-recovery ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class _FakeTargetClosedError(Exception):
|
||||
"""Имитирует playwright TargetClosedError (проверка по имени класса)."""
|
||||
|
||||
|
||||
_FakeTargetClosedError.__name__ = "TargetClosedError"
|
||||
_FakeTargetClosedError.__qualname__ = "TargetClosedError"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_retries_once_on_target_closed_error():
|
||||
"""fetch() при TargetClosedError переподключается и повторяет запрос."""
|
||||
async def test_fetch_retries_once_on_http_error() -> None:
|
||||
"""При HTTPError первый вызов → retry → успешный второй вызов."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
html = "<html>retry-ok</html>"
|
||||
|
||||
# Первая страница кидает краш, вторая отвечает нормально
|
||||
crash_page = MagicMock()
|
||||
crash_page.goto = AsyncMock(side_effect=_FakeTargetClosedError("target closed"))
|
||||
crash_page.close = AsyncMock()
|
||||
|
||||
ok_page = _make_mock_page(html=html)
|
||||
|
||||
browser_mock = MagicMock()
|
||||
browser_mock.close = AsyncMock()
|
||||
call_num = 0
|
||||
|
||||
async def new_page_side_effect() -> MagicMock:
|
||||
nonlocal call_num
|
||||
call_num += 1
|
||||
return crash_page if call_num == 1 else ok_page
|
||||
|
||||
browser_mock.new_page = AsyncMock(side_effect=new_page_side_effect)
|
||||
|
||||
firefox_mock = MagicMock()
|
||||
firefox_mock.connect = AsyncMock(return_value=browser_mock)
|
||||
|
||||
pw_mock = MagicMock()
|
||||
pw_mock.firefox = firefox_mock
|
||||
pw_mock.stop = AsyncMock()
|
||||
|
||||
async def _start() -> MagicMock:
|
||||
return pw_mock
|
||||
|
||||
pw_cm = MagicMock()
|
||||
pw_cm.start = _start
|
||||
|
||||
ms = _mock_settings(wait_ms=0)
|
||||
ok_resp = _make_ok_response(html=html)
|
||||
error_resp = _make_error_response(500)
|
||||
ms = _mock_settings()
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
with patch("playwright.async_api.async_playwright", return_value=pw_cm):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
result = await fetcher.fetch("https://avito.ru/page")
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert fetcher._client is not None
|
||||
# Первый вызов кидает HTTPStatusError, второй возвращает OK
|
||||
post_mock = AsyncMock(side_effect=[error_resp, ok_resp])
|
||||
fetcher._client.post = post_mock # type: ignore[method-assign]
|
||||
|
||||
with patch("app.services.scrapers.browser_fetcher.asyncio.sleep", AsyncMock()):
|
||||
result = await fetcher.fetch("https://avito.ru/retry")
|
||||
|
||||
assert result == html
|
||||
assert call_num == 2
|
||||
# connect вызывался дважды: при __aenter__ + при reconnect
|
||||
assert firefox_mock.connect.await_count == 2
|
||||
assert post_mock.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_propagates_non_crash_exception():
|
||||
"""fetch() не подавляет обычные (не-crash) исключения."""
|
||||
async def test_fetch_raises_after_two_http_errors() -> None:
|
||||
"""Если оба вызова кидают HTTPError — исключение пробрасывается."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
page = MagicMock()
|
||||
page.goto = AsyncMock(side_effect=ValueError("unexpected parse error"))
|
||||
page.close = AsyncMock()
|
||||
|
||||
pw_cm, _pw, _browser = _make_playwright_stack(page=page)
|
||||
ms = _mock_settings(wait_ms=0)
|
||||
error_resp_1 = _make_error_response(500)
|
||||
error_resp_2 = _make_error_response(500)
|
||||
ms = _mock_settings()
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
with patch("playwright.async_api.async_playwright", return_value=pw_cm):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
with pytest.raises(ValueError, match="unexpected parse error"):
|
||||
await fetcher.fetch("https://avito.ru/bad")
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert fetcher._client is not None
|
||||
post_mock = AsyncMock(side_effect=[error_resp_1, error_resp_2])
|
||||
fetcher._client.post = post_mock # type: ignore[method-assign]
|
||||
|
||||
with patch("app.services.scrapers.browser_fetcher.asyncio.sleep", AsyncMock()):
|
||||
with pytest.raises(httpx.HTTPStatusError):
|
||||
await fetcher.fetch("https://avito.ru/fail")
|
||||
|
||||
assert post_mock.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_retries_on_connect_error() -> None:
|
||||
"""ConnectError (подкласс TransportError) тоже триггерит retry."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
html = "<html>connect-retry-ok</html>"
|
||||
ok_resp = _make_ok_response(html=html)
|
||||
connect_err = httpx.ConnectError("Connection refused")
|
||||
ms = _mock_settings()
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert fetcher._client is not None
|
||||
post_mock = AsyncMock(side_effect=[connect_err, ok_resp])
|
||||
fetcher._client.post = post_mock # type: ignore[method-assign]
|
||||
|
||||
with patch("app.services.scrapers.browser_fetcher.asyncio.sleep", AsyncMock()):
|
||||
result = await fetcher.fetch("https://avito.ru/connect")
|
||||
|
||||
assert result == html
|
||||
assert post_mock.await_count == 2
|
||||
|
||||
|
||||
# ── non-http exception пробрасывается без перехвата ───────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_propagates_unexpected_exception() -> None:
|
||||
"""Не-HTTP исключение (ValueError и т.п.) не подавляется и не вызывает retry."""
|
||||
from app.services.scrapers.browser_fetcher import BrowserFetcher
|
||||
|
||||
ms = _mock_settings()
|
||||
|
||||
with patch("app.core.config.settings", ms):
|
||||
fetcher = BrowserFetcher()
|
||||
async with fetcher:
|
||||
assert fetcher._client is not None
|
||||
post_mock = AsyncMock(side_effect=ValueError("unexpected"))
|
||||
fetcher._client.post = post_mock # type: ignore[method-assign]
|
||||
|
||||
with pytest.raises(ValueError, match="unexpected"):
|
||||
await fetcher.fetch("https://avito.ru/bad")
|
||||
|
||||
assert post_mock.await_count == 1
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \
|
|||
WORKDIR /app
|
||||
|
||||
# camoufox[geoip] тянет playwright как зависимость
|
||||
RUN pip install --no-cache-dir "camoufox[geoip]"
|
||||
RUN pip install --no-cache-dir "camoufox[geoip]" aiohttp
|
||||
|
||||
COPY server.py ./server.py
|
||||
|
||||
|
|
|
|||
|
|
@ -1,28 +1,49 @@
|
|||
"""server.py — tradein-browser service main process.
|
||||
|
||||
Этот модуль запускается как точка входа контейнера ``tradein-browser``.
|
||||
Он читает конфигурацию из переменных окружения и вызывает
|
||||
``camoufox.server.launch_server`` для экспозиции Playwright WS-эндпоинта,
|
||||
к которому подключается ``tradein-backend`` через ``playwright.firefox.connect``.
|
||||
Он запускает camoufox **локально** внутри контейнера (AsyncCamoufox) и
|
||||
экспонирует простой HTTP API на базе aiohttp:
|
||||
|
||||
GET /health → {"status": "ok"}
|
||||
POST /fetch → {"url": "..."} → {"html": "..."}
|
||||
|
||||
Такой подход выбран потому, что ``camoufox.server.launch_server`` (Playwright
|
||||
WS-сервер) несовместим с современными версиями playwright (1.45–1.60):
|
||||
``browserServerImpl.js`` отсутствует в дистрибутиве → контейнер краш-лупится
|
||||
с MODULE_NOT_FOUND / RuntimeError при запуске через ``launch_server``.
|
||||
Локальный запуск ``AsyncCamoufox`` работает стабильно — проверено.
|
||||
|
||||
Переменные окружения:
|
||||
BROWSER_PORT — TCP-порт WS-сервера (default: 3000)
|
||||
SCRAPER_PROXY_URL — прокси URL формата ``http://user:pass@host:port`` (опционально)
|
||||
BROWSER_PORT — TCP-порт HTTP-сервера (default: 3000)
|
||||
BROWSER_RECYCLE_PAGES — страниц в одном сеансе браузера до перезапуска (default: 15)
|
||||
BROWSER_NAV_TIMEOUT_MS — таймаут page.goto в мс (default: 60000)
|
||||
BROWSER_WAIT_MS — ожидание после DOMContentLoaded в мс (default: 2500)
|
||||
SCRAPER_PROXY_URL — прокси URL формата ``http://user:pass@host:port`` (опционально)
|
||||
|
||||
WS-эндпоинт доступен внутри Docker-сети как
|
||||
``ws://tradein-browser:<BROWSER_PORT>/playwright``.
|
||||
Эндпоинт /fetch доступен из Docker-сети как ``http://tradein-browser:3000/fetch``.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from aiohttp import web
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── конфигурация из env ────────────────────────────────────────────────────────
|
||||
|
||||
BROWSER_PORT: int = int(os.environ.get("BROWSER_PORT", "3000"))
|
||||
BROWSER_RECYCLE_PAGES: int = int(os.environ.get("BROWSER_RECYCLE_PAGES", "15"))
|
||||
BROWSER_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_NAV_TIMEOUT_MS", "60000"))
|
||||
BROWSER_WAIT_MS: int = int(os.environ.get("BROWSER_WAIT_MS", "2500"))
|
||||
SCRAPER_PROXY_URL: str | None = os.environ.get("SCRAPER_PROXY_URL")
|
||||
|
||||
|
||||
def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None:
|
||||
"""Парсит proxy URL → playwright proxy dict.
|
||||
"""Парсит proxy URL → camoufox proxy dict.
|
||||
|
||||
Формат входного URL: ``http://user:pass@host:port`` или ``http://host:port``.
|
||||
Возвращает None если proxy_url пуст.
|
||||
|
|
@ -47,36 +68,190 @@ def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None:
|
|||
return result
|
||||
|
||||
|
||||
def main() -> None:
|
||||
port = int(os.environ.get("BROWSER_PORT", "3000"))
|
||||
proxy_url = os.environ.get("SCRAPER_PROXY_URL")
|
||||
proxy = _parse_proxy(proxy_url)
|
||||
# ── состояние браузера ─────────────────────────────────────────────────────────
|
||||
|
||||
logger.info(
|
||||
"tradein-browser: запуск camoufox WS-сервера на порту %d (proxy=%s)",
|
||||
port,
|
||||
proxy is not None,
|
||||
)
|
||||
_browser: object | None = None # текущий экземпляр Browser
|
||||
_browser_cm: object | None = None # AsyncCamoufox context manager (для __aexit__)
|
||||
_page_counter: int = 0 # страниц с момента последнего (пере)запуска
|
||||
_fetch_lock: asyncio.Lock | None = None # сериализация fetch-запросов
|
||||
|
||||
# Импортируем camoufox.server здесь — в контейнере Firefox уже установлен.
|
||||
from camoufox.server import launch_server # noqa: PLC0415
|
||||
|
||||
async def _launch_browser() -> None:
|
||||
"""Запускает AsyncCamoufox и сохраняет browser + CM в модульных переменных."""
|
||||
global _browser, _browser_cm, _page_counter
|
||||
|
||||
from camoufox.async_api import AsyncCamoufox # noqa: PLC0415
|
||||
|
||||
proxy = _parse_proxy(SCRAPER_PROXY_URL)
|
||||
kwargs: dict[str, object] = {
|
||||
"headless": True,
|
||||
"os": "windows",
|
||||
"locale": "ru-RU",
|
||||
"geoip": True,
|
||||
"humanize": True,
|
||||
"port": port,
|
||||
# port передаётся через **launch_options в launch_options() → прокидывается
|
||||
# в Playwright launchServer({ port: N }) — сервер слушает на заданном порту.
|
||||
}
|
||||
if proxy is not None:
|
||||
kwargs["proxy"] = proxy
|
||||
|
||||
# launch_server блокируется — это основной процесс контейнера.
|
||||
launch_server(**kwargs) # type: ignore[arg-type]
|
||||
logger.info(
|
||||
"tradein-browser: запуск AsyncCamoufox (proxy=%s, recycle_pages=%d)",
|
||||
proxy is not None,
|
||||
BROWSER_RECYCLE_PAGES,
|
||||
)
|
||||
cm = AsyncCamoufox(**kwargs) # type: ignore[arg-type]
|
||||
browser = await cm.__aenter__()
|
||||
_browser_cm = cm
|
||||
_browser = browser
|
||||
_page_counter = 0
|
||||
logger.info("tradein-browser: браузер запущен")
|
||||
|
||||
|
||||
async def _close_browser() -> None:
|
||||
"""Закрывает текущий экземпляр браузера."""
|
||||
global _browser, _browser_cm, _page_counter
|
||||
|
||||
cm = _browser_cm
|
||||
if cm is not None:
|
||||
try:
|
||||
await cm.__aexit__(None, None, None) # type: ignore[attr-defined]
|
||||
logger.info("tradein-browser: браузер закрыт")
|
||||
except Exception as exc:
|
||||
logger.warning("tradein-browser: ошибка при закрытии браузера: %s", exc)
|
||||
finally:
|
||||
_browser = None
|
||||
_browser_cm = None
|
||||
_page_counter = 0
|
||||
|
||||
|
||||
async def _relaunch_browser() -> None:
|
||||
"""Закрывает текущий браузер и запускает новый (recycle / crash-recovery)."""
|
||||
logger.info("tradein-browser: перезапуск браузера")
|
||||
await _close_browser()
|
||||
await _launch_browser()
|
||||
|
||||
|
||||
# ── aiohttp lifecycle hooks ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def _on_startup(app: web.Application) -> None:
|
||||
global _fetch_lock
|
||||
_fetch_lock = asyncio.Lock()
|
||||
await _launch_browser()
|
||||
|
||||
|
||||
async def _on_cleanup(app: web.Application) -> None:
|
||||
await _close_browser()
|
||||
|
||||
|
||||
# ── handlers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def health_handler(request: web.Request) -> web.Response:
|
||||
return web.json_response({"status": "ok"})
|
||||
|
||||
|
||||
async def fetch_handler(request: web.Request) -> web.Response:
|
||||
"""POST /fetch {"url": "..."} → {"html": "..."}
|
||||
|
||||
Выполняет навигацию в браузере и возвращает полный HTML страницы.
|
||||
Запросы сериализованы через lock (один за раз).
|
||||
"""
|
||||
global _page_counter
|
||||
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
return web.json_response({"error": "invalid JSON body"}, status=400)
|
||||
|
||||
url: str | None = body.get("url")
|
||||
if not url:
|
||||
return web.json_response({"error": "missing 'url' field"}, status=400)
|
||||
|
||||
assert _fetch_lock is not None, "_fetch_lock not initialised"
|
||||
|
||||
async with _fetch_lock:
|
||||
try:
|
||||
html = await _do_fetch(url)
|
||||
except Exception as exc:
|
||||
logger.error("tradein-browser: fetch error url=%r: %s: %s", url, type(exc).__name__, exc)
|
||||
return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=500)
|
||||
|
||||
return web.json_response({"html": html})
|
||||
|
||||
|
||||
async def _do_fetch(url: str) -> str:
|
||||
"""Одна попытка навигации; при краше браузера — перезапускает и повторяет один раз."""
|
||||
global _page_counter
|
||||
|
||||
try:
|
||||
return await _fetch_once(url)
|
||||
except Exception as exc:
|
||||
if _is_browser_crash(exc):
|
||||
logger.warning(
|
||||
"tradein-browser: краш браузера (%s), перезапуск + retry: %s",
|
||||
type(exc).__name__,
|
||||
url,
|
||||
)
|
||||
await _relaunch_browser()
|
||||
return await _fetch_once(url)
|
||||
raise
|
||||
|
||||
|
||||
async def _fetch_once(url: str) -> str:
|
||||
"""Открывает страницу, переходит по URL, ждёт JS, возвращает HTML."""
|
||||
global _page_counter
|
||||
|
||||
browser = _browser
|
||||
assert browser is not None, "browser not launched"
|
||||
|
||||
page = await browser.new_page() # type: ignore[attr-defined]
|
||||
try:
|
||||
await page.goto(url, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded") # type: ignore[attr-defined]
|
||||
if BROWSER_WAIT_MS > 0:
|
||||
await page.wait_for_timeout(BROWSER_WAIT_MS) # type: ignore[attr-defined]
|
||||
html: str = await page.content() # type: ignore[attr-defined]
|
||||
finally:
|
||||
await page.close() # type: ignore[attr-defined]
|
||||
|
||||
_page_counter += 1
|
||||
logger.debug("tradein-browser: fetch OK url=%r pages_since_launch=%d", url, _page_counter)
|
||||
|
||||
if _page_counter >= BROWSER_RECYCLE_PAGES:
|
||||
logger.info(
|
||||
"tradein-browser: recycle threshold (%d) достигнут, перезапуск браузера",
|
||||
BROWSER_RECYCLE_PAGES,
|
||||
)
|
||||
await _relaunch_browser()
|
||||
|
||||
return html
|
||||
|
||||
|
||||
def _is_browser_crash(exc: BaseException) -> bool:
|
||||
"""Проверяет является ли исключение признаком краша / разрыва браузера."""
|
||||
cls_name = type(exc).__name__
|
||||
if cls_name in ("TargetClosedError", "BrowserClosedError", "ConnectionClosedError"):
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return (
|
||||
"browser has been closed" in msg
|
||||
or "target closed" in msg
|
||||
or "connection closed" in msg
|
||||
or "browser disconnected" in msg
|
||||
)
|
||||
|
||||
|
||||
# ── entrypoint ─────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_app() -> web.Application:
|
||||
app = web.Application()
|
||||
app.on_startup.append(_on_startup)
|
||||
app.on_cleanup.append(_on_cleanup)
|
||||
app.router.add_get("/health", health_handler)
|
||||
app.router.add_post("/fetch", fetch_handler)
|
||||
return app
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
logger.info("tradein-browser: старт HTTP-сервера на порту %d", BROWSER_PORT)
|
||||
web.run_app(build_app(), host="0.0.0.0", port=BROWSER_PORT)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue