fix(tradein/estimator): yandex/cian valuation в /estimate через пул прокси, пустой пул не роняет ответ (#3398)

Тот же корень, что у IMV (#3386 / PR #3397): оба вызова шли БЕЗ `proxy_provider`
— `YandexValuationScraper(RealScraperConfig(), delay_provider=...)` и
`estimate_via_cian_valuation(**_c_kwargs)`. `providers/_proxy.py::curl_proxy_url`
считает `use_pool = флаг AND provider is not None`, поэтому пул был выключен по
построению, а curl уходил на env-прокси SCRAPER_PROXY_URL (выключенный узел
#2613): на проде 06.09 каждая проба `/estimate` давала `curl_cffi ProxyError:
CONNECT tunnel failed, response 407`, а в логах это читалось как
«yandex_valuation: empty result» — оба источника мертвы с 02.09.

Kit-стороне правки не нужны: `YandexValuationScraper.__init__` и
`estimate_via_cian_valuation` уже принимают kwarg-only `proxy_provider` и сами
зовут `curl_proxy_url` (yandex — lease на сессию __aenter__/__aexit__, cian —
lease на вызов, release в finally). Провайдер берётся из module-level импорта
`RealProxyProvider` (он stateless: короткая сессия на операцию), поэтому один
инстанс в `_c_kwargs` покрывает и основной вызов, и отложенную фоновую догрузку.

Мягкая деградация сохранена и стала честной в логах: `NoProxyAvailableError`
(проверка по цепочке причин `caused_by_no_proxy`) → WARNING «пул прокси пуст —
продолжаем без Yandex/Cian» и прежний None-путь, а не ERROR «fetch failed» /
«lookup failed» (запрос вообще не уходил — GlitchTip-событие тут было бы шумом).
Строка «yandex_valuation: empty result» переписана: None неразличимо «дом не
найден» и «фетч не дошёл», причина — в строке scraper_kit выше.

Тесты (7, по значению; на main все красные): оба call site получают провайдера;
пустой пул в production → оценка без этих источников, без исключения, HTTP не
уходит, в логе «пул прокси пуст»; lease освобождён ровно один раз на успехе и на
ошибке фетча — через настоящий `curl_proxy_url`, а не мок провайдера.

tradein-mvp/backend/app/services/estimator.py

@@ -1006,14 +1006,27 @@ async def _get_or_fetch_yandex_valuation_cached(
     # Fresh fetch
     try:
         async with YandexValuationScraper(
-            RealScraperConfig(), delay_provider=get_scraper_delay
+            RealScraperConfig(),
+            delay_provider=get_scraper_delay,
+            # #3398: без provider'а `curl_proxy_url` считает use_pool=False (флаг AND
+            # provider is not None) — пул выключен по построению, curl-сессия уходит на
+            # env-прокси SCRAPER_PROXY_URL (мёртвый узел #2613 → 407 на каждой пробе).
+            # Lease один на сессию: acquire в __aenter__, release в __aexit__ — в т.ч.
+            # на исключении и на отмене по таймауту _with_budget.
+            proxy_provider=RealProxyProvider(),
         ) as scraper:
             result = await scraper.fetch_house_history(
                 address=address,
                 offer_category=offer_category,
                 offer_type=offer_type,
             )
-    except Exception:
+    except Exception as e:
+        if caused_by_no_proxy(e):
+            # #3398: «пул прокси пуст» — НАША инфраструктура, HTTP-запрос не уходил вовсе.
+            # Отдельный текст (и WARNING, не ERROR): это не сбой Яндекса, GlitchTip-событие
+            # здесь было бы шумом. Возврат None — тот же graceful путь, /estimate не 5xx.
+            logger.warning("yandex_valuation: пул прокси пуст — продолжаем без Yandex: %s", e)
+            return None
         # logger.exception (не .warning) — намеренно: GlitchTip LoggingIntegration
         # (main.py/scheduler_main.py) слушает event_level=logging.ERROR. WARNING,
         # даже с exc_info=True, остаётся ниже порога и НЕ создаёт событие в GlitchTip
@@ -1023,7 +1036,15 @@ async def _get_or_fetch_yandex_valuation_cached(
         return None
 
     if result is None:
-        logger.info("yandex_valuation: empty result for address=%s", address[:60])
+        # #3398: None здесь НЕразличимо «дом не найден» и «фетч не дошёл» — kit ловит
+        # ProxyError/HTTP-код внутри fetch_house_history и логирует своей строкой
+        # («yandex valuation fetch failed» / «returned NNN»). Прежний текст «empty result»
+        # читался как «у Яндекса нет данных», хотя при мёртвом прокси это был 407.
+        logger.info(
+            "yandex_valuation: пусто для address=%s — данных нет ЛИБО фетч не дошёл "
+            "(причина строкой выше от scraper_kit: прокси/HTTP)",
+            address[:60],
+        )
         return None
 
     # Save to cache (UPSERT on (source, cache_key))
@@ -4624,6 +4645,13 @@ async def estimate_quality(
             "deal_type": "sale",
             "use_cache": True,
             "house_id": target_house_id,
+            # #3398: без provider'а curl_proxy_url считает use_pool=False (флаг AND
+            # provider is not None) → env-прокси CIAN_PROXY_URL/SCRAPER_PROXY_URL, мёртвый
+            # узел (#2613) → «Cian valuation fetch failed: … 407». Провайдер stateless
+            # (сессия на операцию), поэтому один инстанс на оба call site: основной вызов
+            # и отложенная фоновая догрузка ниже. Lease живёт внутри curl_proxy_url:
+            # acquire до запроса, release в finally.
+            "proxy_provider": RealProxyProvider(),
         }
         try:
             cian_val = await _with_budget(
@@ -4643,13 +4671,19 @@ async def estimate_quality(
                     cian_val.sale_accuracy,
                     cian_val.external_house_id,
                 )
-        except Exception:
-            # logger.exception (не .warning) — намеренно: GlitchTip LoggingIntegration
-            # (main.py/scheduler_main.py) слушает event_level=logging.ERROR. WARNING,
-            # даже с exc_info=True, остаётся ниже порога и НЕ создаёт событие в GlitchTip
-            # (только breadcrumb) — config-wiring регрессия здесь была бы не видна
-            # мониторингу. .exception() логирует на ERROR + traceback (#2337).
-            logger.exception("cian_valuation: lookup failed (graceful)")
+        except Exception as e:
+            if caused_by_no_proxy(e):
+                # #3398: пустой пул поднимает NoProxyAvailableError ДО HTTP (curl_proxy_url),
+                # то есть это не сбой Циана. WARNING, не ERROR: в GlitchTip такое событие —
+                # шум. Оценка отдаётся без cian-источника, /estimate не 5xx.
+                logger.warning("cian_valuation: пул прокси пуст — продолжаем без Cian: %s", e)
+            else:
+                # logger.exception (не .warning) — намеренно: GlitchTip LoggingIntegration
+                # (main.py/scheduler_main.py) слушает event_level=logging.ERROR. WARNING,
+                # даже с exc_info=True, остаётся ниже порога и НЕ создаёт событие в GlitchTip
+                # (только breadcrumb) — config-wiring регрессия здесь была бы не видна
+                # мониторингу. .exception() логирует на ERROR + traceback (#2337).
+                logger.exception("cian_valuation: lookup failed (graceful)")
 
     # ── Pre-fetch: same-building anchor comps ─────────────────────────────────
     # Guard mirrors the original in-block guard exactly; when false → ([], None).

tradein-mvp/backend/tests/test_3398_estimator_valuations_proxy_pool.py

@@ -0,0 +1,331 @@
+"""Yandex/Cian valuation в /estimate идут через пул прокси, пустой пул не роняет ответ (#3398).
+
+Корень тот же, что у IMV (#3386/PR #3397): `estimator` звал kit БЕЗ `proxy_provider` —
+`YandexValuationScraper(RealScraperConfig(), delay_provider=...)` и
+`estimate_via_cian_valuation(**_c_kwargs)`. `providers/_proxy.py::curl_proxy_url` считает
+`use_pool = флаг AND proxy_provider is not None`, поэтому пул был выключен ПО ПОСТРОЕНИЮ, а
+curl уходил на env-прокси `SCRAPER_PROXY_URL` (выключенный узел, #2613) → на проде каждая
+проба `/estimate` давала `407 CONNECT tunnel failed` и «yandex_valuation: empty result».
+
+Проверки по значению:
+  (а) оба call site передают provider'а (на main здесь None);
+  (б) пул пуст + environment=production → оценка отдаётся БЕЗ этих источников, без
+      исключения, HTTP не уходит, а причина в логе честная — «пул прокси пуст»;
+  (в) lease освобождён ровно один раз (успех и ошибка фетча) — через настоящий
+      `curl_proxy_url`, а не через мок провайдера.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from contextlib import ExitStack
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import anyio
+import pytest
+from scraper_kit.contracts import ProxyLease
+
+from app.core.config import settings
+from app.schemas.trade_in import TradeInEstimateInput
+from app.services import estimator
+from app.services.estimator import _get_or_fetch_yandex_valuation_cached, estimate_quality
+from app.services.geocoder import GeocodeResult
+
+_ADDRESS = "Первоуральск, ул. Ленина, 5"
+
+
+# ── провайдеры-дублёры ──────────────────────────────────────────────────────
+
+
+class _EmptyPoolProvider:
+    """acquire → None (все узлы забанены/заняты)."""
+
+    def acquire(self, provider: str) -> ProxyLease | None:
+        return None
+
+    def release(self, lease: ProxyLease) -> None:  # pragma: no cover — не должно звучать
+        raise AssertionError("release без lease")
+
+    def mark_health(self, lease: ProxyLease, ok: bool, **kw: Any) -> None:  # pragma: no cover
+        raise AssertionError("mark_health без lease")
+
+    def mark_banned(self, lease: ProxyLease, *, source: str) -> None:  # pragma: no cover
+        raise AssertionError("mark_banned без lease")
+
+
+class _CountingProvider:
+    def __init__(self) -> None:
+        self.acquired: list[str] = []
+        self.released: list[int] = []
+        self.health: list[bool] = []
+
+    def acquire(self, provider: str) -> ProxyLease | None:
+        self.acquired.append(provider)
+        return ProxyLease(id=7, url="http://pool-node:3128", kind="datacenter", rotate_url=None)
+
+    def release(self, lease: ProxyLease) -> None:
+        self.released.append(lease.id)
+
+    def mark_health(self, lease: ProxyLease, ok: bool, **kw: Any) -> None:
+        self.health.append(ok)
+
+    def mark_banned(self, lease: ProxyLease, *, source: str) -> None:
+        pass
+
+
+def _no_http(*a: Any, **kw: Any) -> Any:  # pragma: no cover — не должно вызваться
+    raise AssertionError("HTTP-запрос при пустом пуле — прокси-гейт не сработал")
+
+
+class _FakeResponse:
+    def __init__(self, status_code: int, text: str) -> None:
+        self.status_code = status_code
+        self.text = text
+
+
+def _fake_curl_session(*, get: Any) -> Any:
+    """Заглушка curl_cffi.AsyncSession: и как async-CM (cian), и как объект (yandex)."""
+    session = MagicMock()
+    session.get = get
+    session.close = AsyncMock()
+    session.__aenter__ = AsyncMock(return_value=session)
+    session.__aexit__ = AsyncMock(return_value=False)
+    return lambda *a, **kw: session
+
+
+def _use_pool_in_production(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(settings, "use_proxy_pool_curl", True)
+    monkeypatch.setattr(settings, "environment", "production")
+    # fetch_on_miss=False (фоновый режим) вернул бы None ДО прокси-слоя — тест
+    # проверял бы тогда пустоту кэша, а не проводку пула.
+    monkeypatch.setattr(settings, "estimate_external_sources_background", False)
+
+
+# ── (а) yandex: провайдер доезжает до скрейпера ──────────────────────────────
+
+
+def _db_cache_miss() -> MagicMock:
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.first.return_value = None
+    return db
+
+
+def _call_yandex(db: Any) -> Any:
+    async def _run() -> Any:
+        return await _get_or_fetch_yandex_valuation_cached(db, address=_ADDRESS)
+
+    return anyio.run(_run)
+
+
+def test_yandex_valuation_call_passes_proxy_provider() -> None:
+    """YandexValuationScraper получает proxy_provider (на main здесь None → пул off)."""
+    captured: dict[str, Any] = {}
+
+    def _fake_scraper_cls(*args: Any, **kwargs: Any) -> Any:
+        captured.update(kwargs)
+        scraper = MagicMock()
+        scraper.fetch_house_history = AsyncMock(return_value=None)
+        scraper.__aenter__ = AsyncMock(return_value=scraper)
+        scraper.__aexit__ = AsyncMock(return_value=False)
+        return scraper
+
+    with patch.object(estimator, "YandexValuationScraper", _fake_scraper_cls):
+        assert _call_yandex(_db_cache_miss()) is None
+
+    assert captured, "скрейпер не создавался — тест ничего не проверил"
+    assert captured.get("proxy_provider") is not None
+
+
+# ── (а) cian: провайдер доезжает до kit-функции ──────────────────────────────
+
+
+def _geo() -> GeocodeResult:
+    return GeocodeResult(lat=56.909, lon=59.960, full_address=_ADDRESS, provider="nominatim")
+
+
+def _listing(i: int) -> dict[str, Any]:
+    return {
+        "source": "avito",
+        "source_url": f"https://avito.ru/offer/{i}",
+        "address": f"д. {i}",
+        "lat": 56.909,
+        "lon": 59.960,
+        "rooms": 2,
+        "area_m2": 45.0,
+        "floor": 5,
+        "total_floors": 9,
+        "price_rub": (200_000.0 + i * 1_000) * 45.0,
+        "price_per_m2": 200_000.0 + i * 1_000,
+        "listing_date": datetime(2026, 5, 1, tzinfo=UTC),
+        "days_on_market": 10,
+        "photo_urls": [],
+        "scraped_at": datetime(2026, 5, 20, tzinfo=UTC),
+        "distance_m": 150.0,
+        "relevance_score": 0.1,
+    }
+
+
+def _run_estimate(*, extra_patches: list[Any]) -> Any:
+    """estimate_quality() со всеми внешними источниками кроме Cian заглушенными.
+
+    Cian намеренно НЕ патчится списком по умолчанию: тесты (б)/(в) гоняют настоящую
+    kit-функцию, чтобы прокси-слой (`curl_proxy_url`) реально отработал.
+    """
+    payload = TradeInEstimateInput(address=_ADDRESS, area_m2=45.0, rooms=2, floor=5, total_floors=9)
+
+    patches = [
+        patch("app.services.estimator.geocode", new=AsyncMock(return_value=_geo())),
+        patch("app.services.estimator.dadata_clean_address", new=AsyncMock(return_value=None)),
+        patch("app.services.estimator.match_house_readonly", return_value=None),
+        patch("app.services.estimator.get_house_metadata", new=AsyncMock(return_value=None)),
+        patch(
+            "app.services.estimator._fetch_analogs",
+            return_value=([_listing(i) for i in range(8)], False, "W"),
+        ),
+        patch("app.services.estimator._fetch_anchor_comps", return_value=([], None)),
+        patch("app.services.estimator._fetch_deals", return_value=[]),
+        patch("app.services.estimator._get_or_fetch_imv_cached", new=AsyncMock(return_value=None)),
+        patch(
+            "app.services.estimator._get_or_fetch_yandex_valuation_cached",
+            new=AsyncMock(return_value=None),
+        ),
+        patch("app.services.estimator._fetch_dkp_corridor", return_value=None),
+        patch("app.services.estimator._get_asking_sold_ratio", return_value=(None, None)),
+        *extra_patches,
+    ]
+
+    async def _run() -> Any:
+        with ExitStack() as stack:  # список патчей переменной длины — не `with (...)`
+            for p in patches:
+                stack.enter_context(p)
+            return await estimate_quality(payload, MagicMock())
+
+    return anyio.run(_run)
+
+
+def test_cian_valuation_call_passes_proxy_provider(monkeypatch: pytest.MonkeyPatch) -> None:
+    """estimate_via_cian_valuation получает proxy_provider (на main здесь None)."""
+    _use_pool_in_production(monkeypatch)
+    captured: dict[str, Any] = {}
+
+    async def _fake_cian(_db: Any, **kwargs: Any) -> None:
+        captured.update(kwargs)
+        return None
+
+    est = _run_estimate(
+        extra_patches=[patch("app.services.estimator.estimate_via_cian_valuation", new=_fake_cian)]
+    )
+    assert est is not None
+    assert captured, "cian-вызов не состоялся — тест ничего не проверил"
+    assert captured.get("proxy_provider") is not None
+
+
+# ── (б) пустой пул в проде: деградация, а не 5xx ─────────────────────────────
+
+
+def test_yandex_empty_pool_in_production_degrades(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Пул пуст → None (ответ без Yandex), HTTP не уходит, причина в логе честная."""
+    _use_pool_in_production(monkeypatch)
+    result: Any = "unset"
+
+    with (
+        # Без create=True: RealProxyProvider импортирован в estimator — патч обязан
+        # бить по РЕАЛЬНОМУ атрибуту, иначе переименование импорта оставит тест зелёным.
+        patch.object(estimator, "RealProxyProvider", _EmptyPoolProvider),
+        patch("scraper_kit.providers.yandex.valuation._CurlCffiSession", _no_http),
+        caplog.at_level(logging.WARNING, logger="app.services.estimator"),
+    ):
+        result = _call_yandex(_db_cache_miss())
+
+    assert result is None
+    assert "пул прокси пуст" in caplog.text, caplog.text
+    assert "fetch failed" not in caplog.text, "причина подменена на неспецифичную"
+
+
+def test_cian_empty_pool_in_production_degrades(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Тот же контракт для Cian: оценка возвращается, исключение не всплывает наружу."""
+    _use_pool_in_production(monkeypatch)
+
+    with (
+        patch.object(estimator, "RealProxyProvider", _EmptyPoolProvider),
+        patch("scraper_kit.providers.cian.valuation._load_from_cache", return_value=None),
+        patch(
+            "scraper_kit.providers.cian.valuation.load_session",
+            return_value={"cookie": "x"},
+        ),
+        patch("scraper_kit.providers.cian.valuation.AsyncSession", _no_http),
+        caplog.at_level(logging.WARNING, logger="app.services.estimator"),
+    ):
+        est = _run_estimate(extra_patches=[])
+
+    assert est is not None  # /estimate отвечает, просто без cian-источника
+    assert "пул прокси пуст" in caplog.text, caplog.text
+    assert "lookup failed" not in caplog.text, "причина подменена на неспецифичную"
+
+
+# ── (в) lease освобождён ровно один раз, через настоящий curl_proxy_url ──────
+
+
+def _yandex_with_pool(monkeypatch: pytest.MonkeyPatch, *, get: Any) -> _CountingProvider:
+    _use_pool_in_production(monkeypatch)
+    provider = _CountingProvider()
+    with (
+        patch.object(estimator, "RealProxyProvider", lambda: provider),
+        patch(
+            "scraper_kit.providers.yandex.valuation._CurlCffiSession", _fake_curl_session(get=get)
+        ),
+    ):
+        _call_yandex(_db_cache_miss())
+    return provider
+
+
+def test_yandex_lease_released_once_on_success(monkeypatch: pytest.MonkeyPatch) -> None:
+    provider = _yandex_with_pool(
+        monkeypatch, get=AsyncMock(return_value=_FakeResponse(200, "<html></html>"))
+    )
+    assert provider.acquired == ["yandex"]
+    assert provider.released == [7]
+    assert provider.health == [True]
+
+
+def test_yandex_lease_released_once_on_fetch_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Ошибка транспорта (на проде — 407 от мёртвого env-прокси) не течёт lease'ом."""
+    provider = _yandex_with_pool(monkeypatch, get=AsyncMock(side_effect=OSError("proxy 407")))
+    assert provider.acquired == ["yandex"]
+    assert provider.released == [7]
+
+
+def test_cian_lease_released_once_on_fetch_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    _use_pool_in_production(monkeypatch)
+    provider = _CountingProvider()
+
+    with (
+        patch.object(estimator, "RealProxyProvider", lambda: provider),
+        patch("scraper_kit.providers.cian.valuation._load_from_cache", return_value=None),
+        patch(
+            "scraper_kit.providers.cian.valuation.load_session",
+            return_value={"cookie": "x"},
+        ),
+        patch(
+            "scraper_kit.providers.cian.valuation.AsyncSession",
+            _fake_curl_session(get=AsyncMock(side_effect=OSError("proxy 407"))),
+        ),
+    ):
+        est = _run_estimate(extra_patches=[])
+
+    assert est is not None
+    assert provider.acquired == ["cian"]
+    assert provider.released == [7]
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(pytest.main([__file__, "-q"]))