diff --git a/tradein-mvp/backend/app/scheduler_main.py b/tradein-mvp/backend/app/scheduler_main.py index a0cf5571..a9f2e58c 100644 --- a/tradein-mvp/backend/app/scheduler_main.py +++ b/tradein-mvp/backend/app/scheduler_main.py @@ -136,10 +136,18 @@ async def _run_kit_scheduler() -> None: from app.services.scraper_adapters import ( RealEnrichmentJobs, RealMatcherAdapter, + RealProxyProvider, RealScraperConfig, RealSessionFactory, ) + # Proxy-пул (#2163 curl / #2164 P4 browser) инжектируется ТОЛЬКО когда включён любой + # из флагов. Оба дефолтно False (ship-dark) → proxy_provider=None → curl/browser берут + # прокси из env, прод не меняется. Один RealProxyProvider обслуживает оба пути (curl + # через curl_proxy_url, browser через BrowserFetcher → POST /fetch{proxy}). + use_pool = settings.use_proxy_pool_curl or settings.use_proxy_pool_browser + proxy_provider = RealProxyProvider() if use_pool else None + ctx = SchedulerContext( config=RealScraperConfig(), matcher=RealMatcherAdapter(), @@ -147,6 +155,7 @@ async def _run_kit_scheduler() -> None: session_factory=RealSessionFactory(), shutdown_requested=shutdown_requested, runs=runs_mod, + proxy_provider=proxy_provider, ) registry = build_registry(product_handlers=build_product_handlers(ctx)) logger.info("scheduler_main: kit-scheduler registry built (%d source handlers)", len(registry)) diff --git a/tradein-mvp/backend/app/services/scraper_adapters.py b/tradein-mvp/backend/app/services/scraper_adapters.py index c45c2f5b..32fc9755 100644 --- a/tradein-mvp/backend/app/services/scraper_adapters.py +++ b/tradein-mvp/backend/app/services/scraper_adapters.py @@ -210,6 +210,11 @@ class RealScraperConfig: def use_proxy_pool_curl(self) -> bool: return _settings.use_proxy_pool_curl + # ── Proxy-pool browser (#2164 P4) ───────────────────────────────────── + @property + def use_proxy_pool_browser(self) -> bool: + return _settings.use_proxy_pool_browser + class RealProxyProvider: """ProxyProvider-адаптер над `app.services.proxy_pool` (#2163). diff --git a/tradein-mvp/backend/tests/test_kit_browser_fetcher_proxy_pool.py b/tradein-mvp/backend/tests/test_kit_browser_fetcher_proxy_pool.py new file mode 100644 index 00000000..cbf7ea4b --- /dev/null +++ b/tradein-mvp/backend/tests/test_kit_browser_fetcher_proxy_pool.py @@ -0,0 +1,167 @@ +"""Тесты browser-пула для kit `scraper_kit.browser_fetcher.BrowserFetcher` (#2164 P4). + +Инвариант ship-dark + fallback: + - use_pool=False (дефолт) → в теле POST /fetch НЕТ поля "proxy"; proxy_provider не + трогается (golden-parity: браузер юзает свой env-прокси BROWSER_PROXY_*). + - use_pool=True + пул выдал lease → тело содержит "proxy"=lease.url + "proxy_kind"; + на выходе mark_health(ok) + release (в finally — lease не течёт). + - use_pool=True + пул пуст (acquire→None) → тела без "proxy", НЕ падаем. + - use_pool=True + acquire бросил → fallback без "proxy", НЕ падаем. + - fetch кинул → mark_health(ok=False) + release всё равно (finally). + +httpx полностью замокан: fetcher._client подменяется MagicMock'ом. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from scraper_kit.browser_fetcher import BrowserFetcher +from scraper_kit.contracts import ProxyLease + + +def _mock_client(json_payload: dict[str, Any], *, raise_exc: Exception | None = None) -> MagicMock: + """httpx.AsyncClient-заглушка: .post → resp c raise_for_status/json.""" + resp = MagicMock() + if raise_exc is not None: + resp.raise_for_status.side_effect = raise_exc + else: + resp.raise_for_status.return_value = None + resp.json.return_value = json_payload + client = MagicMock() + client.post = AsyncMock(return_value=resp) + return client + + +class _FakeProxyProvider: + """ProxyProvider-заглушка: acquire отдаёт заданный lease (или None), считает вызовы.""" + + def __init__(self, lease: ProxyLease | None, *, acquire_raises: bool = False) -> None: + self._lease = lease + self._acquire_raises = acquire_raises + self.acquired: list[str] = [] + self.released: list[int] = [] + self.health: list[tuple[int, bool]] = [] + + def acquire(self, provider: str) -> ProxyLease | None: + self.acquired.append(provider) + if self._acquire_raises: + raise RuntimeError("pool boom") + return self._lease + + def release(self, lease: ProxyLease) -> None: + self.released.append(lease.id) + + def mark_health(self, lease: ProxyLease, ok: bool, **_: Any) -> None: + self.health.append((lease.id, ok)) + + +def _fetcher(client: MagicMock, **kwargs: Any) -> BrowserFetcher: + bf = BrowserFetcher(endpoint="http://browser:3000", **kwargs) + bf._client = client + return bf + + +async def test_fetch_pool_off_no_proxy_in_body() -> None: + """use_pool=False (дефолт) → тело без 'proxy', provider не трогается (parity).""" + provider = _FakeProxyProvider(ProxyLease(id=1, url="http://pool:8080", kind="http")) + client = _mock_client({"html": ""}) + bf = _fetcher(client, source="avito", proxy_provider=provider) # use_pool default False + + html = await bf.fetch("https://avito.ru/x") + + assert html == "" + body = client.post.call_args.kwargs["json"] + assert "proxy" not in body + assert provider.acquired == [] # пул не трогали + + +async def test_fetch_pool_on_injects_proxy_and_releases() -> None: + """use_pool=True + lease → 'proxy'/'proxy_kind' в теле; mark_health(True)+release.""" + lease = ProxyLease(id=7, url="http://u:p@pool:8080", kind="http") + provider = _FakeProxyProvider(lease) + client = _mock_client({"html": ""}) + bf = _fetcher(client, source="avito", proxy_provider=provider, use_pool=True) + + await bf.fetch("https://avito.ru/x") + + body = client.post.call_args.kwargs["json"] + assert body["proxy"] == "http://u:p@pool:8080" + assert body["proxy_kind"] == "http" + assert provider.acquired == ["avito"] + assert provider.health == [(7, True)] + assert provider.released == [7] + + +async def test_fetch_pool_empty_falls_back_no_proxy() -> None: + """use_pool=True + пул пуст (acquire→None) → тело без 'proxy', без mark_health/release.""" + provider = _FakeProxyProvider(None) + client = _mock_client({"html": ""}) + bf = _fetcher(client, source="cian", proxy_provider=provider, use_pool=True) + + await bf.fetch("https://cian.ru/x") + + body = client.post.call_args.kwargs["json"] + assert "proxy" not in body + assert provider.acquired == ["cian"] + assert provider.released == [] + assert provider.health == [] + + +async def test_fetch_pool_acquire_error_falls_back() -> None: + """acquire бросил → fallback без 'proxy', сбор НЕ падает.""" + provider = _FakeProxyProvider(None, acquire_raises=True) + client = _mock_client({"html": ""}) + bf = _fetcher(client, source="yandex", proxy_provider=provider, use_pool=True) + + html = await bf.fetch("https://yandex.ru/x") + + assert html == "" + body = client.post.call_args.kwargs["json"] + assert "proxy" not in body + assert provider.released == [] + + +async def test_fetch_error_marks_health_false_and_releases() -> None: + """fetch кинул → mark_health(ok=False) + release всё равно (finally, lease не течёт).""" + lease = ProxyLease(id=3, url="http://pool:8080", kind="http") + provider = _FakeProxyProvider(lease) + # raise_for_status кидает НЕ-httpx ошибку → fetch() не ретраит, пробрасывает наверх. + client = _mock_client({"html": "x"}, raise_exc=ValueError("bad status")) + bf = _fetcher(client, source="avito", proxy_provider=provider, use_pool=True) + + with pytest.raises(ValueError): + await bf.fetch("https://avito.ru/x") + + assert provider.health == [(3, False)] + assert provider.released == [3] + + +async def test_fetch_json_pool_on_injects_proxy() -> None: + """fetch_json тоже прокидывает proxy из пула в тело /fetch-json.""" + lease = ProxyLease(id=9, url="http://pool:8080", kind="http") + provider = _FakeProxyProvider(lease) + client = _mock_client({"status": 200, "body": "{}"}) + bf = _fetcher(client, source="avito", proxy_provider=provider, use_pool=True) + + await bf.fetch_json( + "https://avito.ru/api", method="POST", body="{}", origin="https://avito.ru/" + ) + + body = client.post.call_args.kwargs["json"] + assert body["proxy"] == "http://pool:8080" + assert body["proxy_kind"] == "http" + assert provider.released == [9] + + +async def test_fetch_json_pool_off_no_proxy() -> None: + """use_pool=False → fetch_json без 'proxy' в теле (parity).""" + client = _mock_client({"status": 200, "body": "{}"}) + bf = _fetcher(client, source="avito") + + await bf.fetch_json("https://avito.ru/api") + + body = client.post.call_args.kwargs["json"] + assert "proxy" not in body diff --git a/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py b/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py index 3a8a1585..105f9e78 100644 --- a/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py +++ b/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py @@ -68,6 +68,8 @@ def test_scraper_config_satisfies_protocol() -> None: assert isinstance(config.avito_serp_ekb_only, bool) # #2163: новый флаг proxy-пула присутствует и bool. assert isinstance(config.use_proxy_pool_curl, bool) + # #2164 P4: флаг browser-пула присутствует и bool. + assert isinstance(config.use_proxy_pool_browser, bool) def test_proxy_provider_satisfies_protocol() -> None: diff --git a/tradein-mvp/browser/server.py b/tradein-mvp/browser/server.py index 376b3c7c..7716f607 100644 --- a/tradein-mvp/browser/server.py +++ b/tradein-mvp/browser/server.py @@ -266,6 +266,11 @@ _page_counters: dict[str, int] = {} # provider → страниц с мом _locks: dict[str, asyncio.Lock] = {} # provider → Lock (весь доступ к инстансу) _retry_tasks: dict[str, asyncio.Task[None]] = {} # provider → фоновая retry-задача _last_goto_at: dict[str, float] = {} # provider → loop-time последнего page.goto (пейсинг) +# #2164 P4: proxy-url, с которым СЕЙЧАС запущен инстанс провайдера (env или динамический +# из пула, переданный в теле /fetch). Нужен для политики «relaunch ТОЛЬКО при реальной +# смене прокси» — camoufox берёт proxy на launch, релонч дорогой, поэтому не релончим, +# если запрошенный proxy совпадает с текущим. None = запущен без прокси (прямое). +_launched_proxy: dict[str, str | None] = {} # Guard на ленивое создание per-provider локов: setdefault на обычном dict из # разных корутин гонок не даёт (нет await между read-modify-write), но держим @@ -297,15 +302,21 @@ async def _lock_for(provider: str) -> asyncio.Lock: return _locks.setdefault(provider, asyncio.Lock()) -async def _launch_browser(provider: str) -> None: +async def _launch_browser(provider: str, proxy_override: str | None = None) -> None: """Запускает AsyncCamoufox для поставщика и кладёт в *_…[provider]. - Использует _provider_proxy(provider) + geoip=True (фингерпринт привязан к гео - IP конкретного прокси — поэтому отдельный инстанс на каждого поставщика). + proxy_override (#2164 P4): если задан — запускаем camoufox с ЭТИМ прокси (динамический + из пула, переданный scraper-стороной в теле /fetch). None → env-прокси + _provider_proxy(provider) как раньше. geoip=True (фингерпринт привязан к гео IP + конкретного прокси — поэтому отдельный инстанс на каждого поставщика). + + Эффективный proxy-url сохраняется в _launched_proxy[provider] для политики + «relaunch только при реальной смене» (см. _ensure_browser). """ from camoufox.async_api import AsyncCamoufox - proxy = _parse_proxy(_provider_proxy(provider)) + proxy_url = proxy_override if proxy_override is not None else _provider_proxy(provider) + proxy = _parse_proxy(proxy_url) kwargs: dict[str, object] = { "headless": True, "os": "windows", @@ -336,7 +347,12 @@ async def _launch_browser(provider: str) -> None: _browser_cms[provider] = cm _browsers[provider] = browser _page_counters[provider] = 0 - logger.info("tradein-browser[%s]: браузер запущен", provider) + _launched_proxy[provider] = proxy_url + logger.info( + "tradein-browser[%s]: браузер запущен (proxy_override=%s)", + provider, + proxy_override is not None, + ) async def _close_browser(provider: str) -> None: @@ -356,20 +372,26 @@ async def _close_browser(provider: str) -> None: _browser_cms.pop(provider, None) _page_counters[provider] = 0 _last_goto_at.pop(provider, None) + _launched_proxy.pop(provider, None) -async def _relaunch_browser(provider: str) -> None: +async def _relaunch_browser(provider: str, proxy_override: str | None = None) -> None: """Закрывает и заново запускает per-provider браузер (recycle / crash-recovery). - Caller держит _locks[provider], поэтому параллельных страниц на этом инстансе - нет — teardown безопасен. + proxy_override (#2164 P4): прокинуть конкретный прокси в новый инстанс. None → + СОХРАНИТЬ текущий прокси инстанса (_launched_proxy) — recycle/crash-recovery не должны + терять динамический прокси пула и не должны его менять (это НЕ смена прокси). Если + инстанс ещё не запускался (_launched_proxy пуст) — None → env-прокси в _launch_browser. + Caller держит _locks[provider], поэтому параллельных страниц на этом инстансе нет — + teardown безопасен. """ + effective = proxy_override if proxy_override is not None else _launched_proxy.get(provider) logger.info("tradein-browser[%s]: перезапуск браузера", provider) await _close_browser(provider) - await _launch_browser(provider) + await _launch_browser(provider, proxy_override=effective) -async def _try_launch_browser(provider: str) -> bool: +async def _try_launch_browser(provider: str, proxy_override: str | None = None) -> bool: """Одна попытка launch'а поставщика с санитизированным логом фейла. НЕ пробрасывает исключение наружу — возвращает True/False. Критично для @@ -382,7 +404,7 @@ async def _try_launch_browser(provider: str) -> bool: через _close_browser, чтобы следующий retry стартовал с чистого листа. """ try: - await _launch_browser(provider) + await _launch_browser(provider, proxy_override=proxy_override) return True except Exception as exc: logger.warning( @@ -395,25 +417,41 @@ async def _try_launch_browser(provider: str) -> bool: return False -async def _ensure_browser(provider: str) -> bool: - """Гарантирует поднятый инстанс поставщика; одна lazy-попытка launch'а. +async def _ensure_browser(provider: str, proxy_override: str | None = None) -> bool: + """Гарантирует поднятый инстанс поставщика с нужным прокси; lazy launch / relaunch. - Caller держит _locks[provider] (весь доступ к инстансу сериализован), поэтому - здесь своей сериализации нет — двух одновременных launch'ей на один provider - быть не может. + proxy_override (#2164 P4): желаемый динамический прокси из пула (тело /fetch). None → + желаемый прокси = env (_provider_proxy). Политика «relaunch ТОЛЬКО при реальной смене»: + - инстанса нет → launch с желаемым прокси; + - инстанс есть, но _launched_proxy[provider] != желаемого → close + relaunch; + - инстанс есть и прокси совпадает → no-op (НЕ релончим — camoufox дорого поднимать). + При выключенном флаге на scraper-стороне proxy_override всегда None → желаемый=env= + текущий → релонча нет, поведение идентично прежнему (golden-parity). + + Caller держит _locks[provider] (весь доступ к инстансу сериализован), поэтому здесь + своей сериализации нет — двух одновременных launch'ей на один provider быть не может. Returns: True если браузер доступен (был или поднялся), False если launch не удался (прокси лежит) — caller отдаёт 503. - При неудачном lazy-launch (прокси лёг) запускает фоновую retry-задачу для - ЭТОГО поставщика: при eager-startup её ставил _on_startup, в lazy-режиме — - первый обратившийся /fetch. Поставщики независимы — фоновый retry поднимается - только для тех, к кому был запрос, а не для всех сразу. + При неудачном launch (прокси лёг) запускает фоновую retry-задачу для ЭТОГО + поставщика (env-прокси). Поставщики независимы — фоновый retry поднимается только + для тех, к кому был запрос, а не для всех сразу. """ + desired_proxy = proxy_override if proxy_override is not None else _provider_proxy(provider) if _browsers.get(provider) is not None: - return True - await _try_launch_browser(provider) + if _launched_proxy.get(provider) == desired_proxy: + return True + # Реальная смена прокси → пересобрать инстанс с новым прокси. + logger.info( + "tradein-browser[%s]: прокси изменился (override=%s) — relaunch перед fetch", + provider, + proxy_override is not None, + ) + await _close_browser(provider) + + await _try_launch_browser(provider, proxy_override=proxy_override) if _browsers.get(provider) is not None: return True _start_retry_task(provider) @@ -520,6 +558,32 @@ def _resolve_provider(body: dict, url: str) -> str: return _provider_for_url(url) +def _resolve_proxy_override(body: dict, provider: str) -> str | None: + """Динамический прокси из тела /fetch (#2164 P4). None → env-прокси (текущее поведение). + + Scraper-сторона (BrowserFetcher за флагом use_proxy_pool_browser) кладёт в тело + ``{"proxy": "http://user:pass@host:port", "proxy_kind": "http"}`` — прокси, взятый из + пула через acquire(source). Если поля нет / пусто → None → инстанс поставщика берёт + свой env-прокси (BROWSER_PROXY_*), поведение не меняется (ship-dark parity). + + camoufox НЕ умеет socks5 с авторизацией → принимаем ТОЛЬКО http/https-прокси; иной + proxy_kind игнорируем (→ None, env-fallback) с предупреждением. Отсутствие proxy_kind + трактуем как http (в пуле kind=http). + """ + raw = body.get("proxy") + if not isinstance(raw, str) or not raw: + return None + kind = body.get("proxy_kind") + if isinstance(kind, str) and kind and kind.lower() not in ("http", "https"): + logger.warning( + "tradein-browser[%s]: proxy_kind=%r не поддерживается camoufox — env-прокси", + provider, + kind, + ) + return None + return raw + + async def health_handler(request: web.Request) -> web.Response: # Всегда 200 — иначе compose healthcheck (если появится) убьёт контейнер, # хотя процесс жив и ретраит браузеры. browsers — per-provider readiness для @@ -547,14 +611,16 @@ async def fetch_handler(request: web.Request) -> web.Response: return web.json_response({"error": "missing 'url' field"}, status=400) provider = _resolve_provider(body, url) + proxy_override = _resolve_proxy_override(body, provider) lock = await _lock_for(provider) async with lock: # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка — # вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/браузера: если # всё ещё None → 503 (caller словит как HTTPStatusError → failed fetch, не - # crash). - if not await _ensure_browser(provider): + # crash). proxy_override (#2164 P4): relaunch camoufox с прокси из пула, если + # он отличается от текущего launch-прокси инстанса (иначе no-op). + if not await _ensure_browser(provider, proxy_override=proxy_override): logger.warning( "tradein-browser[%s]: /fetch 503 — браузер недоступен (proxy may be down)", provider, @@ -611,12 +677,14 @@ async def fetch_json_handler(request: web.Request) -> web.Response: origin = f"{parsed.scheme}://{parsed.netloc}/" provider = _resolve_provider(body, url) + proxy_override = _resolve_proxy_override(body, provider) lock = await _lock_for(provider) async with lock: # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал). - # Lazy-попытка, иначе 503 — без прокси/браузера не фетчим. - if not await _ensure_browser(provider): + # Lazy-попытка, иначе 503 — без прокси/браузера не фетчим. proxy_override + # (#2164 P4): relaunch с прокси из пула при реальной смене. + if not await _ensure_browser(provider, proxy_override=proxy_override): logger.warning( "tradein-browser[%s]: /fetch-json 503 — браузер недоступен (proxy may be down)", provider, @@ -705,7 +773,8 @@ async def _do_fetch(provider: str, url: str) -> str: """Одна попытка навигации; при краше браузера — relaunch и один retry. Caller держит _locks[provider] (нет параллельных страниц на этом инстансе), - поэтому relaunch безопасен — никого не рвёт из-под живых страниц. + поэтому relaunch безопасен. Crash-relaunch сохраняет текущий прокси инстанса + (_relaunch_browser reuse _launched_proxy) — динамический прокси пула не теряется. """ try: return await _fetch_once(provider, url) @@ -775,7 +844,8 @@ async def _do_fetch_json( """Одна попытка in-page fetch'а; при краше браузера — relaunch и один retry. Зеркалит _do_fetch: caller держит _locks[provider] (нет параллельных страниц на - этом инстансе), поэтому relaunch безопасен — никого не рвёт из-под живых страниц. + этом инстансе), поэтому relaunch безопасен. Crash-relaunch сохраняет текущий прокси + инстанса (_relaunch_browser reuse _launched_proxy). """ try: return await _fetch_json_once( diff --git a/tradein-mvp/browser/test_server.py b/tradein-mvp/browser/test_server.py index e8405774..474117f9 100644 --- a/tradein-mvp/browser/test_server.py +++ b/tradein-mvp/browser/test_server.py @@ -219,7 +219,7 @@ def test_same_provider_fetches_are_serialized(monkeypatch: pytest.MonkeyPatch) - """Два /fetch к ОДНОМУ поставщику сериализованы (один лок) — peak==1.""" probe = _OverlapProbe() - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: server._browsers[provider] = object() return True @@ -241,7 +241,7 @@ def test_different_providers_fetch_concurrently(monkeypatch: pytest.MonkeyPatch) """Два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи) — global peak>1.""" probe = _OverlapProbe() - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: server._browsers[provider] = object() return True @@ -271,7 +271,7 @@ def test_mixed_load_serializes_within_parallelizes_across( """Смешанная нагрузка: avito×2 сериализованы, но параллельны с cian×1.""" probe = _OverlapProbe() - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: server._browsers[provider] = object() return True diff --git a/tradein-mvp/browser/test_server_fetch_json.py b/tradein-mvp/browser/test_server_fetch_json.py index ab74fe4a..0af86622 100644 --- a/tradein-mvp/browser/test_server_fetch_json.py +++ b/tradein-mvp/browser/test_server_fetch_json.py @@ -109,7 +109,7 @@ def test_fetch_json_returns_evaluate_result(monkeypatch: pytest.MonkeyPatch) -> server._browsers["avito"] = _FakeBrowser(page) monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -137,7 +137,7 @@ def test_fetch_json_navigates_to_origin_not_url(monkeypatch: pytest.MonkeyPatch) server._browsers["avito"] = _FakeBrowser(page) monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -157,7 +157,7 @@ def test_fetch_json_uses_explicit_origin(monkeypatch: pytest.MonkeyPatch) -> Non server._browsers["avito"] = _FakeBrowser(page) monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -182,7 +182,7 @@ def test_fetch_json_passes_post_method_and_body(monkeypatch: pytest.MonkeyPatch) server._browsers["avito"] = _FakeBrowser(page) monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -216,7 +216,7 @@ def test_fetch_json_settle_uses_configured_ms(monkeypatch: pytest.MonkeyPatch) - monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) monkeypatch.setattr(server, "FETCH_JSON_SETTLE_MS", 1777) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -237,7 +237,7 @@ def test_fetch_json_passes_inpage_retry_params(monkeypatch: pytest.MonkeyPatch) monkeypatch.setattr(server, "FETCH_JSON_INPAGE_RETRIES", 2) monkeypatch.setattr(server, "FETCH_JSON_RETRY_DELAY_MS", 654) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) @@ -278,7 +278,7 @@ def test_fetch_json_400_on_invalid_json() -> None: def test_fetch_json_503_when_browser_unavailable(monkeypatch: pytest.MonkeyPatch) -> None: """/fetch-json отдаёт 503 (не 500/краш) когда инстанс поставщика не поднят.""" - async def _no_ensure(provider: str) -> bool: + async def _no_ensure(provider: str, proxy_override: str | None = None) -> bool: return False # прокси всё ещё недоступен → браузер остаётся None monkeypatch.setattr(server, "_ensure_browser", _no_ensure) @@ -299,7 +299,7 @@ def test_fetch_json_500_on_evaluate_error(monkeypatch: pytest.MonkeyPatch) -> No server._browsers["avito"] = _FakeBrowser(page) monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) - async def _ensure(provider: str) -> bool: + async def _ensure(provider: str, proxy_override: str | None = None) -> bool: return True monkeypatch.setattr(server, "_ensure_browser", _ensure) diff --git a/tradein-mvp/browser/test_server_proxy_override.py b/tradein-mvp/browser/test_server_proxy_override.py new file mode 100644 index 00000000..301402ab --- /dev/null +++ b/tradein-mvp/browser/test_server_proxy_override.py @@ -0,0 +1,153 @@ +"""test_server_proxy_override.py — динамический прокси из тела /fetch (#2164 P4). + +Проверяет: + 1. _resolve_proxy_override — берёт body["proxy"] (kind http/https), None если нет / + если proxy_kind несовместим (socks5 → camoufox не умеет); + 2. политика relaunch «только при реальной смене прокси» через _ensure_browser: + - инстанса нет → launch с proxy_override; + - тот же прокси → НЕ релончим (launch не зовётся повторно); + - другой прокси → close + relaunch с новым; + - proxy_override=None + инстанс на env → НЕ релончим (golden-parity). + +camoufox НЕ запускается: _launch_browser мокается заглушкой, которая ставит +_browsers/_launched_proxy как реальный launch. + +Запуск (из tradein-mvp/browser/):: + + python -m pytest test_server_proxy_override.py -q +""" + +from __future__ import annotations + +import asyncio +import importlib.util +from pathlib import Path + +import pytest + +_SERVER_PATH = Path(__file__).resolve().parent / "server.py" +_spec = importlib.util.spec_from_file_location("tradein_browser_server", _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) + + +@pytest.fixture(autouse=True) +def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(server, "_browsers", {}) + monkeypatch.setattr(server, "_browser_cms", {}) + monkeypatch.setattr(server, "_page_counters", {}) + monkeypatch.setattr(server, "_locks", {}) + monkeypatch.setattr(server, "_retry_tasks", {}) + monkeypatch.setattr(server, "_last_goto_at", {}) + monkeypatch.setattr(server, "_launched_proxy", {}) + monkeypatch.setattr(server, "_locks_guard", asyncio.Lock()) + + +# ── _resolve_proxy_override ────────────────────────────────────────────────────── + + +def test_resolve_proxy_override_http() -> None: + body = {"proxy": "http://u:p@pool:8080", "proxy_kind": "http"} + assert server._resolve_proxy_override(body, "avito") == "http://u:p@pool:8080" + + +def test_resolve_proxy_override_missing_returns_none() -> None: + assert server._resolve_proxy_override({"url": "x"}, "avito") is None + assert server._resolve_proxy_override({"proxy": ""}, "avito") is None + + +def test_resolve_proxy_override_kind_defaults_http_when_absent() -> None: + """proxy без proxy_kind трактуем как http (в пуле kind=http).""" + assert server._resolve_proxy_override({"proxy": "http://pool:8080"}, "cian") == ( + "http://pool:8080" + ) + + +def test_resolve_proxy_override_socks5_ignored() -> None: + """socks5 camoufox не умеет → None (env-fallback).""" + body = {"proxy": "socks5://u:p@pool:1080", "proxy_kind": "socks5"} + assert server._resolve_proxy_override(body, "avito") is None + + +# ── relaunch policy: _ensure_browser ───────────────────────────────────────────── + + +def _install_fake_launch(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str | None]]: + """Мок _launch_browser: пишет _browsers/_launched_proxy как реальный, пишет calls.""" + calls: list[tuple[str, str | None]] = [] + + async def _fake_launch(provider: str, proxy_override: str | None = None) -> None: + calls.append((provider, proxy_override)) + server._browsers[provider] = object() + server._launched_proxy[provider] = ( + proxy_override if proxy_override is not None else server._provider_proxy(provider) + ) + + monkeypatch.setattr(server, "_launch_browser", _fake_launch) + return calls + + +def test_ensure_browser_launches_with_override(monkeypatch: pytest.MonkeyPatch) -> None: + calls = _install_fake_launch(monkeypatch) + + async def _run() -> bool: + return await server._ensure_browser("avito", proxy_override="http://p1:8080") + + ok = asyncio.run(_run()) + assert ok is True + assert calls == [("avito", "http://p1:8080")] + assert server._launched_proxy["avito"] == "http://p1:8080" + + +def test_ensure_browser_same_proxy_no_relaunch(monkeypatch: pytest.MonkeyPatch) -> None: + calls = _install_fake_launch(monkeypatch) + + async def _run() -> None: + await server._ensure_browser("avito", proxy_override="http://p1:8080") + await server._ensure_browser("avito", proxy_override="http://p1:8080") + + asyncio.run(_run()) + # Только один launch — второй вызов с тем же прокси НЕ релончит. + assert calls == [("avito", "http://p1:8080")] + + +def test_ensure_browser_changed_proxy_relaunches(monkeypatch: pytest.MonkeyPatch) -> None: + calls = _install_fake_launch(monkeypatch) + + async def _run() -> None: + await server._ensure_browser("avito", proxy_override="http://p1:8080") + await server._ensure_browser("avito", proxy_override="http://p2:8080") + + asyncio.run(_run()) + assert calls == [("avito", "http://p1:8080"), ("avito", "http://p2:8080")] + assert server._launched_proxy["avito"] == "http://p2:8080" + + +def test_ensure_browser_none_override_env_no_relaunch(monkeypatch: pytest.MonkeyPatch) -> None: + """proxy_override=None (флаг off) + инстанс на env → НЕ релончим (golden-parity).""" + for var in ("BROWSER_PROXY_AVITO", "AVITO_PROXY_URL"): + monkeypatch.setenv(var, "http://env:8080") + calls = _install_fake_launch(monkeypatch) + + async def _run() -> None: + await server._ensure_browser("avito") # lazy launch на env + await server._ensure_browser("avito") # второй раз — env тот же + + asyncio.run(_run()) + assert calls == [("avito", None)] + assert server._launched_proxy["avito"] == "http://env:8080" + + +def test_ensure_browser_env_then_pool_relaunches(monkeypatch: pytest.MonkeyPatch) -> None: + """env-launched инстанс + пришёл pool-proxy → relaunch (реальная смена).""" + for var in ("BROWSER_PROXY_AVITO", "AVITO_PROXY_URL"): + monkeypatch.setenv(var, "http://env:8080") + calls = _install_fake_launch(monkeypatch) + + async def _run() -> None: + await server._ensure_browser("avito") # env + await server._ensure_browser("avito", proxy_override="http://pool:8080") # смена + + asyncio.run(_run()) + assert calls == [("avito", None), ("avito", "http://pool:8080")] diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py index f134a886..023bc913 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py @@ -22,9 +22,15 @@ from __future__ import annotations import asyncio import logging +from collections.abc import Iterator +from contextlib import contextmanager +from typing import TYPE_CHECKING import httpx +if TYPE_CHECKING: + from scraper_kit.contracts import ProxyProvider + logger = logging.getLogger(__name__) _RETRY_SLEEP_S: float = 1.0 @@ -46,6 +52,8 @@ class BrowserFetcher: fetch_timeout_s: float = _HTTP_TIMEOUT_S, *, endpoint: str, + proxy_provider: ProxyProvider | None = None, + use_pool: bool = False, ) -> None: # source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер # роутит /fetch по нему на отдельный браузер+прокси, когда включён @@ -56,10 +64,20 @@ class BrowserFetcher: # endpoint — HTTP-эндпоинт tradein-browser сервиса. Инжектируется вызывающей # стороной (продуктовый код передаёт settings.browser_http_endpoint / # ScraperConfig.browser_http_endpoint) — kit НЕ импортирует app.core.config. + # + # #2164 P4 (ship-dark за use_pool=config.use_proxy_pool_browser): + # proxy_provider + use_pool → на каждый /fetch scraper-сторона делает + # acquire(source), кладёт lease.url в тело запроса ({"proxy": ...}), а + # tradein-browser релончит camoufox с этим прокси (только при реальной смене). + # После fetch — mark_health + release (finally, lease не течёт). use_pool=False + # (дефолт) ИЛИ пустой пул → proxy в теле не шлём, браузер юзает свой env-прокси + # (BROWSER_PROXY_*), поведение не меняется. self._source = source self._fetch_timeout_s = fetch_timeout_s self._client: httpx.AsyncClient | None = None self._endpoint: str | None = endpoint + self._proxy_provider = proxy_provider + self._use_pool = use_pool # ── lifecycle ────────────────────────────────────────────────────────────── @@ -183,18 +201,77 @@ class BrowserFetcher: # ── internal ─────────────────────────────────────────────────────────────── + @contextmanager + def _pool_proxy(self) -> Iterator[tuple[str | None, str | None]]: + """Взять прокси из пула на один /fetch (#2164 P4, ship-dark + fallback). + + use_pool=False (дефолт) ИЛИ proxy_provider=None → yield (None, None): proxy в тело + /fetch не кладётся, tradein-browser юзает свой env-прокси (BROWSER_PROXY_*), + поведение не меняется. Флаг on + пул выдал lease → yield (lease.url, lease.kind); + на выходе mark_health(ok) + release (в finally — lease не течёт). Пул пуст/ошибка + acquire → fallback (None, None), НЕ падаем (сбор цел). ok=False если внутри блока + поднялось исключение (бан/сетевая ошибка) → mark_health(False). + """ + use_pool = self._use_pool and self._proxy_provider is not None + lease = None + if use_pool: + assert self._proxy_provider is not None # type-narrowing (use_pool гарантирует) + try: + lease = self._proxy_provider.acquire(self._source) + except Exception: + logger.warning( + "BrowserFetcher: proxy_pool acquire(%s) failed — fallback to env proxy", + self._source, + exc_info=True, + ) + lease = None + + if lease is None: + # off / пул пуст / ошибка acquire → без proxy в теле (browser юзает env). + yield None, None + return + + assert self._proxy_provider is not None + ok = True + try: + yield lease.url, lease.kind + except Exception: + ok = False + raise + finally: + # mark_health + release — best-effort: проблема пула не должна ронять сбор. + try: + self._proxy_provider.mark_health(lease, ok) + except Exception: + logger.warning( + "BrowserFetcher: proxy_pool mark_health failed for %s", + self._source, + exc_info=True, + ) + try: + self._proxy_provider.release(lease) # ОБЯЗАТЕЛЬНО — lease не течёт + except Exception: + logger.warning( + "BrowserFetcher: proxy_pool release failed for %s", + self._source, + exc_info=True, + ) + async def _post_fetch(self, url: str) -> str: """Один HTTP POST к /fetch эндпоинту сервиса.""" assert self._client is not None assert self._endpoint is not None - resp = await self._client.post( - f"{self._endpoint}/fetch", - json={"url": url, "source": self._source}, - ) - resp.raise_for_status() - data: dict[str, str] = resp.json() - html = data["html"] + with self._pool_proxy() as (proxy_url, proxy_kind): + payload: dict[str, str] = {"url": url, "source": self._source} + if proxy_url: + payload["proxy"] = proxy_url + if proxy_kind: + payload["proxy_kind"] = proxy_kind + resp = await self._client.post(f"{self._endpoint}/fetch", json=payload) + 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 @@ -210,19 +287,22 @@ class BrowserFetcher: assert self._client is not None assert self._endpoint is not None - resp = await self._client.post( - f"{self._endpoint}/fetch-json", - json={ + with self._pool_proxy() as (proxy_url, proxy_kind): + payload: dict = { "url": url, "source": self._source, "method": method, "headers": headers or {}, "body": body, "origin": origin, - }, - ) - resp.raise_for_status() - data: dict = resp.json() + } + if proxy_url: + payload["proxy"] = proxy_url + if proxy_kind: + payload["proxy_kind"] = proxy_kind + resp = await self._client.post(f"{self._endpoint}/fetch-json", json=payload) + resp.raise_for_status() + data: dict = resp.json() # Defensive: контракт сервера — {"status": int, "body": str} (зеркалит как # _post_fetch читает data["html"]). Если ключи пропали (несовместимый сервер / # прокинутый error-payload) — падаем с понятной ошибкой, а не KeyError ниже по diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py index 86ce15e9..e7e1cc11 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py @@ -160,6 +160,12 @@ class ScraperConfig(Protocol): # env-прокси (config.*_proxy_url). Ship-dark: дефолт False → curl-пути ходят как # сейчас. True + пустой пул → fallback на env (сбор не ломается). См. ProxyProvider. use_proxy_pool_curl: bool + # Feature-флаг (#2164 P4): брать прокси из пула на browser-путях (BrowserFetcher → + # POST /fetch{proxy}). Scraper-сторона делает acquire(source) и передаёт lease.url в + # теле запроса; tradein-browser релончит camoufox с этим прокси (relaunch ТОЛЬКО при + # реальной смене). Ship-dark: дефолт False → браузер берёт прокси из env + # (BROWSER_PROXY_*), прод не меняется. True + пустой пул → fallback на env. + use_proxy_pool_browser: bool @runtime_checkable diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index 49ddda42..35655234 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -72,7 +72,12 @@ from scraper_kit.providers.yandex.serp import ( if TYPE_CHECKING: from sqlalchemy.orm import Session - from scraper_kit.contracts import EnrichmentJobs, HouseMatcher, ScraperConfig + from scraper_kit.contracts import ( + EnrichmentJobs, + HouseMatcher, + ProxyProvider, + ScraperConfig, + ) logger = logging.getLogger(__name__) @@ -315,6 +320,7 @@ async def run_avito_pipeline( request_delay_sec: float | None = None, shared_session: AsyncSession | None = None, shared_browser: BrowserFetcher | None = None, + proxy_provider: ProxyProvider | None = None, region_code: int = DEFAULT_REGION_CODE, ) -> PipelineResult: """Full Avito search → houses → detail enrichment pipeline. @@ -357,7 +363,10 @@ async def run_avito_pipeline( browser_fetcher = shared_browser if browser_fetcher is None: browser_fetcher = BrowserFetcher( - source="avito", endpoint=config.browser_http_endpoint + source="avito", + endpoint=config.browser_http_endpoint, + proxy_provider=proxy_provider, + use_pool=config.use_proxy_pool_browser, ) await browser_fetcher.__aenter__() own_browser = True @@ -752,6 +761,7 @@ async def run_avito_city_sweep( matcher: HouseMatcher, enrichment: EnrichmentJobs, shutdown_requested: Callable[[], bool] = lambda: False, + proxy_provider: ProxyProvider | None = None, anchors: list[tuple[float, float, str]] | None = None, radius_m: int = 1500, pages_per_anchor: int = 3, @@ -811,7 +821,12 @@ async def run_avito_city_sweep( shared_bf: BrowserFetcher | None = None if browser_mode: shared_bf = await stack.enter_async_context( - BrowserFetcher(source="avito", endpoint=config.browser_http_endpoint) + BrowserFetcher( + source="avito", + endpoint=config.browser_http_endpoint, + proxy_provider=proxy_provider, + use_pool=config.use_proxy_pool_browser, + ) ) else: session = await stack.enter_async_context( @@ -1457,6 +1472,7 @@ async def run_avito_newbuilding_sweep( config: ScraperConfig, matcher: HouseMatcher, shutdown_requested: Callable[[], bool] = lambda: False, + proxy_provider: ProxyProvider | None = None, pages: int = 20, request_delay_sec: float = 7.0, region_code: int = DEFAULT_REGION_CODE, @@ -1484,7 +1500,12 @@ async def run_avito_newbuilding_sweep( shared_bf: BrowserFetcher | None = None if browser_mode: shared_bf = await stack.enter_async_context( - BrowserFetcher(source="avito", endpoint=config.browser_http_endpoint) + BrowserFetcher( + source="avito", + endpoint=config.browser_http_endpoint, + proxy_provider=proxy_provider, + use_pool=config.use_proxy_pool_browser, + ) ) else: session = await stack.enter_async_context( @@ -2117,6 +2138,7 @@ async def run_cian_city_sweep( config: ScraperConfig, matcher: HouseMatcher, shutdown_requested: Callable[[], bool] = lambda: False, + proxy_provider: ProxyProvider | None = None, anchors: list[tuple[float, float, str]] | None = None, radius_m: int = 1500, pages_per_anchor: int = 3, @@ -2257,7 +2279,9 @@ async def run_cian_city_sweep( source_url: str = row["source_url"] counters.detail_attempted += 1 try: - cian_enrichment = await cian_fetch_detail(source_url) + cian_enrichment = await cian_fetch_detail( + source_url, config=config, proxy_provider=proxy_provider + ) if cian_enrichment is not None: cian_save_detail_enrichment(db, listing_id, cian_enrichment) counters.detail_enriched += 1 @@ -2539,6 +2563,7 @@ async def run_cian_full_load( config: ScraperConfig, matcher: HouseMatcher, shutdown_requested: Callable[[], bool] = lambda: False, + proxy_provider: ProxyProvider | None = None, price_cap_per_bucket: int = 1400, detail_top_n: int = 0, request_delay_sec: float = 4.0, @@ -2728,7 +2753,9 @@ async def run_cian_full_load( source_url: str = row["source_url"] counters.detail_attempted += 1 try: - cian_enrichment = await cian_fetch_detail(source_url) + cian_enrichment = await cian_fetch_detail( + source_url, config=config, proxy_provider=proxy_provider + ) if cian_enrichment is not None: cian_save_detail_enrichment(db, listing_id, cian_enrichment) counters.detail_enriched += 1 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py index ff0fd5ba..e3c91e7e 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py @@ -53,6 +53,7 @@ if TYPE_CHECKING: from scraper_kit.contracts import ( EnrichmentJobs, HouseMatcher, + ProxyProvider, ScraperConfig, SessionFactory, ) @@ -109,6 +110,12 @@ class SchedulerContext: enrichment: EnrichmentJobs session_factory: SessionFactory shutdown_requested: Callable[[], bool] = lambda: False + # Proxy-пул (#2163 curl / #2164 P4 browser). None → пул выключен, curl/browser + # берут прокси из env (ship-dark). RealProxyProvider инжектируется scheduler_main, + # когда включён любой из флагов use_proxy_pool_curl / use_proxy_pool_browser. + # Прокидывается в sweep-пайплайны, откуда попадает в провайдеры (curl_proxy_url) и + # BrowserFetcher (POST /fetch{proxy}). + proxy_provider: ProxyProvider | None = None # runs-модуль (create_run/update_heartbeat/mark_done/… ). По умолчанию — kit-копия # scraper_kit.orchestration.runs; в тестах подменяется recorder'ом. runs: Any = _kit_runs @@ -440,6 +447,7 @@ async def _job_avito_city_sweep( matcher=ctx.matcher, enrichment=ctx.enrichment, shutdown_requested=ctx.shutdown_requested, + proxy_provider=ctx.proxy_provider, pages_per_anchor=int(params.get("pages_per_anchor", 3)), detail_top_n=int(params.get("detail_top_n", 20)), request_delay_sec=float(params.get("request_delay_sec", 7.0)), @@ -457,6 +465,7 @@ async def _job_avito_newbuilding_sweep( config=ctx.config, matcher=ctx.matcher, shutdown_requested=ctx.shutdown_requested, + proxy_provider=ctx.proxy_provider, pages=int(params.get("pages", 20)), request_delay_sec=float(params.get("request_delay_sec", 7.0)), ) @@ -528,6 +537,7 @@ async def _job_cian_city_sweep( config=ctx.config, matcher=ctx.matcher, shutdown_requested=ctx.shutdown_requested, + proxy_provider=ctx.proxy_provider, pages_per_anchor=int(params.get("pages_per_anchor", 3)), request_delay_sec=float(params.get("request_delay_sec", 5.0)), radius_m=int(params.get("radius_m", 1500)), @@ -546,6 +556,7 @@ async def _job_cian_full_load( config=ctx.config, matcher=ctx.matcher, shutdown_requested=ctx.shutdown_requested, + proxy_provider=ctx.proxy_provider, price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)), concurrency=int(params.get("concurrency", 5)), request_delay_sec=float(params.get("request_delay_sec", 4.0)),