From b9eb478c8fba9a35106d9c7237e95fa5e6e721ed Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 09:01:50 +0300 Subject: [PATCH] feat(browser): per-source proxy pool behind FEATURE_BROWSER_POOL_ENABLED (Phase 1) Route each scraper source (avito/cian/yandex/domclick) to its own camoufox browser+proxy so they no longer wedge each other through a single global egress. Feature-flagged (FEATURE_BROWSER_POOL_ENABLED, default OFF): with the flag off the /fetch and /login paths are byte-for-byte the existing single-browser behavior. When on, /fetch routes by body["source"] to a per-proxy browser via BROWSER_PROXY_MAP (BROWSER_PROXY_AVITO/CIAN/YANDEX/DOMCLICK with legacy fallbacks), each guarded by its own lazy-launched lock. /login stays single-browser in Phase 1. BrowserFetcher gains a source arg (default avito) and sends it in the /fetch body; all scraper callsites pass their source. No docker-compose/.env.runtime changes (Phase 2, owner-gated). --- tradein-mvp/backend/app/api/v1/admin.py | 2 +- .../backend/app/services/scrape_pipeline.py | 4 +- .../backend/app/services/scrapers/avito.py | 2 +- .../app/services/scrapers/browser_fetcher.py | 9 +- .../app/services/scrapers/cian_newbuilding.py | 2 +- .../backend/app/services/scrapers/domclick.py | 2 +- .../services/scrapers/yandex_newbuilding.py | 4 +- .../app/services/scrapers/yandex_realty.py | 2 +- .../app/tasks/avito_detail_backfill.py | 2 +- .../app/tasks/cian_history_backfill.py | 2 +- .../tests/scrapers/test_browser_fetcher.py | 31 +- .../tests/test_browser_pool_routing.py | 240 ++++++++++++++++ .../backend/tests/test_cian_newbuilding.py | 4 +- .../backend/tests/test_scraper_proxy.py | 8 +- tradein-mvp/browser/server.py | 265 +++++++++++++++++- 15 files changed, 556 insertions(+), 23 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_browser_pool_routing.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 3a3f4f88..981c5157 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -341,7 +341,7 @@ async def cian_auto_login( ) try: - async with BrowserFetcher() as fetcher: + async with BrowserFetcher(source="cian") as fetcher: raw_cookies = await fetcher.login( url=settings.cian_login_url, email=email, diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 651b0c52..c8b2531d 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -195,7 +195,7 @@ async def run_avito_pipeline( if browser_mode: browser_fetcher = shared_browser if browser_fetcher is None: - browser_fetcher = BrowserFetcher() + browser_fetcher = BrowserFetcher(source="avito") await browser_fetcher.__aenter__() own_browser = True scraper._browser = browser_fetcher @@ -541,7 +541,7 @@ async def run_avito_city_sweep( session: AsyncSession | None = None shared_bf: BrowserFetcher | None = None if browser_mode: - shared_bf = await stack.enter_async_context(BrowserFetcher()) + shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito")) else: session = await stack.enter_async_context( AsyncSession( diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index 2a1ed86b..d5002826 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/tradein-mvp/backend/app/services/scrapers/avito.py @@ -261,7 +261,7 @@ class AvitoScraper(BaseScraper): "avito: proactive IP rotate at sweep start raised — proceeding", exc_info=True ) if settings.scraper_fetch_mode == "browser": - self._browser = BrowserFetcher() + self._browser = BrowserFetcher(source="avito") await self._browser.__aenter__() logger.info("avito: SERP fetch via BrowserFetcher (camoufox) — #901") return self diff --git a/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py b/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py index e0614534..b175041c 100644 --- a/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py +++ b/tradein-mvp/backend/app/services/scrapers/browser_fetcher.py @@ -40,7 +40,12 @@ class BrowserFetcher: html = await fetcher.fetch("https://example.com") """ - def __init__(self) -> None: + def __init__(self, source: str = "avito") -> None: + # source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер + # роутит /fetch по нему на отдельный браузер+прокси, когда включён + # FEATURE_BROWSER_POOL_ENABLED (Phase 1). При выключенном флаге source + # игнорируется — поведение не меняется. + self._source = source self._client: httpx.AsyncClient | None = None self._endpoint: str | None = None @@ -143,7 +148,7 @@ class BrowserFetcher: resp = await self._client.post( f"{self._endpoint}/fetch", - json={"url": url}, + json={"url": url, "source": self._source}, ) resp.raise_for_status() data: dict[str, str] = resp.json() diff --git a/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py b/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py index 20c9b4cf..4e12f0e0 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py +++ b/tradein-mvp/backend/app/services/scrapers/cian_newbuilding.py @@ -137,7 +137,7 @@ async def fetch_newbuilding( Returns: NewbuildingEnrichment, or None если fetch / parse failed. """ - async with BrowserFetcher() as browser: + async with BrowserFetcher(source="cian") as browser: html = await browser.fetch(zhk_url) # Cian ЖК-карточка: MFE 'newbuilding-card-desktop-frontend', key 'initialState' diff --git a/tradein-mvp/backend/app/services/scrapers/domclick.py b/tradein-mvp/backend/app/services/scrapers/domclick.py index a135e184..ccb71cdb 100644 --- a/tradein-mvp/backend/app/services/scrapers/domclick.py +++ b/tradein-mvp/backend/app/services/scrapers/domclick.py @@ -176,7 +176,7 @@ class DomClickScraper(BaseScraper): # Формируем список room-значений для sweep'ов room_values: list[int | None] = [None] if not rooms else list(rooms) - async with BrowserFetcher() as fetcher: + async with BrowserFetcher(source="domclick") as fetcher: for room_val in room_values: room_label = f"rooms={room_val}" if room_val is not None else "all_rooms" logger.info( diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_newbuilding.py b/tradein-mvp/backend/app/services/scrapers/yandex_newbuilding.py index ff54c385..81b666b0 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_newbuilding.py +++ b/tradein-mvp/backend/app/services/scrapers/yandex_newbuilding.py @@ -161,7 +161,7 @@ class YandexNewbuildingScraper(BaseScraper): url = f"{self.base_url}/{city}/kupit/novostrojka/{jk_slug}-{jk_id}/" try: - async with BrowserFetcher() as fetcher: + async with BrowserFetcher(source="yandex") as fetcher: html = await fetcher.fetch(url) except Exception: logger.exception("yandex nb browser fetch failed: %s", url) @@ -318,7 +318,7 @@ async def resolve_yandex_jk_slug( # вариативности разметки. Прямой realty.yandex.ru SERP — стабильнее. serp_url = f"https://realty.yandex.ru/{city}/kupit/novostrojka/?siteId={jk_id}" try: - async with BrowserFetcher() as fetcher: + async with BrowserFetcher(source="yandex") as fetcher: html = await fetcher.fetch(serp_url) except Exception as exc: logger.warning("resolve_yandex_jk_slug jk_id=%s browser fetch failed: %s", jk_id, exc) diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py index 35004ce2..71282f38 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py +++ b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py @@ -431,7 +431,7 @@ class YandexRealtyScraper(BaseScraper): escalation that raw system-curl over SOCKS5 triggers. Opening per-page is expensive and unnecessary. """ - self._browser = BrowserFetcher() + self._browser = BrowserFetcher(source="yandex") await self._browser.__aenter__() return self diff --git a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py index f60bde42..601e7f3f 100644 --- a/tradein-mvp/backend/app/tasks/avito_detail_backfill.py +++ b/tradein-mvp/backend/app/tasks/avito_detail_backfill.py @@ -101,7 +101,7 @@ async def run_avito_detail_backfill( try: # Setup session (mirrors run_avito_pipeline lines 161-183) if browser_mode: - browser_fetcher = BrowserFetcher() + browser_fetcher = BrowserFetcher(source="avito") await browser_fetcher.__aenter__() own_browser = True scraper._browser = browser_fetcher diff --git a/tradein-mvp/backend/app/tasks/cian_history_backfill.py b/tradein-mvp/backend/app/tasks/cian_history_backfill.py index 41d7a508..c6f8c3d8 100644 --- a/tradein-mvp/backend/app/tasks/cian_history_backfill.py +++ b/tradein-mvp/backend/app/tasks/cian_history_backfill.py @@ -113,7 +113,7 @@ async def backfill_cian_history( else: # One BrowserFetcher instance shared across all listings in this batch. # priceChanges requires JS rendering — curl_cffi returns empty list (#1574). - async with BrowserFetcher() as bf: + async with BrowserFetcher(source="cian") as bf: for row in rows: listing_id: int = row["id"] source_url: str = row["source_url"] diff --git a/tradein-mvp/backend/tests/scrapers/test_browser_fetcher.py b/tradein-mvp/backend/tests/scrapers/test_browser_fetcher.py index 749abeef..2eff8d63 100644 --- a/tradein-mvp/backend/tests/scrapers/test_browser_fetcher.py +++ b/tradein-mvp/backend/tests/scrapers/test_browser_fetcher.py @@ -87,7 +87,11 @@ async def test_fetch_returns_html_from_json_response() -> None: @pytest.mark.asyncio async def test_fetch_posts_to_correct_endpoint() -> None: - """fetch() делает POST к {browser_http_endpoint}/fetch с {"url": ...}.""" + """fetch() делает POST к {browser_http_endpoint}/fetch с {"url", "source"}. + + Дефолтный source="avito" — обратная совместимость со старым поведением, когда + сервер игнорирует source при выключенном FEATURE_BROWSER_POOL_ENABLED. + """ from app.services.scrapers.browser_fetcher import BrowserFetcher endpoint = "http://fake-browser:3000" @@ -104,7 +108,30 @@ async def test_fetch_posts_to_correct_endpoint() -> None: post_mock.assert_called_once_with( f"{endpoint}/fetch", - json={"url": "https://example.com/page"}, + json={"url": "https://example.com/page", "source": "avito"}, + ) + + +@pytest.mark.asyncio +async def test_fetch_posts_source_field_when_set() -> None: + """BrowserFetcher(source="yandex")._post_fetch шлёт body с "source": "yandex".""" + from app.services.scrapers.browser_fetcher import BrowserFetcher + + endpoint = "http://fake-browser:3000" + ok_resp = _make_ok_response() + ms = _mock_settings(browser_http_endpoint=endpoint) + + with patch("app.core.config.settings", ms): + fetcher = BrowserFetcher(source="yandex") + 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://realty.yandex.ru/ekb/") + + post_mock.assert_called_once_with( + f"{endpoint}/fetch", + json={"url": "https://realty.yandex.ru/ekb/", "source": "yandex"}, ) diff --git a/tradein-mvp/backend/tests/test_browser_pool_routing.py b/tradein-mvp/backend/tests/test_browser_pool_routing.py new file mode 100644 index 00000000..8a3b5ce8 --- /dev/null +++ b/tradein-mvp/backend/tests/test_browser_pool_routing.py @@ -0,0 +1,240 @@ +"""test_browser_pool_routing.py — per-source proxy pool routing (Phase 1, #browser-pool). + +Юнит-тесты server.fetch_handler с включённым/выключенным FEATURE_BROWSER_POOL_ENABLED. +camoufox не нужен: launch/_do_fetch_pool замоканы. + +server.py живёт в tradein-mvp/browser/ (отдельный сервис) и импортит `from aiohttp +import web`. aiohttp в backend-venv НЕТ (только в browser-образе), поэтому здесь в +sys.modules инжектится минимальный стаб aiohttp.web ДО загрузки server.py — ровно те +имена, что использует модуль (Application/Request/Response/json_response). Так +routing-логика гоняется прямо в backend pytest-gate без тяжёлой aiohttp-зависимости. + +Проверяем: + - флаг ON → fetch_handler роутит по body["source"] на нужный proxy_url и держит + его lock во время _do_fetch_pool (замоканного); + - неизвестный/отсутствующий source → fallback на avito-прокси; + - launch недоступного прокси → 503; + - флаг OFF → берётся одиночный путь (_fetch_via_single), source игнорируется; + - _build_proxy_map отбрасывает пустые env и применяет legacy-fallback. +""" + +from __future__ import annotations + +import asyncio +import importlib.util +import json +import sys +import types +from pathlib import Path +from typing import Any + +import pytest + +# ── минимальный стаб aiohttp.web (server.py юзает только эти имена) ────────────── + + +class _StubResponse: + """Имитация web.Response от json_response: хранит status + JSON-байты в .body.""" + + def __init__(self, payload: dict[str, Any], status: int = 200) -> None: + self.status = status + self.body = json.dumps(payload).encode() + + +def _json_response(payload: dict[str, Any], status: int = 200) -> _StubResponse: + return _StubResponse(payload, status=status) + + +def _install_aiohttp_stub() -> None: + """Кладёт фейковый aiohttp + aiohttp.web в sys.modules, если настоящего нет.""" + try: + import aiohttp # noqa: F401 + + return # настоящий aiohttp доступен — стаб не нужен + except ModuleNotFoundError: + pass + + web = types.ModuleType("aiohttp.web") + web.Application = type("Application", (), {}) # type: ignore[attr-defined] + web.Request = type("Request", (), {}) # type: ignore[attr-defined] + web.Response = _StubResponse # type: ignore[attr-defined] + web.json_response = _json_response # type: ignore[attr-defined] + + aiohttp_mod = types.ModuleType("aiohttp") + aiohttp_mod.web = web # type: ignore[attr-defined] + sys.modules["aiohttp"] = aiohttp_mod + sys.modules["aiohttp.web"] = web + + +_install_aiohttp_stub() + +# server.py — отдельный сервис вне backend-пакета. Грузим по относительному пути. +_SERVER_PATH = Path(__file__).resolve().parents[2] / "browser" / "server.py" +_spec = importlib.util.spec_from_file_location("tradein_browser_server_pool", _SERVER_PATH) +assert _spec is not None and _spec.loader is not None +server = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(server) + + +class _StubRequest: + """Минимальный stub aiohttp-request: handler дёргает только .json().""" + + def __init__(self, body: dict[str, Any]) -> None: + self._body = body + + async def json(self) -> dict[str, Any]: + return self._body + + +def _json_body(response: Any) -> dict[str, Any]: + return json.loads(response.body.decode()) + + +def _make_fetch_request(body: dict[str, Any]) -> Any: + return _StubRequest(body) + + +# ── флаг ON: routing по source ─────────────────────────────────────────────────── + + +def test_pool_routes_by_source_and_holds_proxy_lock( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ON: source=yandex → fetch на yandex-прокси, держа именно его lock.""" + proxy_map = { + "avito": "http://avito-proxy:1", + "cian": "http://cian-proxy:2", + "yandex": "http://yandex-proxy:3", + } + monkeypatch.setattr(server, "FEATURE_BROWSER_POOL_ENABLED", True) + monkeypatch.setattr(server, "BROWSER_PROXY_MAP", proxy_map) + monkeypatch.setattr(server, "_pool_locks", {}) + monkeypatch.setattr(server, "_pool_locks_guard", asyncio.Lock()) + + seen: dict[str, Any] = {} + + async def _fake_get_or_launch(proxy_url: str) -> object: + seen["launched_proxy"] = proxy_url + return object() # «браузер поднят» + + async def _fake_do_fetch_pool(url: str, proxy_url: str) -> str: + seen["fetch_proxy"] = proxy_url + # lock этого proxy_url должен быть удержан вызывающим (handler). + seen["lock_held"] = server._pool_locks[proxy_url].locked() + return "yandex" + + monkeypatch.setattr(server, "_get_or_launch_pool_browser", _fake_get_or_launch) + monkeypatch.setattr(server, "_do_fetch_pool", _fake_do_fetch_pool) + + request = _make_fetch_request({"url": "https://realty.yandex.ru/x", "source": "yandex"}) + response = asyncio.run(server.fetch_handler(request)) + + assert response.status == 200 + assert _json_body(response)["html"] == "yandex" + assert seen["launched_proxy"] == "http://yandex-proxy:3" + assert seen["fetch_proxy"] == "http://yandex-proxy:3" + assert seen["lock_held"] is True + assert "http://yandex-proxy:3" in server._pool_locks + + +def test_pool_unknown_source_falls_back_to_avito( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ON: domclick без своего прокси → fallback на avito-прокси.""" + proxy_map = {"avito": "http://avito-proxy:1"} # нет domclick + monkeypatch.setattr(server, "FEATURE_BROWSER_POOL_ENABLED", True) + monkeypatch.setattr(server, "BROWSER_PROXY_MAP", proxy_map) + monkeypatch.setattr(server, "_pool_locks", {}) + monkeypatch.setattr(server, "_pool_locks_guard", asyncio.Lock()) + + seen: dict[str, Any] = {} + + async def _fake_get_or_launch(proxy_url: str) -> object: + return object() + + async def _fake_do_fetch_pool(url: str, proxy_url: str) -> str: + seen["fetch_proxy"] = proxy_url + return "fallback" + + monkeypatch.setattr(server, "_get_or_launch_pool_browser", _fake_get_or_launch) + monkeypatch.setattr(server, "_do_fetch_pool", _fake_do_fetch_pool) + + request = _make_fetch_request({"url": "https://domclick.ru/x", "source": "domclick"}) + response = asyncio.run(server.fetch_handler(request)) + + assert response.status == 200 + assert seen["fetch_proxy"] == "http://avito-proxy:1" + + +def test_pool_returns_503_when_browser_unavailable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ON: launch недоступного прокси → None → 503 (та же форма, что одиночный путь).""" + monkeypatch.setattr(server, "FEATURE_BROWSER_POOL_ENABLED", True) + monkeypatch.setattr(server, "BROWSER_PROXY_MAP", {"avito": "http://avito-proxy:1"}) + monkeypatch.setattr(server, "_pool_locks", {}) + monkeypatch.setattr(server, "_pool_locks_guard", asyncio.Lock()) + + async def _fake_get_or_launch(proxy_url: str) -> None: + return None # прокси лежит + + monkeypatch.setattr(server, "_get_or_launch_pool_browser", _fake_get_or_launch) + + request = _make_fetch_request({"url": "https://avito.ru/x", "source": "avito"}) + response = asyncio.run(server.fetch_handler(request)) + + assert response.status == 503 + assert "browser unavailable" in _json_body(response)["error"] + + +# ── флаг OFF: одиночный путь, source игнорируется ──────────────────────────────── + + +def test_flag_off_uses_single_path_and_ignores_source( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """OFF: берётся _fetch_via_single, pool-путь НЕ вызывается, source неважен.""" + monkeypatch.setattr(server, "FEATURE_BROWSER_POOL_ENABLED", False) + + single_called = {"n": 0} + + async def _fake_single(url: str) -> Any: + single_called["n"] += 1 + return server.web.json_response({"html": "single"}) + + async def _fail_pool(url: str, source: str | None) -> Any: + raise AssertionError("pool path не должен вызываться при OFF-флаге") + + monkeypatch.setattr(server, "_fetch_via_single", _fake_single) + monkeypatch.setattr(server, "_fetch_via_pool", _fail_pool) + + request = _make_fetch_request({"url": "https://avito.ru/x", "source": "yandex"}) + response = asyncio.run(server.fetch_handler(request)) + + assert response.status == 200 + assert _json_body(response)["html"] == "single" + assert single_called["n"] == 1 + + +def test_build_proxy_map_drops_empties(monkeypatch: pytest.MonkeyPatch) -> None: + """_build_proxy_map: пустые/None env не попадают в map; legacy-fallback работает.""" + for var in ( + "BROWSER_PROXY_AVITO", + "AVITO_PROXY_URL", + "SCRAPER_PROXY_URL", + "BROWSER_PROXY_CIAN", + "CIAN_PROXY_URL", + "BROWSER_PROXY_YANDEX", + "YANDEX_PROXY_URL", + "BROWSER_PROXY_DOMCLICK", + ): + monkeypatch.delenv(var, raising=False) + + monkeypatch.setenv("AVITO_PROXY_URL", "http://avito:1") # avito через legacy-fallback + monkeypatch.setenv("BROWSER_PROXY_CIAN", "http://cian:2") + # yandex/domclick не заданы → отсутствуют в map. + + result = server._build_proxy_map() + assert result == {"avito": "http://avito:1", "cian": "http://cian:2"} + assert "yandex" not in result + assert "domclick" not in result diff --git a/tradein-mvp/backend/tests/test_cian_newbuilding.py b/tradein-mvp/backend/tests/test_cian_newbuilding.py index 015d71c5..2364ba0b 100644 --- a/tradein-mvp/backend/tests/test_cian_newbuilding.py +++ b/tradein-mvp/backend/tests/test_cian_newbuilding.py @@ -263,7 +263,7 @@ def _make_browser_fetcher_mock(html: str, monkeypatch: pytest.MonkeyPatch) -> No monkeypatch.setattr( "app.services.scrapers.cian_newbuilding.BrowserFetcher", - lambda: mock_fetcher, + lambda *a, **k: mock_fetcher, ) @@ -291,7 +291,7 @@ async def test_fetch_newbuilding_browser_error_propagates(monkeypatch): mock_fetcher.__aexit__ = AsyncMock(return_value=None) monkeypatch.setattr( "app.services.scrapers.cian_newbuilding.BrowserFetcher", - lambda: mock_fetcher, + lambda *a, **k: mock_fetcher, ) with pytest.raises(RuntimeError, match="browser timeout"): diff --git a/tradein-mvp/backend/tests/test_scraper_proxy.py b/tradein-mvp/backend/tests/test_scraper_proxy.py index 123c4260..025e4b39 100644 --- a/tradein-mvp/backend/tests/test_scraper_proxy.py +++ b/tradein-mvp/backend/tests/test_scraper_proxy.py @@ -187,7 +187,9 @@ async def test_yandex_realty_opens_browser_fetcher(monkeypatch): mock_browser.__aenter__ = AsyncMock(return_value=mock_browser) mock_browser.__aexit__ = AsyncMock(return_value=None) - monkeypatch.setattr("app.services.scrapers.yandex_realty.BrowserFetcher", lambda: mock_browser) + monkeypatch.setattr( + "app.services.scrapers.yandex_realty.BrowserFetcher", lambda *a, **k: mock_browser + ) scraper = YandexRealtyScraper() await scraper.__aenter__() @@ -271,7 +273,7 @@ async def test_cian_newbuilding_own_session_receives_proxies(): session_ctor_calls.append(kwargs) return MagicMock() - with patch.object(cian_newbuilding, "BrowserFetcher", lambda: mock_fetcher): + with patch.object(cian_newbuilding, "BrowserFetcher", lambda *a, **k: mock_fetcher): with patch.object(cian_newbuilding, "AsyncSession", _fake_async_session): await cian_newbuilding.fetch_newbuilding(zhk_url) @@ -307,7 +309,7 @@ async def test_cian_newbuilding_shared_session_not_recreated(): sentinel_session = MagicMock(name="external_session") - with patch.object(cian_newbuilding, "BrowserFetcher", lambda: mock_fetcher): + with patch.object(cian_newbuilding, "BrowserFetcher", lambda *a, **k: mock_fetcher): with patch.object(cian_newbuilding, "AsyncSession", _fake_async_session): await cian_newbuilding.fetch_newbuilding(zhk_url, session=sentinel_session) diff --git a/tradein-mvp/browser/server.py b/tradein-mvp/browser/server.py index eda91f0b..a71250df 100644 --- a/tradein-mvp/browser/server.py +++ b/tradein-mvp/browser/server.py @@ -84,6 +84,46 @@ SCRAPER_PROXY_URL: str | None = os.environ.get("AVITO_PROXY_URL") or os.environ. "SCRAPER_PROXY_URL" ) +# ── per-source proxy pool (Phase 1, feature-flagged) ─────────────────────────── +# По умолчанию ВЫКЛЮЧЕН: при FEATURE_BROWSER_POOL_ENABLED!=true поведение байт-в-байт +# идентично текущему (один глобальный браузер на SCRAPER_PROXY_URL). Когда флаг +# включён, /fetch роутится по полю body["source"] на отдельный браузер+прокси из +# BROWSER_PROXY_MAP — источники (avito/cian/yandex/...) больше не клинят друг друга +# через единственный egress-прокси. +# +# Deploy-env (Phase 2, owner-gated — задаётся в .env.runtime): +# FEATURE_BROWSER_POOL_ENABLED — "true" включает per-source pool (default false) +# BROWSER_PROXY_AVITO — прокси для source=avito (fallback AVITO_PROXY_URL/SCRAPER_PROXY_URL) +# BROWSER_PROXY_CIAN — прокси для source=cian (fallback CIAN_PROXY_URL) +# BROWSER_PROXY_YANDEX — прокси для source=yandex (fallback YANDEX_PROXY_URL) +# BROWSER_PROXY_DOMCLICK — прокси для source=domclick (нет fallback → роутится на avito) +FEATURE_BROWSER_POOL_ENABLED: bool = ( + os.environ.get("FEATURE_BROWSER_POOL_ENABLED", "false").lower() == "true" +) + + +def _build_proxy_map() -> dict[str, str]: + """Строит source → proxy_url из env, отбрасывая пустые значения. + + Каждый источник берёт свою явную BROWSER_PROXY_, иначе падает на + legacy-переменную (AVITO_PROXY_URL/CIAN_PROXY_URL/...). Пустые/None НЕ + попадают в map — caller трактует отсутствие как «нет прокси → direct». + """ + raw: dict[str, str | None] = { + "avito": ( + os.environ.get("BROWSER_PROXY_AVITO") + or os.environ.get("AVITO_PROXY_URL") + or os.environ.get("SCRAPER_PROXY_URL") + ), + "cian": os.environ.get("BROWSER_PROXY_CIAN") or os.environ.get("CIAN_PROXY_URL"), + "yandex": os.environ.get("BROWSER_PROXY_YANDEX") or os.environ.get("YANDEX_PROXY_URL"), + "domclick": os.environ.get("BROWSER_PROXY_DOMCLICK"), + } + return {src: url for src, url in raw.items() if url} + + +BROWSER_PROXY_MAP: dict[str, str] = _build_proxy_map() + def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None: """Парсит proxy URL → camoufox proxy dict. @@ -137,6 +177,19 @@ _BROWSER_RETRY_MIN_S: float = 30.0 # стартовый интервал re _BROWSER_RETRY_MAX_S: float = 60.0 # потолок backoff'а _browser_retry_task: asyncio.Task[None] | None = None # handle фоновой retry-задачи +# ── per-source pool state (Phase 1) ────────────────────────────────────────────── +# Параллельные структуры к одиночным глобалам выше, но keyed по proxy_url. Заполняются +# ЛЕНИВО (только при FEATURE_BROWSER_POOL_ENABLED и первом /fetch на данный proxy_url), +# поэтому OFF-путь их вообще не трогает. Один браузер+lock на каждый уникальный прокси. +_browser_pool: dict[str, object | None] = {} # proxy_url → Browser +_browser_cm_pool: dict[str, object | None] = {} # proxy_url → AsyncCamoufox CM +_page_counter_pool: dict[str, int] = {} # proxy_url → страниц с launch'а +_pool_locks: dict[str, asyncio.Lock] = {} # proxy_url → Lock (launch+fetch сериализация) +# Guard на создание per-proxy локов: setdefault на обычном dict из разных корутин +# гонок не даёт (нет await между read-modify-write), но держим явный guard на случай +# будущей сложной инициализации. Создаётся в _on_startup. +_pool_locks_guard: asyncio.Lock | None = None + async def _launch_browser() -> None: """Запускает AsyncCamoufox и сохраняет browser + CM в модульных переменных.""" @@ -313,13 +366,169 @@ def _start_retry_task(app: web.Application) -> None: app["browser_retry_task"] = task +# ── per-source pool helpers (Phase 1) ──────────────────────────────────────────── + + +async def _get_pool_lock(proxy_url: str) -> asyncio.Lock: + """Возвращает (создавая лениво под guard'ом) per-proxy лок для proxy_url. + + Лок сериализует launch+fetch на конкретном браузере: один прокси = один + браузер = N последовательных страниц. Guard защищает создание лока от гонки + параллельных корутин на первом запросе к новому проксе. + """ + existing = _pool_locks.get(proxy_url) + if existing is not None: + return existing + assert _pool_locks_guard is not None, "_pool_locks_guard not initialised" + async with _pool_locks_guard: + return _pool_locks.setdefault(proxy_url, asyncio.Lock()) + + +async def _launch_browser_for_proxy(proxy_url: str) -> None: + """Запускает AsyncCamoufox для конкретного proxy_url и кладёт в *_pool[proxy_url]. + + Зеркало _launch_browser, но параметризовано прокси и пишет в pool-словари + (keyed по proxy_url), а не в одиночные глобалы. Пустой proxy_url → _parse_proxy + вернёт None → direct (совпадает с поведением одиночного браузера без прокси). + """ + from camoufox.async_api import AsyncCamoufox + + proxy = _parse_proxy(proxy_url) + kwargs: dict[str, object] = { + "headless": True, + "os": "windows", + "locale": "ru-RU", + "geoip": True, + "humanize": True, + } + if proxy is not None: + kwargs["proxy"] = proxy + + logger.info( + "tradein-browser[pool]: запуск 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_pool[proxy_url] = cm + _browser_pool[proxy_url] = browser + _page_counter_pool[proxy_url] = 0 + logger.info("tradein-browser[pool]: браузер запущен (proxy=%s)", proxy is not None) + + +async def _close_pool_browser(proxy_url: str) -> None: + """Закрывает per-proxy браузер и чистит pool-состояние для proxy_url.""" + cm = _browser_cm_pool.get(proxy_url) + if cm is not None: + try: + await cm.__aexit__(None, None, None) # type: ignore[attr-defined] + logger.info("tradein-browser[pool]: браузер закрыт") + except Exception as exc: + logger.warning( + "tradein-browser[pool]: ошибка при закрытии браузера: %s", type(exc).__name__ + ) + _browser_pool[proxy_url] = None + _browser_cm_pool[proxy_url] = None + _page_counter_pool[proxy_url] = 0 + + +async def _relaunch_pool_browser(proxy_url: str) -> None: + """Закрывает и заново запускает per-proxy браузер (recycle / crash-recovery).""" + logger.info("tradein-browser[pool]: перезапуск браузера") + await _close_pool_browser(proxy_url) + await _launch_browser_for_proxy(proxy_url) + + +async def _get_or_launch_pool_browser(proxy_url: str) -> object | None: + """Возвращает поднятый per-proxy браузер, лениво запуская при необходимости. + + Caller держит соответствующий _pool_locks[proxy_url], поэтому здесь нет своей + сериализации. При фейле launch'а (прокси лежит) лог санитизирован (только + type(exc).__name__, без str(exc) — camoufox светит user:pass@host в тексте) и + возвращается None → caller отдаёт 503 (как одиночный путь при недоступном прокси). + """ + existing = _browser_pool.get(proxy_url) + if existing is not None: + return existing + try: + await _launch_browser_for_proxy(proxy_url) + except Exception as exc: + logger.warning( + "tradein-browser[pool]: browser launch failed: proxy unreachable " + "(exc_type=%s) — отдаём 503, retry на следующем запросе", + type(exc).__name__, + ) + await _close_pool_browser(proxy_url) + return None + return _browser_pool.get(proxy_url) + + +async def _do_fetch_pool(url: str, proxy_url: str) -> str: + """Pool-вариант _do_fetch: при краше браузера — relaunch и один retry. + + Caller держит _pool_locks[proxy_url] (launch+fetch сериализованы на этот прокси), + поэтому relaunch безопасен — нет параллельных страниц на этом браузере. + """ + try: + return await _fetch_once_pool(url, proxy_url) + except Exception as exc: + if _is_browser_crash(exc): + logger.warning( + "tradein-browser[pool]: краш браузера (%s), перезапуск + retry: %s", + type(exc).__name__, + url, + ) + await _relaunch_pool_browser(proxy_url) + browser = _browser_pool.get(proxy_url) + if browser is None: + raise + return await _fetch_once_pool(url, proxy_url) + raise + + +async def _fetch_once_pool(url: str, proxy_url: str) -> str: + """Открывает страницу на per-proxy браузере, грузит URL, возвращает HTML. + + Caller держит _pool_locks[proxy_url], поэтому страницы на этом браузере не + параллелятся — recycle через _relaunch_pool_browser безопасен прямо здесь. + """ + browser = _browser_pool.get(proxy_url) + assert browser is not None, "pool 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_pool[proxy_url] = _page_counter_pool.get(proxy_url, 0) + 1 + if _page_counter_pool[proxy_url] >= BROWSER_RECYCLE_PAGES: + logger.info( + "tradein-browser[pool]: recycle threshold (%d) достигнут, перезапуск браузера", + BROWSER_RECYCLE_PAGES, + ) + await _relaunch_pool_browser(proxy_url) + + return html + + # ── aiohttp lifecycle hooks ──────────────────────────────────────────────────── async def _on_startup(app: web.Application) -> None: - global _launch_lock, _fetch_sem + global _launch_lock, _fetch_sem, _pool_locks_guard _launch_lock = asyncio.Lock() _fetch_sem = asyncio.Semaphore(BROWSER_CONCURRENCY) + _pool_locks_guard = asyncio.Lock() + if FEATURE_BROWSER_POOL_ENABLED: + logger.info( + "tradein-browser: per-source proxy pool ВКЛЮЧЁН (sources=%s)", + sorted(BROWSER_PROXY_MAP.keys()), + ) logger.info( "tradein-browser: BROWSER_CONCURRENCY=%d (параллельных /fetch)", BROWSER_CONCURRENCY ) @@ -350,6 +559,10 @@ async def _on_cleanup(app: web.Application) -> None: ) _browser_retry_task = None await _close_browser() + # Закрыть все per-source pool браузеры (если pool использовался). + for proxy_url in list(_browser_pool.keys()): + if _browser_pool.get(proxy_url) is not None: + await _close_pool_browser(proxy_url) # ── handlers ─────────────────────────────────────────────────────────────────── @@ -371,8 +584,6 @@ async def fetch_handler(request: web.Request) -> web.Response: До BROWSER_CONCURRENCY запросов обрабатываются параллельно (семафор), у каждого — своя страница (изоляция крашей). """ - global _inflight - try: body = await request.json() except Exception: @@ -382,6 +593,16 @@ async def fetch_handler(request: web.Request) -> web.Response: if not url: return web.json_response({"error": "missing 'url' field"}, status=400) + if FEATURE_BROWSER_POOL_ENABLED: + source: str | None = body.get("source") + return await _fetch_via_pool(url, source) + return await _fetch_via_single(url) + + +async def _fetch_via_single(url: str) -> web.Response: + """Текущий (OFF-флаг) путь: один глобальный браузер + семафор. Без изменений.""" + global _inflight + assert _fetch_sem is not None, "_fetch_sem not initialised" async with _fetch_sem: @@ -409,6 +630,44 @@ async def fetch_handler(request: web.Request) -> web.Response: return web.json_response({"html": html}) +async def _fetch_via_pool(url: str, source: str | None) -> web.Response: + """Per-source путь (ON-флаг): роутит на отдельный браузер+прокси по source. + + Выбор прокси: BROWSER_PROXY_MAP[source] → fallback на avito-прокси → "" (direct, + как одиночный путь при пустом env). Один lock на proxy_url сериализует launch+fetch + на этот браузер — источники с разными проксями работают независимо, не клиня друг + друга. Lazy-launch на первом запросе к данному проксе; при недоступном прокси → 503. + """ + proxy_url = BROWSER_PROXY_MAP.get(source or "avito") or BROWSER_PROXY_MAP.get("avito") + if not proxy_url: + # Ни одного прокси не сконфигурировано → direct (как одиночный путь при пустом env). + proxy_url = "" + + lock = await _get_pool_lock(proxy_url) + async with lock: + browser = await _get_or_launch_pool_browser(proxy_url) + if browser is None: + logger.warning( + "tradein-browser[pool]: /fetch 503 — браузер недоступен (proxy may be down)" + ) + return web.json_response( + {"error": "browser unavailable (proxy may be down)"}, status=503 + ) + try: + html = await _do_fetch_pool(url, proxy_url) + except Exception as exc: + logger.error( + "tradein-browser[pool]: fetch error url=%r source=%r: %s: %s", + url, + source, + 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: """Одна попытка навигации; при краше браузера — перезапускает и повторяет один раз.