fix(tradein/cian): houses-фаза cian_city_sweep идёт через пул прокси + стоп на пустом пуле (#3394)

`run_cian_city_sweep` звал `fetch_newbuilding(zhk_url, config=config)` без `proxy_provider`, хотя провайдер лежит в аргументах самого свипа и соседние фазы (SERP через CianScraper, detail через cian_fetch_detail) его передают. Внутри это давало `build_browser_fetcher(config, "cian", proxy_provider=None)` → `use_pool` эффективно False → POST /fetch без "proxy" → сайдкар брал свой env-узел SCRAPER_PROXY_URL, на проде выключенный (407 → camoufox InvalidIP → /fetch 503, факт #3386). Фаза houses давала 0/30 с 02.09 (run 6179: 24 × `houses failed ... 503 Service Unavailable`), строк `override=True` в логах сайдкара по ней не было ни одной. Остальные вызывающие `fetch_newbuilding` / `resolve_cian_zhk_url_via_search` провайдер уже передают (#2767/#2830/#3382), этот вызов был последним мимо пула. Второе: пустой пул поднимается ДО запроса, следующий дом упрётся ровно в то же самое — общий `except Exception` на дом превращал это в 30 одинаковых houses_failed и прогон уходил в 'done'. Теперь NoProxyAvailableError рвёт фазу и свип: `no_proxy_stop=1` в counters, mark_banned с ban_kind='infra' и сохранённым done_buckets (образец — #3389 yandex-nb-sweep, #3382). Ветка стоит ДО generic-except, иначе наш отказ инфраструктуры читался бы как «IP likely blocked» — бан площадки.
2026-09-06 09:45:56 +05:00 · 2026-09-06 09:45:56 +05:00 · e12ece8fe6
commit e12ece8fe6
parent aeb1b2b3b9
2 changed files with 285 additions and 2 deletions
--- a/tradein-mvp/backend/tests/test_3394_cian_sweep_houses_proxy_pool.py
+++ b/tradein-mvp/backend/tests/test_3394_cian_sweep_houses_proxy_pool.py
@ -0,0 +1,222 @@
+"""#3394 — houses-фаза `cian_city_sweep` ходила в сайдкар мимо прокси-пула.
+
+`run_cian_city_sweep` звал `fetch_newbuilding(zhk_url, config=config)` БЕЗ
+`proxy_provider`, хотя провайдер лежал прямо в аргументах свипа и SERP/detail-фазы
+рядом его передают. Внутри `fetch_newbuilding` это давало
+`build_browser_fetcher(config, "cian", proxy_provider=None)` → `use_pool` эффективно
+False → POST /fetch без "proxy" → сайдкар брал env-узел `SCRAPER_PROXY_URL`, на проде
+выключенный (407 → camoufox `InvalidIP` → `/fetch` 503, факт #3386). Фаза давала 0/30 с
+02.09 (run 6179: 24 × `houses failed ... 503 Service Unavailable`).
+
+Тесты идут через НАСТОЯЩИЙ `run_cian_city_sweep` и НАСТОЯЩИЙ `fetch_newbuilding`;
+подделан только сам `BrowserFetcher` в `providers/_base` — то есть проверяется тот
+тракт, по которому строится фетчер на проде, а не отдельный вызов из теста.
+
+Второй тест — про исход пустого пула: он поднимается ДО запроса, следующий дом упрётся
+ровно в то же самое, поэтому фаза (и свип) обязаны оборваться на первом доме с
+`no_proxy_stop=1`, а не писать 30 одинаковых отказов и уходить в 'done' (образец —
+#3389, yandex-nb-sweep).
+
+Сеть/БД/камуфокс замоканы; в сеть тест не ходит.
+"""
+
+from __future__ import annotations
+
+import os
+from types import SimpleNamespace
+from typing import Any, ClassVar
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+from scraper_kit.orchestration.pipeline import run_cian_city_sweep
+from scraper_kit.proxy_errors import NoProxyAvailableError
+
+PFX = "scraper_kit.orchestration.pipeline"
+_BASE_FETCHER = "scraper_kit.providers._base.BrowserFetcher"
+
+_ZHK_URL = "https://zhk-test-ekb-i.cian.ru/"
+
+
+class _CapturingFetcher:
+    """Собирает kwargs конструктора; `fetch()` отдаёт HTML без initialState."""
+
+    captured: ClassVar[list[dict[str, Any]]] = []
+
+    def __init__(self, **kwargs: Any) -> None:
+        _CapturingFetcher.captured.append(kwargs)
+
+    async def __aenter__(self) -> _CapturingFetcher:
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+    async def fetch(self, _url: str, **_kwargs: Any) -> str:
+        return "<html></html>"
+
+    def report_ban(self, _reason: str) -> None:
+        return None
+
+
+class _EmptyPoolFetcher:
+    """Пул пуст: `fetch()` поднимает `NoProxyAvailableError` ДО запроса (#2616)."""
+
+    calls: ClassVar[list[str]] = []
+
+    def __init__(self, **_kwargs: Any) -> None:
+        pass
+
+    async def __aenter__(self) -> _EmptyPoolFetcher:
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+    async def fetch(self, url: str, **_kwargs: Any) -> str:
+        _EmptyPoolFetcher.calls.append(url)
+        raise NoProxyAvailableError("cian")
+
+    def report_ban(self, _reason: str) -> None:
+        return None
+
+
+class _RunsRecorder:
+    """Двойник scrape_runs: пишет финализаторы, is_cancelled всегда False."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, dict[str, Any]]] = []
+
+    def is_cancelled(self, _db: Any, _run_id: int) -> bool:
+        return False
+
+    def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, Any]) -> None:
+        self.calls.append(("update_heartbeat", dict(counters)))
+
+    def mark_done(self, _db: Any, _run_id: int, counters: dict[str, Any]) -> None:
+        self.calls.append(("mark_done", dict(counters)))
+
+    def mark_banned(
+        self,
+        _db: Any,
+        _run_id: int,
+        _error: str,
+        counters: dict[str, Any],
+        *,
+        ban_kind: str = "unknown",
+    ) -> None:
+        self.calls.append(("mark_banned", dict(counters)))
+
+    def mark_failed(self, _db: Any, _run_id: int, _error: str, counters: dict[str, Any]) -> None:
+        self.calls.append(("mark_failed", dict(counters)))
+
+
+def _config() -> SimpleNamespace:
+    return SimpleNamespace(
+        scraper_fetch_mode="curl_cffi",
+        browser_http_endpoint="http://tradein-browser:3000/fetch",
+        scraper_proxy_url=None,
+        cian_proxy_url=None,
+        use_proxy_pool_browser=True,
+        environment="production",
+        cian_proxy_max_rotations=0,
+        proxy_rotate_attempts=1,
+        proxy_rotate_attempt_timeout_s=1.0,
+        scraper_skip_seen_today=False,
+    )
+
+
+def _db(house_rows: list[dict[str, Any]]) -> MagicMock:
+    """db.execute: houses-запрос отдаёт дома, любой другой — пусто."""
+
+    def _execute(stmt: Any, *_a: Any, **_kw: Any) -> MagicMock:
+        res = MagicMock()
+        rows = house_rows if "FROM houses h" in str(stmt) else []
+        res.mappings.return_value.all.return_value = rows
+        res.fetchone.return_value = None
+        res.scalar_one.return_value = 0
+        return res
+
+    db = MagicMock()
+    db.execute.side_effect = _execute
+    return db
+
+
+def _nb_lot() -> MagicMock:
+    return MagicMock(
+        listing_segment="novostroyki",
+        house_source="cian_newbuilding",
+        house_ext_id="48853",
+    )
+
+
+async def _drive(
+    *, fetcher: type, house_rows: list[dict[str, Any]]
+) -> tuple[dict[str, int], list[tuple[str, dict[str, Any]]], MagicMock]:
+    recorder = _RunsRecorder()
+    provider = MagicMock(name="proxy_provider")
+    scraper = MagicMock()
+    scraper.__aenter__ = AsyncMock(return_value=scraper)
+    scraper.__aexit__ = AsyncMock(return_value=None)
+    scraper.fetch_around_multi_room = AsyncMock(return_value=[_nb_lot()])
+    scraper.state_extraction_attempts = 1
+    scraper.state_extraction_failures = 0
+    with (
+        patch(f"{PFX}.CianScraper", return_value=scraper),
+        patch(f"{PFX}.save_listings", MagicMock(return_value=(1, 0))),
+        patch(f"{PFX}.runs", recorder),
+        patch(_BASE_FETCHER, fetcher),
+    ):
+        counters = await run_cian_city_sweep(
+            _db(house_rows),
+            config=_config(),
+            matcher=MagicMock(),
+            run_id=1,
+            proxy_provider=provider,
+            anchors=[(56.84, 60.60, "A1")],
+            radius_m=1000,
+            pages_per_anchor=1,
+            request_delay_sec=0.0,
+            detail_top_n=0,
+            enrich_houses=True,
+            newbuilding_only=True,
+        )
+    return counters.to_dict(), recorder.calls, provider
+
+
+@pytest.mark.asyncio
+async def test_houses_phase_builds_fetcher_with_proxy_provider() -> None:
+    """(а) фетчер houses-фазы собран С провайдером пула — тот, что пришёл в свип."""
+    _CapturingFetcher.captured.clear()
+    counters, _calls, provider = await _drive(
+        fetcher=_CapturingFetcher,
+        house_rows=[{"id": 7, "cian_zhk_url": _ZHK_URL}],
+    )
+    assert counters["houses_attempted"] == 1
+    assert len(_CapturingFetcher.captured) == 1, _CapturingFetcher.captured
+    kwargs = _CapturingFetcher.captured[0]
+    # Ровно те три аргумента, без которых сайдкар уходит на env-узел.
+    assert kwargs["proxy_provider"] is provider
+    assert kwargs["use_pool"] is True
+    assert kwargs["environment"] == "production"
+
+
+@pytest.mark.asyncio
+async def test_empty_pool_stops_houses_phase_after_first_house() -> None:
+    """(б) пустой пул → один дом, no_proxy_stop=1, свип оборван (mark_banned)."""
+    _EmptyPoolFetcher.calls.clear()
+    counters, calls, _provider = await _drive(
+        fetcher=_EmptyPoolFetcher,
+        house_rows=[
+            {"id": 7, "cian_zhk_url": _ZHK_URL},
+            {"id": 8, "cian_zhk_url": _ZHK_URL},
+            {"id": 9, "cian_zhk_url": _ZHK_URL},
+        ],
+    )
+    assert _EmptyPoolFetcher.calls == [_ZHK_URL], "второй дом упрётся в то же самое"
+    assert counters["houses_attempted"] == 1
+    assert counters["houses_failed"] == 1
+    assert counters["no_proxy_stop"] == 1
+    assert calls[-1][0] == "mark_banned"
+    assert calls[-1][1]["no_proxy_stop"] == 1
--- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py
+++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py
@ -78,7 +78,7 @@ from scraper_kit.providers.yandex.serp import (
    ROOM_PATH,
    YandexRealtyScraper,
 )
-from scraper_kit.proxy_errors import NoProxyAvailableError
+from scraper_kit.proxy_errors import NoProxyAvailableError, caused_by_no_proxy

 if TYPE_CHECKING:
    from sqlalchemy.orm import Session
@ -2845,6 +2845,11 @@ class CianCitySweepCounters:
    houses_enriched: int = 0
    houses_failed: int = 0
    errors_count: int = 0
+    # #3394: 1 — прогон оборван «пул прокси пуст» (к площадке не ходили), 0 — нет.
+    # int, а не bool: счётчики читают SQL'ем `counters->>'no_proxy_stop' = '1'`,
+    # мимо JSON-ного `true` он промахнётся молча (тот же довод — в
+    # app/tasks/yandex_newbuilding_sweep.py::SweepResult.to_dict).
+    no_proxy_stop: int = 0

    def to_dict(self) -> dict[str, int]:
        return {f.name: getattr(self, f.name) for f in fields(self)}
@ -3221,7 +3226,18 @@ async def run_cian_city_sweep(
                    zhk_url: str = hrow["cian_zhk_url"]
                    counters.houses_attempted += 1
                    try:
-                        nb_enrichment = await fetch_newbuilding(zhk_url, config=config)
+                        # proxy_provider (#3394): без него build_browser_fetcher внутри
+                        # fetch_newbuilding получал proxy_provider=None → use_pool
+                        # эффективно False → POST /fetch уходил БЕЗ "proxy", и сайдкар
+                        # брал свой env-узел SCRAPER_PROXY_URL — на проде выключенный
+                        # (407 → camoufox InvalidIP → /fetch 503). Фаза houses шла 0/30
+                        # с 02.09, хотя detail-фаза выше ходила пулом всё это время.
+                        # Соседние вызывающие fetch_newbuilding провайдер передают:
+                        # cian_history_backfill.py:352 (#3382),
+                        # newbuilding_enrich_backfill.py:546 (#2767), admin.py:2175.
+                        nb_enrichment = await fetch_newbuilding(
+                            zhk_url, config=config, proxy_provider=proxy_provider
+                        )
                        if nb_enrichment is not None:
                            save_newbuilding_enrichment(db, house_id_val, nb_enrichment)
                            counters.houses_enriched += 1
@ -3237,6 +3253,23 @@ async def run_cian_city_sweep(
                    except Exception as exc:
                        counters.houses_failed += 1
                        counters.errors_count += 1
+                        if caused_by_no_proxy(exc):
+                            # #3394 (образец — #3197/#3389): пустой пул это НЕ отказ
+                            # площадки — запрос не уходил вовсе, и следующий дом
+                            # упрётся ровно в то же самое. Перебирать оставшиеся
+                            # значило бы писать 30 одинаковых houses_failed и уходить
+                            # в 'done'. Обрываем фазу и свип (ветка
+                            # `except NoProxyAvailableError` в цикле по anchor'ам).
+                            logger.error(
+                                "cian-sweep run_id=%d: houses СТОП — пул прокси пуст, "
+                                "к площадке не ходили (house_id=%d, дом %d/%d): %s",
+                                run_id,
+                                house_id_val,
+                                hidx + 1,
+                                len(house_rows),
+                                exc,
+                            )
+                            raise
                        logger.warning(
                            "cian-sweep run_id=%d: houses failed house_id=%d: %s",
                            run_id,
@ -3298,6 +3331,34 @@ async def run_cian_city_sweep(
                counters.anchors_done = idx
                runs.update_heartbeat(db, run_id, counters.to_dict())
                continue
+            except NoProxyAvailableError as exc:
+                # #3394: пул пуст — свип обрывается на первом же доме/лоте, а не
+                # перебирает остаток города одинаковыми отказами. Ветка стоит ДО
+                # generic-except: там NoProxyAvailableError уходил бы в
+                # consecutive_failures и (через 3 якоря) в mark_banned с диагнозом
+                # «IP likely blocked», то есть наш отказ читался бы как бан площадки.
+                # done_buckets сохраняем — следующий прогон не начнёт с нуля (#2686).
+                counters.no_proxy_stop = 1
+                counters.errors_count += 1
+                logger.error(
+                    "cian-sweep run_id=%d: СТОП на anchor #%d/%d (%s) — пул прокси "
+                    "пуст, к площадке не ходили: %s",
+                    run_id,
+                    idx,
+                    len(_anchors),
+                    name,
+                    exc,
+                )
+                _ckpt = {**counters.to_dict(), "done_buckets": sorted(_done_anchors)}
+                runs.update_heartbeat(db, run_id, _ckpt)
+                runs.mark_banned(
+                    db,
+                    run_id,
+                    f"cian sweep aborted: {exc}",
+                    _ckpt,
+                    ban_kind=ban_kind_of_exception(exc),
+                )
+                return counters
            except Exception as e:
                logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name)
                counters.errors_count += 1