diff --git a/tradein-mvp/backend/app/services/scraper_adapters.py b/tradein-mvp/backend/app/services/scraper_adapters.py index e193747f..b2b4122a 100644 --- a/tradein-mvp/backend/app/services/scraper_adapters.py +++ b/tradein-mvp/backend/app/services/scraper_adapters.py @@ -21,6 +21,7 @@ from typing import TYPE_CHECKING, Any from app.core.config import settings as _settings from app.core.db import SessionLocal as _SessionLocal +from app.services import proxy_pool as _proxy_pool from app.services.house_dedup_merge import run_house_dedup_merge as _run_house_dedup_merge from app.services.house_imv_backfill import backfill_house_imv as _backfill_house_imv from app.services.house_imv_backfill import ( @@ -40,6 +41,7 @@ from app.services.yandex_price_history import ( ) if TYPE_CHECKING: + from scraper_kit.contracts import ProxyLease from sqlalchemy.orm import Session from app.services.house_imv_backfill import HouseIMVBackfillResult @@ -201,6 +203,59 @@ class RealScraperConfig: def cian_full_load_per_fetch_timeout_s(self) -> float: return _settings.cian_full_load_per_fetch_timeout_s + # ── Proxy-pool (#2163) ──────────────────────────────────────────────── + @property + def use_proxy_pool_curl(self) -> bool: + return _settings.use_proxy_pool_curl + + +class RealProxyProvider: + """ProxyProvider-адаптер над `app.services.proxy_pool` (#2163). + + Прячет `Session` от kit'а: открывает короткую сессию на каждую операцию + (acquire/release/mark_health), проксирует в модульные функции proxy_pool и + коммитит внутри них. Возвращает kit-`ProxyLease` (структурно совпадает с + `proxy_pool.ProxyLease`), чтобы провайдеры не импортировали `app.*`. + + Не-run вызовы (curl-путь эстиматора вне scrape_run) лизят под + NON_RUN_LEASE_MARKER — reap_stale_leases освободит, если release не дошёл. + """ + + def acquire(self, provider: str) -> ProxyLease | None: + from scraper_kit.contracts import ProxyLease as _KitProxyLease + + db = _SessionLocal() + try: + lease = _proxy_pool.acquire(db, provider) + finally: + db.close() + if lease is None: + return None + return _KitProxyLease( + id=lease.id, url=lease.url, kind=lease.kind, rotate_url=lease.rotate_url + ) + + def release(self, lease: ProxyLease) -> None: + db = _SessionLocal() + try: + _proxy_pool.release(db, lease.id) + finally: + db.close() + + def mark_health( + self, + lease: ProxyLease, + ok: bool, + *, + exit_ip: str | None = None, + latency_ms: int | None = None, + ) -> None: + db = _SessionLocal() + try: + _proxy_pool.mark_health(db, lease.id, ok, exit_ip=exit_ip, latency_ms=latency_ms) + finally: + db.close() + class RealSessionFactory: """SessionFactory-адаптер над `app.core.db.SessionLocal`.""" @@ -281,6 +336,7 @@ class RealEnrichmentJobs: __all__ = [ "RealEnrichmentJobs", "RealMatcherAdapter", + "RealProxyProvider", "RealScraperConfig", "RealSessionFactory", ] diff --git a/tradein-mvp/backend/tests/test_proxy_pool_curl_paths.py b/tradein-mvp/backend/tests/test_proxy_pool_curl_paths.py new file mode 100644 index 00000000..3d9f3405 --- /dev/null +++ b/tradein-mvp/backend/tests/test_proxy_pool_curl_paths.py @@ -0,0 +1,175 @@ +"""P3 (#2163): kit curl-пути берут прокси из пула за флагом USE_PROXY_POOL_CURL. + +Покрывает инвариант ship-dark + fallback на уровне helper'а `curl_proxy_url` и +class-based провайдера (YandexValuationScraper): + - флаг off / proxy_provider=None → env-прокси, пул не трогается (golden-parity); + - флаг on + пул пуст (acquire→None) → fallback env, не падаем; + - флаг on + lease → fetch через lease.url, mark_health вызван, release в finally; + - исключение внутри блока → mark_health(ok=False) + release всё равно (lease не течёт); + - acquire кинул → fallback env (сбор не ломаем). +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.contracts import ProxyLease +from scraper_kit.providers._proxy import curl_proxy_url + +# ── Фейки ───────────────────────────────────────────────────────────────────── + + +@dataclass +class _FakeConfig: + use_proxy_pool_curl: bool = False + scraper_proxy_url: str | None = None + + +class _SpyProvider: + """ProxyProvider-заглушка, записывающая вызовы.""" + + def __init__(self, lease: ProxyLease | None, *, acquire_raises: bool = False) -> None: + self._lease = lease + self._acquire_raises = acquire_raises + self.acquire_calls: list[str] = [] + self.release_calls: list[int] = [] + self.mark_health_calls: list[tuple[int, bool]] = [] + + def acquire(self, provider: str) -> ProxyLease | None: + self.acquire_calls.append(provider) + if self._acquire_raises: + raise RuntimeError("boom acquire") + return self._lease + + def release(self, lease: ProxyLease) -> None: + self.release_calls.append(lease.id) + + def mark_health( + self, lease: ProxyLease, ok: bool, *, exit_ip: Any = None, latency_ms: Any = None + ) -> None: + self.mark_health_calls.append((lease.id, ok)) + + +_LEASE = ProxyLease(id=7, url="http://user:pass@pool-proxy:3128", kind="http", rotate_url=None) + + +# ── curl_proxy_url: helper-инвариант ────────────────────────────────────────── + + +def test_flag_off_yields_env_and_never_touches_pool() -> None: + cfg = _FakeConfig(use_proxy_pool_curl=False) + spy = _SpyProvider(_LEASE) + with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128") as url: + assert url == "http://env:3128" + # golden-parity: пул не задействован при выключенном флаге + assert spy.acquire_calls == [] + assert spy.mark_health_calls == [] + assert spy.release_calls == [] + + +def test_provider_none_yields_env() -> None: + cfg = _FakeConfig(use_proxy_pool_curl=True) + with curl_proxy_url(cfg, None, "cian", env_fallback_url="http://env:3128") as url: + assert url == "http://env:3128" + + +def test_flag_on_empty_pool_falls_back_to_env() -> None: + cfg = _FakeConfig(use_proxy_pool_curl=True) + spy = _SpyProvider(None) # acquire → None (пул пуст/выключен) + with curl_proxy_url(cfg, spy, "avito", env_fallback_url="http://env:3128") as url: + assert url == "http://env:3128" + assert spy.acquire_calls == ["avito"] + # lease не выдан → mark_health/release не зовём + assert spy.mark_health_calls == [] + assert spy.release_calls == [] + + +def test_flag_on_lease_used_and_released_on_success() -> None: + cfg = _FakeConfig(use_proxy_pool_curl=True) + spy = _SpyProvider(_LEASE) + with curl_proxy_url(cfg, spy, "yandex", env_fallback_url="http://env:3128") as url: + assert url == _LEASE.url # fetch идёт через lease.url, НЕ env + assert spy.acquire_calls == ["yandex"] + assert spy.mark_health_calls == [(7, True)] + assert spy.release_calls == [7] # release в finally + + +def test_flag_on_exception_marks_fail_and_still_releases() -> None: + cfg = _FakeConfig(use_proxy_pool_curl=True) + spy = _SpyProvider(_LEASE) + with pytest.raises(RuntimeError, match="ban"): + with curl_proxy_url(cfg, spy, "cian", env_fallback_url=None) as url: + assert url == _LEASE.url + raise RuntimeError("ban 403") + # исключение → ok=False, но lease ОБЯЗАТЕЛЬНО освобождён (не течёт) + assert spy.mark_health_calls == [(7, False)] + assert spy.release_calls == [7] + + +def test_acquire_raises_falls_back_to_env() -> None: + cfg = _FakeConfig(use_proxy_pool_curl=True) + spy = _SpyProvider(_LEASE, acquire_raises=True) + with curl_proxy_url(cfg, spy, "cian", env_fallback_url="http://env:3128") as url: + assert url == "http://env:3128" # ошибка acquire → env, не падаем + assert spy.release_calls == [] + + +# ── YandexValuationScraper: lease держится на всё время сессии ───────────────── + + +@pytest.mark.asyncio +async def test_yandex_scraper_flag_off_no_pool() -> None: + from scraper_kit.providers.yandex.valuation import YandexValuationScraper + + cfg = _FakeConfig(use_proxy_pool_curl=False, scraper_proxy_url="http://env:3128") + spy = _SpyProvider(_LEASE) + with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls: + sess_cls.return_value = MagicMock(close=AsyncMock()) + async with YandexValuationScraper(cfg, proxy_provider=spy): + pass + # флаг off → env-прокси, пул не тронут + _, kwargs = sess_cls.call_args + assert kwargs["proxies"] == {"http": "http://env:3128", "https": "http://env:3128"} + assert spy.acquire_calls == [] + + +@pytest.mark.asyncio +async def test_yandex_scraper_flag_on_lease_lifecycle() -> None: + from scraper_kit.providers.yandex.valuation import YandexValuationScraper + + cfg = _FakeConfig(use_proxy_pool_curl=True, scraper_proxy_url="http://env:3128") + spy = _SpyProvider(_LEASE) + with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls: + sess_cls.return_value = MagicMock(close=AsyncMock()) + async with YandexValuationScraper(cfg, proxy_provider=spy): + # сессия создана через lease.url, lease ещё удерживается + _, kwargs = sess_cls.call_args + assert kwargs["proxies"] == {"http": _LEASE.url, "https": _LEASE.url} + assert spy.release_calls == [] + # выход из сессии → mark_health(ok) + release + assert spy.acquire_calls == ["yandex"] + assert spy.mark_health_calls == [(7, True)] + assert spy.release_calls == [7] + + +@pytest.mark.asyncio +async def test_yandex_scraper_exception_releases_lease() -> None: + from scraper_kit.providers.yandex.valuation import YandexValuationScraper + + cfg = _FakeConfig(use_proxy_pool_curl=True, scraper_proxy_url=None) + spy = _SpyProvider(_LEASE) + with patch("scraper_kit.providers.yandex.valuation._CurlCffiSession") as sess_cls: + sess_cls.return_value = MagicMock(close=AsyncMock()) + with pytest.raises(RuntimeError, match="scrape blew up"): + async with YandexValuationScraper(cfg, proxy_provider=spy): + raise RuntimeError("scrape blew up") + # исключение в теле → ok=False + release (lease не течёт) + assert spy.mark_health_calls == [(7, False)] + assert spy.release_calls == [7] diff --git a/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py b/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py index 93352f8b..3a8a1585 100644 --- a/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py +++ b/tradein-mvp/backend/tests/test_scraper_adapters_contracts.py @@ -15,6 +15,7 @@ from __future__ import annotations from scraper_kit.contracts import ( EnrichmentJobs, HouseMatcher, + ProxyProvider, ScraperConfig, SessionFactory, ) @@ -22,6 +23,7 @@ from scraper_kit.contracts import ( from app.services.scraper_adapters import ( RealEnrichmentJobs, RealMatcherAdapter, + RealProxyProvider, RealScraperConfig, RealSessionFactory, ) @@ -43,6 +45,10 @@ def _check_enrichment(j: EnrichmentJobs) -> None: """Статическая проверка: RealEnrichmentJobs ⊑ EnrichmentJobs.""" +def _check_proxy_provider(p: ProxyProvider) -> None: + """Статическая проверка: RealProxyProvider ⊑ ProxyProvider.""" + + def test_matcher_adapter_satisfies_protocol() -> None: adapter = RealMatcherAdapter() _check_matcher(adapter) @@ -60,6 +66,17 @@ def test_scraper_config_satisfies_protocol() -> None: assert isinstance(config.browser_http_endpoint, str) assert isinstance(config.avito_proxy_max_rotations, int) assert isinstance(config.avito_serp_ekb_only, bool) + # #2163: новый флаг proxy-пула присутствует и bool. + assert isinstance(config.use_proxy_pool_curl, bool) + + +def test_proxy_provider_satisfies_protocol() -> None: + provider = RealProxyProvider() + _check_proxy_provider(provider) + assert isinstance(provider, ProxyProvider) + assert callable(provider.acquire) + assert callable(provider.release) + assert callable(provider.mark_health) def test_session_factory_satisfies_protocol() -> None: diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py index e2e065d9..290c0d34 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py @@ -23,12 +23,29 @@ from __future__ import annotations from collections.abc import Callable +from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable if TYPE_CHECKING: from sqlalchemy.orm import Session +@dataclass(frozen=True) +class ProxyLease: + """Арендованный из пула прокси (лёгкий транспорт между kit и backend). + + Структурно совпадает с `app.services.proxy_pool.ProxyLease` (id/url/kind/rotate_url), + но живёт в контрактах, чтобы kit не импортировал `app.*`. Провайдеры читают только + `url` (готов для curl_cffi `proxies=`); `id`/`rotate_url` нужны backend-адаптеру для + release / mark_health / (будущей) ротации. + """ + + id: int + url: str + kind: str + rotate_url: str | None = None + + @runtime_checkable class HouseMatcher(Protocol): """Кросс-источниковый матчинг домов/листингов (продуктовая логика matching-сервиса). @@ -134,6 +151,48 @@ class ScraperConfig(Protocol): scraper_skip_seen_today: bool # Per-fetch watchdog-таймаут Cian full-load (секунды). cian_full_load_per_fetch_timeout_s: float + # ── Proxy-pool (#2163) ──────────────────────────────────────────────────── + # Feature-флаг: брать прокси из пула (scrape_proxies) на curl_cffi-путях вместо + # env-прокси (config.*_proxy_url). Ship-dark: дефолт False → curl-пути ходят как + # сейчас. True + пустой пул → fallback на env (сбор не ломается). См. ProxyProvider. + use_proxy_pool_curl: bool + + +@runtime_checkable +class ProxyProvider(Protocol): + """Источник прокси из пула для curl-путей (#2163), за флагом use_proxy_pool_curl. + + Инкапсулирует lease/health-механику `app.services.proxy_pool` (которая требует + `Session`), пряча БД от kit'а. Реализуется + `app.services.scraper_adapters.RealProxyProvider` (открывает короткую сессию на + операцию). Провайдеры вызывают `acquire` → используют `lease.url` → на выходе + `mark_health` + `release` (в finally — lease не должен течь). + + Контракт ship-dark/fallback: `acquire` возвращает None если свободных здоровых + прокси нет — caller обязан откатиться на env-прокси, а НЕ падать. + """ + + def acquire(self, provider: str) -> ProxyLease | None: + """Взять прокси под провайдера (avito/cian/yandex/generic/any). + + Returns ProxyLease или None если пул пуст/выключен → caller делает env-fallback. + """ + ... + + def release(self, lease: ProxyLease) -> None: + """Освободить lease (идемпотентно). Вызывать в finally — иначе прокси течёт.""" + ... + + def mark_health( + self, + lease: ProxyLease, + ok: bool, + *, + exit_ip: str | None = ..., + latency_ms: int | None = ..., + ) -> None: + """Записать исход использования прокси (ok=True успех, False бан/ошибка).""" + ... @runtime_checkable @@ -242,6 +301,8 @@ class EnrichmentJobs(Protocol): __all__ = [ "EnrichmentJobs", "HouseMatcher", + "ProxyLease", + "ProxyProvider", "ScraperConfig", "SessionFactory", ] diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/_proxy.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/_proxy.py new file mode 100644 index 00000000..5556a17b --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/_proxy.py @@ -0,0 +1,83 @@ +"""Общий helper выбора прокси для curl_cffi-путей (#2163), за флагом USE_PROXY_POOL_CURL. + +Инвариант ship-dark + fallback: + - config.use_proxy_pool_curl=False (дефолт) ИЛИ proxy_provider=None → yield env-прокси + (env_fallback_url) — curl-пути ходят ровно как сейчас, прод не меняется. + - Флаг on + пул выдал lease → yield lease.url; на выходе mark_health(ok) + release(lease). + ok=True если блок отработал без исключения, ok=False если внутри поднялось (бан/ошибка). + - Флаг on + пул пуст/ошибка acquire → fallback на env_fallback_url, НЕ падаем (сбор цел). + +release ВСЕГДА в finally — lease не должен течь, даже если fetch кинул. mark_health/release +обёрнуты в best-effort try (проблема пула не должна ронять сбор). +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterator +from contextlib import contextmanager +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from scraper_kit.contracts import ProxyProvider, ScraperConfig + +logger = logging.getLogger(__name__) + + +@contextmanager +def curl_proxy_url( + config: ScraperConfig | None, + proxy_provider: ProxyProvider | None, + provider: str, + *, + env_fallback_url: str | None, +) -> Iterator[str | None]: + """Отдать effective proxy-url для curl_cffi (`proxies={http/https: url}`). + + Args: + config: ScraperConfig (читается флаг use_proxy_pool_curl). None → env-fallback. + proxy_provider: пул прокси или None (off). None → env-fallback. + provider: имя провайдера для affinity ("avito"/"cian"/"yandex"/...). + env_fallback_url: прокси-url как сейчас (config.cian_proxy_url / scraper_proxy_url). + + Yields effective url (может быть None = прямое подключение, как и раньше). + """ + use_pool = ( + config is not None + and getattr(config, "use_proxy_pool_curl", False) + and proxy_provider is not None + ) + + lease = None + if use_pool: + assert proxy_provider is not None # для type-narrowing (use_pool это гарантирует) + try: + lease = proxy_provider.acquire(provider) + except Exception: + logger.warning( + "proxy_pool: acquire(%s) failed — fallback to env proxy", provider, exc_info=True + ) + lease = None + + if lease is None: + # off / пул пуст / ошибка acquire → env как сейчас (сбор не ломаем). + yield env_fallback_url + return + + assert proxy_provider is not None + ok = True + try: + yield lease.url + except Exception: + ok = False + raise + finally: + # mark_health + release — best-effort: проблема пула не должна ронять сбор. + try: + proxy_provider.mark_health(lease, ok) + except Exception: + logger.warning("proxy_pool: mark_health failed for %s", provider, exc_info=True) + try: + proxy_provider.release(lease) # ОБЯЗАТЕЛЬНО — lease не течёт + except Exception: + logger.warning("proxy_pool: release failed for %s", provider, exc_info=True) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/imv.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/imv.py index b5c321a7..69173da2 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/imv.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/imv.py @@ -23,6 +23,7 @@ import hashlib import json import logging import re +from contextlib import ExitStack from dataclasses import dataclass, field from datetime import date from typing import TYPE_CHECKING, Any @@ -32,6 +33,7 @@ from uuid import UUID from sqlalchemy import text from sqlalchemy.orm import Session +from scraper_kit.providers._proxy import curl_proxy_url from scraper_kit.providers.avito.shared import _unix_to_date if TYPE_CHECKING: @@ -39,7 +41,7 @@ if TYPE_CHECKING: # hard import cycle browser_fetcher → config → ... и лишней зависимости в # offline-тестах парсинга, которым browser не нужен. from scraper_kit.browser_fetcher import BrowserFetcher - from scraper_kit.contracts import ScraperConfig + from scraper_kit.contracts import ProxyProvider, ScraperConfig logger = logging.getLogger(__name__) @@ -437,6 +439,7 @@ async def evaluate_via_imv( cffi_session: Any | None = None, browser_fetcher: BrowserFetcher | None = None, config: ScraperConfig | None = None, + proxy_provider: ProxyProvider | None = None, ) -> IMVEvaluation: """Avito IMV — 3 HTTP requests (current contract 2026-05+): @@ -458,55 +461,65 @@ async def evaluate_via_imv( # (#562/#853). _own_session=True → finally дёрнет adapter.close() (no-op, # браузер принадлежит caller'у — backfill открывает один на батч). Этот путь # НЕ требует curl_cffi, поэтому импорт пакета остаётся только в else-ветке. - if browser_fetcher is not None: - cffi_session = _BrowserSessionAdapter(browser_fetcher, origin=WARMUP_URL) - _own_session = True - else: - # Импортируем здесь чтобы избежать циклических зависимостей при тестах без сети + # ExitStack держит прокси-lease пула (#2163) на всё время own-session: warm-up + + # geocode + evaluate. Регистрируется ТОЛЬКО когда сами создаём curl_cffi-сессию; + # для browser_fetcher / переданной cffi_session — no-op. Исключение из блока + # (бан/ошибка) прокинется в curl_proxy_url.__exit__ → mark_health(ok=False) + release; + # чистый выход → mark_health(ok=True) + release. Прокси-переключение не трогает + # warm-up/cookie-последовательность (сессия создаётся один раз, до _warmup). + with ExitStack() as _proxy_stack: + if browser_fetcher is not None: + cffi_session = _BrowserSessionAdapter(browser_fetcher, origin=WARMUP_URL) + _own_session = True + else: + # Импортируем здесь чтобы избежать циклических зависимостей при тестах без сети + try: + from curl_cffi.requests import AsyncSession as CffiAsyncSession + + _own_session = False + if cffi_session is None: + # Зеркалим production-набор из AvitoScraper.__aenter__ (avito.py:150-162): + # chrome120 TLS + document-заголовки + timeout. Затем warm-up GET для + # seed anti-bot cookies — bare-session XHR Avito банит на server-IP. + # Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env + # scraper_proxy_url. proxy=None → прямое подключение (dev). + _env = config.scraper_proxy_url if config is not None else None + _proxy_url = _proxy_stack.enter_context( + curl_proxy_url(config, proxy_provider, "avito", env_fallback_url=_env) + ) + _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None + cffi_session = CffiAsyncSession( + impersonate="chrome120", + timeout=_HTTP_TIMEOUT_SEC, + proxies=_proxies, + headers=_DOC_HEADERS, + ) + _own_session = True + except ImportError as exc: + raise RuntimeError( + "curl_cffi не установлен. Добавь 'curl-cffi>=0.7.0' в pyproject.toml." + ) from exc + try: - from curl_cffi.requests import AsyncSession as CffiAsyncSession - - _own_session = False - if cffi_session is None: - # Зеркалим production-набор из AvitoScraper.__aenter__ (avito.py:150-162): - # chrome120 TLS + document-заголовки + timeout. Затем warm-up GET для - # seed anti-bot cookies — bare-session XHR Avito банит на server-IP. - # Mobile proxy wiring (#806 follow-up): только для own-session (переданная - # сессия уже с прокси от AvitoScraper). proxy=None → прямое подключение. - _proxy_url = config.scraper_proxy_url if config is not None else None - _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None - cffi_session = CffiAsyncSession( - impersonate="chrome120", - timeout=_HTTP_TIMEOUT_SEC, - proxies=_proxies, - headers=_DOC_HEADERS, - ) - _own_session = True - except ImportError as exc: - raise RuntimeError( - "curl_cffi не установлен. Добавь 'curl-cffi>=0.7.0' в pyproject.toml." - ) from exc - - try: - if _own_session: - await _warmup(cffi_session) - geo = await _geocode(cffi_session, address) - evaluation = await _imv_evaluate( - cffi_session, - geo=geo, - address=address, - rooms=rooms, - area_m2=area_m2, - floor=floor, - floor_at_home=floor_at_home, - house_type=house_type, - renovation_type=renovation_type, - has_balcony=has_balcony, - has_loggia=has_loggia, - ) - finally: - if _own_session: - await cffi_session.close() + if _own_session: + await _warmup(cffi_session) + geo = await _geocode(cffi_session, address) + evaluation = await _imv_evaluate( + cffi_session, + geo=geo, + address=address, + rooms=rooms, + area_m2=area_m2, + floor=floor, + floor_at_home=floor_at_home, + house_type=house_type, + renovation_type=renovation_type, + has_balcony=has_balcony, + has_loggia=has_loggia, + ) + finally: + if _own_session: + await cffi_session.close() return evaluation diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py index 0152cdd0..573ce47b 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/detail.py @@ -22,6 +22,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from scraper_kit.cian_state_parser import extract_all_states, extract_state +from scraper_kit.providers._proxy import curl_proxy_url from scraper_kit.repair_state_normalizer import ( infer_repair_state_from_text, normalize_repair_state, @@ -30,7 +31,7 @@ from scraper_kit.snapshot_writer import upsert_listing_snapshot if TYPE_CHECKING: from scraper_kit.browser_fetcher import BrowserFetcher - from scraper_kit.contracts import ScraperConfig + from scraper_kit.contracts import ProxyProvider, ScraperConfig logger = logging.getLogger(__name__) @@ -75,6 +76,7 @@ async def fetch_detail( config: ScraperConfig | None = None, session: AsyncSession | None = None, browser_fetcher: BrowserFetcher | None = None, + proxy_provider: ProxyProvider | None = None, ) -> DetailEnrichment | None: """Fetch detail page and extract all containers. @@ -99,15 +101,22 @@ async def fetch_detail( except Exception as exc: logger.warning("Cian detail browser fetch failed %s: %s", offer_url, exc) return None + elif session is not None: + # Shared curl_cffi-сессия (прокси уже применён caller'ом) — пул не трогаем. + resp = await session.get(offer_url, allow_redirects=True) + if resp.status_code != 200: + logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code) + return None + html = resp.text else: - # curl_cffi path (legacy, back-compat). - close_session = False - if session is None: - # proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian. - # Пусто (config не передан / env не задан) → прямое подключение (dev/no-op). - _proxy_url = config.cian_proxy_url if config is not None else None + # curl_cffi own-session path (legacy, back-compat). + # proxies: mobile-proxy egress (#806) — без прокси datacenter-IP блокируется Cian. + # Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env cian_proxy_url. + # Пусто → прямое подключение (dev/no-op). curl_proxy_url: mark_health + release на выходе. + _env = config.cian_proxy_url if config is not None else None + with curl_proxy_url(config, proxy_provider, "cian", env_fallback_url=_env) as _proxy_url: _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None - session = AsyncSession( + own_session = AsyncSession( impersonate="chrome120", timeout=25.0, proxies=_proxies, @@ -116,17 +125,14 @@ async def fetch_detail( "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", }, ) - close_session = True - - try: - resp = await session.get(offer_url, allow_redirects=True) - if resp.status_code != 200: - logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code) - return None - html = resp.text - finally: - if close_session: - await session.close() + try: + resp = await own_session.get(offer_url, allow_redirects=True) + if resp.status_code != 200: + logger.warning("Cian detail fetch %s → HTTP %d", offer_url, resp.status_code) + return None + html = resp.text + finally: + await own_session.close() # ── Shared parse path (identical regardless of HTML source) ────────────── diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py index 11a18eea..fed6fc27 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/valuation.py @@ -32,10 +32,11 @@ from sqlalchemy import text from sqlalchemy.orm import Session from scraper_kit.cian_state_parser import extract_state +from scraper_kit.providers._proxy import curl_proxy_url from scraper_kit.providers.cian.session import load_session, mark_session_invalid if TYPE_CHECKING: - from scraper_kit.contracts import ScraperConfig + from scraper_kit.contracts import ProxyProvider, ScraperConfig logger = logging.getLogger(__name__) @@ -142,6 +143,7 @@ async def estimate_via_cian_valuation( use_cache: bool = True, house_id: int | None = None, listing_id: int | None = None, + proxy_provider: ProxyProvider | None = None, ) -> CianValuationResult | None: """Estimate value via Cian's authenticated Valuation Calculator. @@ -179,25 +181,30 @@ async def estimate_via_cian_valuation( # 4. Fetch с TLS fingerprint (curl_cffi, impersonate chrome120) # Mobile proxy wiring (#806 follow-up): Cian валюация — такой же datacenter-бан риск - # как SERP. Используем cian_proxy_url. proxy=None → прямое подключение (dev). - _proxy_url = config.cian_proxy_url - _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None - try: - async with AsyncSession( - impersonate="chrome120", - cookies=cookies, - timeout=25.0, - proxies=_proxies, - headers={"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"}, - ) as session: - resp = await session.get(url, allow_redirects=True) - if resp.status_code != 200: - logger.warning("Cian valuation fetch → HTTP %d for %s", resp.status_code, url) - return None - html = resp.text - except Exception as exc: - logger.warning("Cian valuation fetch failed: %s", exc) - return None + # как SERP. Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env cian_proxy_url. + # proxy=None → прямое подключение (dev). curl_proxy_url на выходе mark_health + release. + with curl_proxy_url( + config, proxy_provider, "cian", env_fallback_url=config.cian_proxy_url + ) as _proxy_url: + _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None + try: + async with AsyncSession( + impersonate="chrome120", + cookies=cookies, + timeout=25.0, + proxies=_proxies, + headers={ + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" + }, + ) as session: + resp = await session.get(url, allow_redirects=True) + if resp.status_code != 200: + logger.warning("Cian valuation fetch → HTTP %d for %s", resp.status_code, url) + return None + html = resp.text + except Exception as exc: + logger.warning("Cian valuation fetch failed: %s", exc) + return None # 5. Parse state из _cianConfig['valuation-for-agent-frontend'] state = extract_state(html, mfe=VALUATION_MFE, key="initialState") diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py index 2d4d9825..0c8c883c 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/valuation.py @@ -25,6 +25,7 @@ from __future__ import annotations import logging import re from collections.abc import Callable +from contextlib import ExitStack from datetime import date from typing import TYPE_CHECKING, Any from urllib.parse import urlencode @@ -34,6 +35,7 @@ from pydantic import BaseModel, Field from selectolax.parser import HTMLParser from scraper_kit.base import BaseScraper +from scraper_kit.providers._proxy import curl_proxy_url from scraper_kit.yandex_helpers import ( parse_dmy, parse_house_type, @@ -41,7 +43,7 @@ from scraper_kit.yandex_helpers import ( ) if TYPE_CHECKING: - from scraper_kit.contracts import ScraperConfig + from scraper_kit.contracts import ProxyProvider, ScraperConfig logger = logging.getLogger(__name__) @@ -170,6 +172,7 @@ class YandexValuationScraper(BaseScraper): config: ScraperConfig, *, delay_provider: Callable[[str], float] | None = None, + proxy_provider: ProxyProvider | None = None, ) -> None: super().__init__() # Strangler-инжекция (#2133): конфиг (прокси-egress) и провайдер задержки @@ -179,22 +182,47 @@ class YandexValuationScraper(BaseScraper): if delay_provider is not None: self.request_delay_sec = delay_provider(self.name) self._cffi_session: _CurlCffiSession | None = None + # Прокси-пул (#2163): lease держится на всё время сессии (__aenter__→__aexit__). + self._proxy_provider = proxy_provider + self._proxy_stack: ExitStack | None = None async def __aenter__(self) -> YandexValuationScraper: # type: ignore[override] # Override: Yandex valuation endpoint gates SSR data on Chrome TLS # fingerprint. Plain httpx returns shell HTML (CSR-only). # Sibling: yandex_realty.py / scripts/local-sweep-ekb-yandex.py - # Mobile proxy wiring (#806 follow-up): route via mobile proxy to avoid - # datacenter-IP blocks. proxy=None → прямое подключение (dev). - _proxy_url = self._config.scraper_proxy_url + # Прокси: пул за флагом use_proxy_pool_curl (#2163), иначе env scraper_proxy_url. + # proxy=None → прямое подключение (dev). lease освобождается в __aexit__. + self._proxy_stack = ExitStack() + _proxy_url = self._proxy_stack.enter_context( + curl_proxy_url( + self._config, + self._proxy_provider, + "yandex", + env_fallback_url=self._config.scraper_proxy_url, + ) + ) _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None - self._cffi_session = _CurlCffiSession(impersonate="chrome120", proxies=_proxies) + try: + self._cffi_session = _CurlCffiSession(impersonate="chrome120", proxies=_proxies) + except Exception: + # session-создание упало → не течём lease'ом + self._proxy_stack.close() + self._proxy_stack = None + raise return self async def __aexit__(self, *args: object) -> None: # type: ignore[override] if self._cffi_session is not None: await self._cffi_session.close() self._cffi_session = None + # Прокинуть инфо об исключении в curl_proxy_url.__exit__ → mark_health(ok)+release. + if self._proxy_stack is not None: + exc = args[1] if len(args) > 1 and isinstance(args[1], BaseException) else None + if exc is not None: + self._proxy_stack.__exit__(type(exc), exc, exc.__traceback__) + else: + self._proxy_stack.close() + self._proxy_stack = None async def _http_get(self, url: str, **kwargs: object) -> object: # type: ignore[override] """curl_cffi-based GET with Chrome120 impersonation.