feat(proxy-pool): P4 — browser dynamic pool-proxy (acquire→/fetch→relaunch) + orchestration wiring, ship-dark (#2164) #2219
12 changed files with 589 additions and 59 deletions
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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).
|
||||
|
|
|
|||
167
tradein-mvp/backend/tests/test_kit_browser_fetcher_proxy_pool.py
Normal file
167
tradein-mvp/backend/tests/test_kit_browser_fetcher_proxy_pool.py
Normal 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 + пул пуст (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": "<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
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
if _launched_proxy.get(provider) == desired_proxy:
|
||||
return True
|
||||
await _try_launch_browser(provider)
|
||||
# Реальная смена прокси → пересобрать инстанс с новым прокси.
|
||||
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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
153
tradein-mvp/browser/test_server_proxy_override.py
Normal file
153
tradein-mvp/browser/test_server_proxy_override.py
Normal 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")]
|
||||
|
|
@ -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,15 +201,74 @@ 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},
|
||||
)
|
||||
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"]
|
||||
|
|
@ -210,17 +287,20 @@ 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,
|
||||
},
|
||||
)
|
||||
}
|
||||
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} (зеркалит как
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)),
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue