diff --git 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 new file mode 100644 index 00000000..e80e1d4f --- /dev/null +++ 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 "" + + 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 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index 0f41f6b4..ada518f9 100644 --- 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