feat(proxy-pool): P4 — browser dynamic pool-proxy (acquire→/fetch→relaunch) + orchestration wiring, ship-dark (#2164) #2219

Merged
lekss361 merged 1 commit from feat/proxy-pool-p4-browser into main 2026-07-02 19:20:30 +00:00
12 changed files with 589 additions and 59 deletions

View file

@ -136,10 +136,18 @@ async def _run_kit_scheduler() -> None:
from app.services.scraper_adapters import ( from app.services.scraper_adapters import (
RealEnrichmentJobs, RealEnrichmentJobs,
RealMatcherAdapter, RealMatcherAdapter,
RealProxyProvider,
RealScraperConfig, RealScraperConfig,
RealSessionFactory, 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( ctx = SchedulerContext(
config=RealScraperConfig(), config=RealScraperConfig(),
matcher=RealMatcherAdapter(), matcher=RealMatcherAdapter(),
@ -147,6 +155,7 @@ async def _run_kit_scheduler() -> None:
session_factory=RealSessionFactory(), session_factory=RealSessionFactory(),
shutdown_requested=shutdown_requested, shutdown_requested=shutdown_requested,
runs=runs_mod, runs=runs_mod,
proxy_provider=proxy_provider,
) )
registry = build_registry(product_handlers=build_product_handlers(ctx)) registry = build_registry(product_handlers=build_product_handlers(ctx))
logger.info("scheduler_main: kit-scheduler registry built (%d source handlers)", len(registry)) logger.info("scheduler_main: kit-scheduler registry built (%d source handlers)", len(registry))

View file

@ -210,6 +210,11 @@ class RealScraperConfig:
def use_proxy_pool_curl(self) -> bool: def use_proxy_pool_curl(self) -> bool:
return _settings.use_proxy_pool_curl 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: class RealProxyProvider:
"""ProxyProvider-адаптер над `app.services.proxy_pool` (#2163). """ProxyProvider-адаптер над `app.services.proxy_pool` (#2163).

View file

@ -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 + пул пуст (acquireNone) тела без "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": "<ok>"})
bf = _fetcher(client, source="avito", proxy_provider=provider) # use_pool default False
html = await bf.fetch("https://avito.ru/x")
assert html == "<ok>"
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": "<ok>"})
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": "<ok>"})
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": "<ok>"})
bf = _fetcher(client, source="yandex", proxy_provider=provider, use_pool=True)
html = await bf.fetch("https://yandex.ru/x")
assert html == "<ok>"
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

View file

@ -68,6 +68,8 @@ def test_scraper_config_satisfies_protocol() -> None:
assert isinstance(config.avito_serp_ekb_only, bool) assert isinstance(config.avito_serp_ekb_only, bool)
# #2163: новый флаг proxy-пула присутствует и bool. # #2163: новый флаг proxy-пула присутствует и bool.
assert isinstance(config.use_proxy_pool_curl, 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: def test_proxy_provider_satisfies_protocol() -> None:

View file

@ -266,6 +266,11 @@ _page_counters: dict[str, int] = {} # provider → страниц с мом
_locks: dict[str, asyncio.Lock] = {} # provider → Lock (весь доступ к инстансу) _locks: dict[str, asyncio.Lock] = {} # provider → Lock (весь доступ к инстансу)
_retry_tasks: dict[str, asyncio.Task[None]] = {} # provider → фоновая retry-задача _retry_tasks: dict[str, asyncio.Task[None]] = {} # provider → фоновая retry-задача
_last_goto_at: dict[str, float] = {} # provider → loop-time последнего page.goto (пейсинг) _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 из # Guard на ленивое создание per-provider локов: setdefault на обычном dict из
# разных корутин гонок не даёт (нет await между read-modify-write), но держим # разных корутин гонок не даёт (нет await между read-modify-write), но держим
@ -297,15 +302,21 @@ async def _lock_for(provider: str) -> asyncio.Lock:
return _locks.setdefault(provider, 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]. """Запускает AsyncCamoufox для поставщика и кладёт в *_…[provider].
Использует _provider_proxy(provider) + geoip=True (фингерпринт привязан к гео proxy_override (#2164 P4): если задан — запускаем camoufox с ЭТИМ прокси (динамический
IP конкретного прокси поэтому отдельный инстанс на каждого поставщика). из пула, переданный scraper-стороной в теле /fetch). None env-прокси
_provider_proxy(provider) как раньше. geoip=True (фингерпринт привязан к гео IP
конкретного прокси поэтому отдельный инстанс на каждого поставщика).
Эффективный proxy-url сохраняется в _launched_proxy[provider] для политики
«relaunch только при реальной смене» (см. _ensure_browser).
""" """
from camoufox.async_api import AsyncCamoufox 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] = { kwargs: dict[str, object] = {
"headless": True, "headless": True,
"os": "windows", "os": "windows",
@ -336,7 +347,12 @@ async def _launch_browser(provider: str) -> None:
_browser_cms[provider] = cm _browser_cms[provider] = cm
_browsers[provider] = browser _browsers[provider] = browser
_page_counters[provider] = 0 _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: async def _close_browser(provider: str) -> None:
@ -356,20 +372,26 @@ async def _close_browser(provider: str) -> None:
_browser_cms.pop(provider, None) _browser_cms.pop(provider, None)
_page_counters[provider] = 0 _page_counters[provider] = 0
_last_goto_at.pop(provider, None) _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). """Закрывает и заново запускает per-provider браузер (recycle / crash-recovery).
Caller держит _locks[provider], поэтому параллельных страниц на этом инстансе proxy_override (#2164 P4): прокинуть конкретный прокси в новый инстанс. None →
нет teardown безопасен. СОХРАНИТЬ текущий прокси инстанса (_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) logger.info("tradein-browser[%s]: перезапуск браузера", provider)
await _close_browser(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'а поставщика с санитизированным логом фейла. """Одна попытка launch'а поставщика с санитизированным логом фейла.
НЕ пробрасывает исключение наружу возвращает True/False. Критично для НЕ пробрасывает исключение наружу возвращает True/False. Критично для
@ -382,7 +404,7 @@ async def _try_launch_browser(provider: str) -> bool:
через _close_browser, чтобы следующий retry стартовал с чистого листа. через _close_browser, чтобы следующий retry стартовал с чистого листа.
""" """
try: try:
await _launch_browser(provider) await _launch_browser(provider, proxy_override=proxy_override)
return True return True
except Exception as exc: except Exception as exc:
logger.warning( logger.warning(
@ -395,25 +417,41 @@ async def _try_launch_browser(provider: str) -> bool:
return False return False
async def _ensure_browser(provider: str) -> bool: async def _ensure_browser(provider: str, proxy_override: str | None = None) -> bool:
"""Гарантирует поднятый инстанс поставщика; одна lazy-попытка launch'а. """Гарантирует поднятый инстанс поставщика с нужным прокси; lazy launch / relaunch.
Caller держит _locks[provider] (весь доступ к инстансу сериализован), поэтому proxy_override (#2164 P4): желаемый динамический прокси из пула (тело /fetch). None →
здесь своей сериализации нет двух одновременных launch'ей на один provider желаемый прокси = 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: Returns:
True если браузер доступен (был или поднялся), False если launch не удался True если браузер доступен (был или поднялся), False если launch не удался
(прокси лежит) caller отдаёт 503. (прокси лежит) caller отдаёт 503.
При неудачном lazy-launch (прокси лёг) запускает фоновую retry-задачу для При неудачном launch (прокси лёг) запускает фоновую retry-задачу для ЭТОГО
ЭТОГО поставщика: при eager-startup её ставил _on_startup, в lazy-режиме поставщика (env-прокси). Поставщики независимы фоновый retry поднимается только
первый обратившийся /fetch. Поставщики независимы фоновый retry поднимается для тех, к кому был запрос, а не для всех сразу.
только для тех, к кому был запрос, а не для всех сразу.
""" """
desired_proxy = proxy_override if proxy_override is not None else _provider_proxy(provider)
if _browsers.get(provider) is not None: if _browsers.get(provider) is not None:
return True if _launched_proxy.get(provider) == desired_proxy:
await _try_launch_browser(provider) 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: if _browsers.get(provider) is not None:
return True return True
_start_retry_task(provider) _start_retry_task(provider)
@ -520,6 +558,32 @@ def _resolve_provider(body: dict, url: str) -> str:
return _provider_for_url(url) 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: async def health_handler(request: web.Request) -> web.Response:
# Всегда 200 — иначе compose healthcheck (если появится) убьёт контейнер, # Всегда 200 — иначе compose healthcheck (если появится) убьёт контейнер,
# хотя процесс жив и ретраит браузеры. browsers — per-provider readiness для # хотя процесс жив и ретраит браузеры. 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) return web.json_response({"error": "missing 'url' field"}, status=400)
provider = _resolve_provider(body, url) provider = _resolve_provider(body, url)
proxy_override = _resolve_proxy_override(body, provider)
lock = await _lock_for(provider) lock = await _lock_for(provider)
async with lock: async with lock:
# Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка — # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка —
# вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/браузера: если # вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/браузера: если
# всё ещё None → 503 (caller словит как HTTPStatusError → failed fetch, не # всё ещё None → 503 (caller словит как HTTPStatusError → failed fetch, не
# crash). # crash). proxy_override (#2164 P4): relaunch camoufox с прокси из пула, если
if not await _ensure_browser(provider): # он отличается от текущего launch-прокси инстанса (иначе no-op).
if not await _ensure_browser(provider, proxy_override=proxy_override):
logger.warning( logger.warning(
"tradein-browser[%s]: /fetch 503 — браузер недоступен (proxy may be down)", "tradein-browser[%s]: /fetch 503 — браузер недоступен (proxy may be down)",
provider, provider,
@ -611,12 +677,14 @@ async def fetch_json_handler(request: web.Request) -> web.Response:
origin = f"{parsed.scheme}://{parsed.netloc}/" origin = f"{parsed.scheme}://{parsed.netloc}/"
provider = _resolve_provider(body, url) provider = _resolve_provider(body, url)
proxy_override = _resolve_proxy_override(body, provider)
lock = await _lock_for(provider) lock = await _lock_for(provider)
async with lock: async with lock:
# Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал). # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал).
# Lazy-попытка, иначе 503 — без прокси/браузера не фетчим. # Lazy-попытка, иначе 503 — без прокси/браузера не фетчим. proxy_override
if not await _ensure_browser(provider): # (#2164 P4): relaunch с прокси из пула при реальной смене.
if not await _ensure_browser(provider, proxy_override=proxy_override):
logger.warning( logger.warning(
"tradein-browser[%s]: /fetch-json 503 — браузер недоступен (proxy may be down)", "tradein-browser[%s]: /fetch-json 503 — браузер недоступен (proxy may be down)",
provider, provider,
@ -705,7 +773,8 @@ async def _do_fetch(provider: str, url: str) -> str:
"""Одна попытка навигации; при краше браузера — relaunch и один retry. """Одна попытка навигации; при краше браузера — relaunch и один retry.
Caller держит _locks[provider] (нет параллельных страниц на этом инстансе), Caller держит _locks[provider] (нет параллельных страниц на этом инстансе),
поэтому relaunch безопасен никого не рвёт из-под живых страниц. поэтому relaunch безопасен. Crash-relaunch сохраняет текущий прокси инстанса
(_relaunch_browser reuse _launched_proxy) динамический прокси пула не теряется.
""" """
try: try:
return await _fetch_once(provider, url) return await _fetch_once(provider, url)
@ -775,7 +844,8 @@ async def _do_fetch_json(
"""Одна попытка in-page fetch'а; при краше браузера — relaunch и один retry. """Одна попытка in-page fetch'а; при краше браузера — relaunch и один retry.
Зеркалит _do_fetch: caller держит _locks[provider] (нет параллельных страниц на Зеркалит _do_fetch: caller держит _locks[provider] (нет параллельных страниц на
этом инстансе), поэтому relaunch безопасен никого не рвёт из-под живых страниц. этом инстансе), поэтому relaunch безопасен. Crash-relaunch сохраняет текущий прокси
инстанса (_relaunch_browser reuse _launched_proxy).
""" """
try: try:
return await _fetch_json_once( return await _fetch_json_once(

View file

@ -219,7 +219,7 @@ def test_same_provider_fetches_are_serialized(monkeypatch: pytest.MonkeyPatch) -
"""Два /fetch к ОДНОМУ поставщику сериализованы (один лок) — peak==1.""" """Два /fetch к ОДНОМУ поставщику сериализованы (один лок) — peak==1."""
probe = _OverlapProbe() probe = _OverlapProbe()
async def _ensure(provider: str) -> bool: async def _ensure(provider: str, proxy_override: str | None = None) -> bool:
server._browsers[provider] = object() server._browsers[provider] = object()
return True return True
@ -241,7 +241,7 @@ def test_different_providers_fetch_concurrently(monkeypatch: pytest.MonkeyPatch)
"""Два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи) — global peak>1.""" """Два /fetch к РАЗНЫМ поставщикам идут параллельно (разные локи) — global peak>1."""
probe = _OverlapProbe() probe = _OverlapProbe()
async def _ensure(provider: str) -> bool: async def _ensure(provider: str, proxy_override: str | None = None) -> bool:
server._browsers[provider] = object() server._browsers[provider] = object()
return True return True
@ -271,7 +271,7 @@ def test_mixed_load_serializes_within_parallelizes_across(
"""Смешанная нагрузка: avito×2 сериализованы, но параллельны с cian×1.""" """Смешанная нагрузка: avito×2 сериализованы, но параллельны с cian×1."""
probe = _OverlapProbe() probe = _OverlapProbe()
async def _ensure(provider: str) -> bool: async def _ensure(provider: str, proxy_override: str | None = None) -> bool:
server._browsers[provider] = object() server._browsers[provider] = object()
return True return True

View file

@ -109,7 +109,7 @@ def test_fetch_json_returns_evaluate_result(monkeypatch: pytest.MonkeyPatch) ->
server._browsers["avito"] = _FakeBrowser(page) server._browsers["avito"] = _FakeBrowser(page)
monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) 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) server._browsers["avito"] = _FakeBrowser(page)
monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) 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) server._browsers["avito"] = _FakeBrowser(page)
monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) 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) server._browsers["avito"] = _FakeBrowser(page)
monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) 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, "BROWSER_RECYCLE_PAGES", 10_000)
monkeypatch.setattr(server, "FETCH_JSON_SETTLE_MS", 1777) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) 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_INPAGE_RETRIES", 2)
monkeypatch.setattr(server, "FETCH_JSON_RETRY_DELAY_MS", 654) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) 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: def test_fetch_json_503_when_browser_unavailable(monkeypatch: pytest.MonkeyPatch) -> None:
"""/fetch-json отдаёт 503 (не 500/краш) когда инстанс поставщика не поднят.""" """/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 return False # прокси всё ещё недоступен → браузер остаётся None
monkeypatch.setattr(server, "_ensure_browser", _no_ensure) 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) server._browsers["avito"] = _FakeBrowser(page)
monkeypatch.setattr(server, "BROWSER_RECYCLE_PAGES", 10_000) 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 return True
monkeypatch.setattr(server, "_ensure_browser", _ensure) monkeypatch.setattr(server, "_ensure_browser", _ensure)

View file

@ -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")]

View file

@ -22,9 +22,15 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
from collections.abc import Iterator
from contextlib import contextmanager
from typing import TYPE_CHECKING
import httpx import httpx
if TYPE_CHECKING:
from scraper_kit.contracts import ProxyProvider
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_RETRY_SLEEP_S: float = 1.0 _RETRY_SLEEP_S: float = 1.0
@ -46,6 +52,8 @@ class BrowserFetcher:
fetch_timeout_s: float = _HTTP_TIMEOUT_S, fetch_timeout_s: float = _HTTP_TIMEOUT_S,
*, *,
endpoint: str, endpoint: str,
proxy_provider: ProxyProvider | None = None,
use_pool: bool = False,
) -> None: ) -> None:
# source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер # source — логический источник ("avito"/"cian"/"yandex"/"domclick"). Сервер
# роутит /fetch по нему на отдельный браузер+прокси, когда включён # роутит /fetch по нему на отдельный браузер+прокси, когда включён
@ -56,10 +64,20 @@ class BrowserFetcher:
# endpoint — HTTP-эндпоинт tradein-browser сервиса. Инжектируется вызывающей # endpoint — HTTP-эндпоинт tradein-browser сервиса. Инжектируется вызывающей
# стороной (продуктовый код передаёт settings.browser_http_endpoint / # стороной (продуктовый код передаёт settings.browser_http_endpoint /
# ScraperConfig.browser_http_endpoint) — kit НЕ импортирует app.core.config. # 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._source = source
self._fetch_timeout_s = fetch_timeout_s self._fetch_timeout_s = fetch_timeout_s
self._client: httpx.AsyncClient | None = None self._client: httpx.AsyncClient | None = None
self._endpoint: str | None = endpoint self._endpoint: str | None = endpoint
self._proxy_provider = proxy_provider
self._use_pool = use_pool
# ── lifecycle ────────────────────────────────────────────────────────────── # ── lifecycle ──────────────────────────────────────────────────────────────
@ -183,18 +201,77 @@ class BrowserFetcher:
# ── internal ─────────────────────────────────────────────────────────────── # ── 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: async def _post_fetch(self, url: str) -> str:
"""Один HTTP POST к /fetch эндпоинту сервиса.""" """Один HTTP POST к /fetch эндпоинту сервиса."""
assert self._client is not None assert self._client is not None
assert self._endpoint is not None assert self._endpoint is not None
resp = await self._client.post( with self._pool_proxy() as (proxy_url, proxy_kind):
f"{self._endpoint}/fetch", payload: dict[str, str] = {"url": url, "source": self._source}
json={"url": url, "source": self._source}, if proxy_url:
) payload["proxy"] = proxy_url
resp.raise_for_status() if proxy_kind:
data: dict[str, str] = resp.json() payload["proxy_kind"] = proxy_kind
html = data["html"] 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)) logger.debug("BrowserFetcher: fetch OK url=%r html_len=%d", url, len(html))
return html return html
@ -210,19 +287,22 @@ class BrowserFetcher:
assert self._client is not None assert self._client is not None
assert self._endpoint is not None assert self._endpoint is not None
resp = await self._client.post( with self._pool_proxy() as (proxy_url, proxy_kind):
f"{self._endpoint}/fetch-json", payload: dict = {
json={
"url": url, "url": url,
"source": self._source, "source": self._source,
"method": method, "method": method,
"headers": headers or {}, "headers": headers or {},
"body": body, "body": body,
"origin": origin, "origin": origin,
}, }
) if proxy_url:
resp.raise_for_status() payload["proxy"] = proxy_url
data: dict = resp.json() 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} (зеркалит как # Defensive: контракт сервера — {"status": int, "body": str} (зеркалит как
# _post_fetch читает data["html"]). Если ключи пропали (несовместимый сервер / # _post_fetch читает data["html"]). Если ключи пропали (несовместимый сервер /
# прокинутый error-payload) — падаем с понятной ошибкой, а не KeyError ниже по # прокинутый error-payload) — падаем с понятной ошибкой, а не KeyError ниже по

View file

@ -160,6 +160,12 @@ class ScraperConfig(Protocol):
# env-прокси (config.*_proxy_url). Ship-dark: дефолт False → curl-пути ходят как # env-прокси (config.*_proxy_url). Ship-dark: дефолт False → curl-пути ходят как
# сейчас. True + пустой пул → fallback на env (сбор не ломается). См. ProxyProvider. # сейчас. True + пустой пул → fallback на env (сбор не ломается). См. ProxyProvider.
use_proxy_pool_curl: bool 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 @runtime_checkable

View file

@ -72,7 +72,12 @@ from scraper_kit.providers.yandex.serp import (
if TYPE_CHECKING: if TYPE_CHECKING:
from sqlalchemy.orm import Session 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__) logger = logging.getLogger(__name__)
@ -315,6 +320,7 @@ async def run_avito_pipeline(
request_delay_sec: float | None = None, request_delay_sec: float | None = None,
shared_session: AsyncSession | None = None, shared_session: AsyncSession | None = None,
shared_browser: BrowserFetcher | None = None, shared_browser: BrowserFetcher | None = None,
proxy_provider: ProxyProvider | None = None,
region_code: int = DEFAULT_REGION_CODE, region_code: int = DEFAULT_REGION_CODE,
) -> PipelineResult: ) -> PipelineResult:
"""Full Avito search → houses → detail enrichment pipeline. """Full Avito search → houses → detail enrichment pipeline.
@ -357,7 +363,10 @@ async def run_avito_pipeline(
browser_fetcher = shared_browser browser_fetcher = shared_browser
if browser_fetcher is None: if browser_fetcher is None:
browser_fetcher = BrowserFetcher( 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__() await browser_fetcher.__aenter__()
own_browser = True own_browser = True
@ -752,6 +761,7 @@ async def run_avito_city_sweep(
matcher: HouseMatcher, matcher: HouseMatcher,
enrichment: EnrichmentJobs, enrichment: EnrichmentJobs,
shutdown_requested: Callable[[], bool] = lambda: False, shutdown_requested: Callable[[], bool] = lambda: False,
proxy_provider: ProxyProvider | None = None,
anchors: list[tuple[float, float, str]] | None = None, anchors: list[tuple[float, float, str]] | None = None,
radius_m: int = 1500, radius_m: int = 1500,
pages_per_anchor: int = 3, pages_per_anchor: int = 3,
@ -811,7 +821,12 @@ async def run_avito_city_sweep(
shared_bf: BrowserFetcher | None = None shared_bf: BrowserFetcher | None = None
if browser_mode: if browser_mode:
shared_bf = await stack.enter_async_context( 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: else:
session = await stack.enter_async_context( session = await stack.enter_async_context(
@ -1457,6 +1472,7 @@ async def run_avito_newbuilding_sweep(
config: ScraperConfig, config: ScraperConfig,
matcher: HouseMatcher, matcher: HouseMatcher,
shutdown_requested: Callable[[], bool] = lambda: False, shutdown_requested: Callable[[], bool] = lambda: False,
proxy_provider: ProxyProvider | None = None,
pages: int = 20, pages: int = 20,
request_delay_sec: float = 7.0, request_delay_sec: float = 7.0,
region_code: int = DEFAULT_REGION_CODE, region_code: int = DEFAULT_REGION_CODE,
@ -1484,7 +1500,12 @@ async def run_avito_newbuilding_sweep(
shared_bf: BrowserFetcher | None = None shared_bf: BrowserFetcher | None = None
if browser_mode: if browser_mode:
shared_bf = await stack.enter_async_context( 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: else:
session = await stack.enter_async_context( session = await stack.enter_async_context(
@ -2117,6 +2138,7 @@ async def run_cian_city_sweep(
config: ScraperConfig, config: ScraperConfig,
matcher: HouseMatcher, matcher: HouseMatcher,
shutdown_requested: Callable[[], bool] = lambda: False, shutdown_requested: Callable[[], bool] = lambda: False,
proxy_provider: ProxyProvider | None = None,
anchors: list[tuple[float, float, str]] | None = None, anchors: list[tuple[float, float, str]] | None = None,
radius_m: int = 1500, radius_m: int = 1500,
pages_per_anchor: int = 3, pages_per_anchor: int = 3,
@ -2257,7 +2279,9 @@ async def run_cian_city_sweep(
source_url: str = row["source_url"] source_url: str = row["source_url"]
counters.detail_attempted += 1 counters.detail_attempted += 1
try: 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: if cian_enrichment is not None:
cian_save_detail_enrichment(db, listing_id, cian_enrichment) cian_save_detail_enrichment(db, listing_id, cian_enrichment)
counters.detail_enriched += 1 counters.detail_enriched += 1
@ -2539,6 +2563,7 @@ async def run_cian_full_load(
config: ScraperConfig, config: ScraperConfig,
matcher: HouseMatcher, matcher: HouseMatcher,
shutdown_requested: Callable[[], bool] = lambda: False, shutdown_requested: Callable[[], bool] = lambda: False,
proxy_provider: ProxyProvider | None = None,
price_cap_per_bucket: int = 1400, price_cap_per_bucket: int = 1400,
detail_top_n: int = 0, detail_top_n: int = 0,
request_delay_sec: float = 4.0, request_delay_sec: float = 4.0,
@ -2728,7 +2753,9 @@ async def run_cian_full_load(
source_url: str = row["source_url"] source_url: str = row["source_url"]
counters.detail_attempted += 1 counters.detail_attempted += 1
try: 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: if cian_enrichment is not None:
cian_save_detail_enrichment(db, listing_id, cian_enrichment) cian_save_detail_enrichment(db, listing_id, cian_enrichment)
counters.detail_enriched += 1 counters.detail_enriched += 1

View file

@ -53,6 +53,7 @@ if TYPE_CHECKING:
from scraper_kit.contracts import ( from scraper_kit.contracts import (
EnrichmentJobs, EnrichmentJobs,
HouseMatcher, HouseMatcher,
ProxyProvider,
ScraperConfig, ScraperConfig,
SessionFactory, SessionFactory,
) )
@ -109,6 +110,12 @@ class SchedulerContext:
enrichment: EnrichmentJobs enrichment: EnrichmentJobs
session_factory: SessionFactory session_factory: SessionFactory
shutdown_requested: Callable[[], bool] = lambda: False 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-копия # runs-модуль (create_run/update_heartbeat/mark_done/… ). По умолчанию — kit-копия
# scraper_kit.orchestration.runs; в тестах подменяется recorder'ом. # scraper_kit.orchestration.runs; в тестах подменяется recorder'ом.
runs: Any = _kit_runs runs: Any = _kit_runs
@ -440,6 +447,7 @@ async def _job_avito_city_sweep(
matcher=ctx.matcher, matcher=ctx.matcher,
enrichment=ctx.enrichment, enrichment=ctx.enrichment,
shutdown_requested=ctx.shutdown_requested, shutdown_requested=ctx.shutdown_requested,
proxy_provider=ctx.proxy_provider,
pages_per_anchor=int(params.get("pages_per_anchor", 3)), pages_per_anchor=int(params.get("pages_per_anchor", 3)),
detail_top_n=int(params.get("detail_top_n", 20)), detail_top_n=int(params.get("detail_top_n", 20)),
request_delay_sec=float(params.get("request_delay_sec", 7.0)), request_delay_sec=float(params.get("request_delay_sec", 7.0)),
@ -457,6 +465,7 @@ async def _job_avito_newbuilding_sweep(
config=ctx.config, config=ctx.config,
matcher=ctx.matcher, matcher=ctx.matcher,
shutdown_requested=ctx.shutdown_requested, shutdown_requested=ctx.shutdown_requested,
proxy_provider=ctx.proxy_provider,
pages=int(params.get("pages", 20)), pages=int(params.get("pages", 20)),
request_delay_sec=float(params.get("request_delay_sec", 7.0)), request_delay_sec=float(params.get("request_delay_sec", 7.0)),
) )
@ -528,6 +537,7 @@ async def _job_cian_city_sweep(
config=ctx.config, config=ctx.config,
matcher=ctx.matcher, matcher=ctx.matcher,
shutdown_requested=ctx.shutdown_requested, shutdown_requested=ctx.shutdown_requested,
proxy_provider=ctx.proxy_provider,
pages_per_anchor=int(params.get("pages_per_anchor", 3)), pages_per_anchor=int(params.get("pages_per_anchor", 3)),
request_delay_sec=float(params.get("request_delay_sec", 5.0)), request_delay_sec=float(params.get("request_delay_sec", 5.0)),
radius_m=int(params.get("radius_m", 1500)), radius_m=int(params.get("radius_m", 1500)),
@ -546,6 +556,7 @@ async def _job_cian_full_load(
config=ctx.config, config=ctx.config,
matcher=ctx.matcher, matcher=ctx.matcher,
shutdown_requested=ctx.shutdown_requested, shutdown_requested=ctx.shutdown_requested,
proxy_provider=ctx.proxy_provider,
price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)), price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
concurrency=int(params.get("concurrency", 5)), concurrency=int(params.get("concurrency", 5)),
request_delay_sec=float(params.get("request_delay_sec", 4.0)), request_delay_sec=float(params.get("request_delay_sec", 4.0)),