diff --git a/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py b/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py index 51f75564..9715d395 100644 --- a/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py +++ b/tradein-mvp/backend/tests/test_3359_yandex_exhaustive_containment_skip.py @@ -142,7 +142,7 @@ def _walk_degraded(skip_buckets: set[str] | None) -> tuple[list[tuple[str, bool] s._rotate_ip = no_rotate # type: ignore[method-assign] marked: list[tuple[str, bool]] = [] - def on_bucket(key: str, _count: int, complete: bool = True) -> None: + def on_bucket(key: str, _lots: list[Any], complete: bool = True) -> None: marked.append((key, complete)) asyncio.run( diff --git a/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py b/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py index 9dbf9637..71aea485 100644 --- a/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py +++ b/tradein-mvp/backend/tests/test_3368_yandex_leaf_cap_complete.py @@ -84,7 +84,7 @@ def _walk( s._fetch_page_json = fake_fetch # type: ignore[method-assign] marked: list[tuple[str, bool]] = [] - def on_bucket(key: str, _count: int, complete: bool = True) -> None: + def on_bucket(key: str, _lots: list[ScrapedLot], complete: bool = True) -> None: marked.append((key, complete)) seen: dict[str, ScrapedLot] = {} diff --git a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py index d4e3eff7..ebc1d3a1 100644 --- a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py +++ b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py @@ -30,6 +30,7 @@ import pytest os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") +from scraper_kit.base import ScrapedLot from scraper_kit.orchestration.pipeline import ( run_avito_full_load, run_avito_newbuilding_sweep, @@ -40,6 +41,7 @@ from scraper_kit.orchestration.pipeline import ( run_yandex_full_load, ) from scraper_kit.providers.domclick.serp import ROOM_BUCKETS +from scraper_kit.providers.yandex.serp import YandexRealtyScraper PFX = "scraper_kit.orchestration.pipeline" @@ -619,7 +621,12 @@ async def _drive_full_load(*, source: str, capture: dict[str, Any] | None = None @pytest.mark.asyncio @pytest.mark.parametrize("source", ["avito", "cian", "yandex"]) async def test_full_load_smoke(source: str) -> None: - """Full load: 2 бакета через on_bucket → save + heartbeat → mark_done.""" + """Full load: 2 бакета через on_bucket → save + heartbeat → mark_done. + + ВНИМАНИЕ: `_full_load_scraper` шлёт в on_bucket список ПО ПОСТРОЕНИЮ — контракт + «что провайдер реально кладёт в колбэк» здесь не проверяется вовсе (#3375). + Настоящий контракт — в тесте ниже. + """ counters, calls = await _drive_full_load(source=source) assert counters["unique_fetched"] == 3 assert counters["saved_inserted"] == 3 @@ -627,6 +634,89 @@ async def test_full_load_smoke(source: str) -> None: assert calls[-1][0] == "mark_done" +# ── #3375: yandex full-load — контракт on_bucket через НАСТОЯЩИЙ провайдер ──── +# +# Фикстура выше подменяет скрапер целиком, поэтому в save_listings всегда приезжал +# список — а yandex/serp.py звал `on_bucket(bucket_key, len(seen), complete)` (int), +# и `_on_bucket` отдавал это в `save_listings` (`for lot in lots`) → TypeError: +# ручной full-load Яндекса не сохранял ничего. Ниже гоняется НАСТОЯЩИЙ +# YandexRealtyScraper (fetch_all_secondary → _walk_price_range → _leaf), подменён +# только gate-JSON транспорт. Фальсификация: вернуть в serp.py `len(seen)` — тест +# краснеет TypeError'ом на save_listings, как прод. + + +class _StubbedYandexScraper(YandexRealtyScraper): + """Настоящий скрапер яндекса, у которого замокан ТОЛЬКО gate-JSON транспорт.""" + + async def __aenter__(self) -> _StubbedYandexScraper: + return self # без camoufox/BrowserFetcher + + async def __aexit__(self, *_exc: Any) -> bool: + return False + + async def fetch_all_secondary(self, **kwargs: Any) -> list[Any]: + """Одна комнатность вместо пяти — путь до `_leaf` тот же, прогон короче.""" + return await super().fetch_all_secondary(rooms_buckets=["2"], **kwargs) + + async def _fetch_page_json( + self, + rooms: str | None, + page: int, + price_min: int | None = None, + price_max: int | None = None, + new_flat: str = "NO", + ) -> dict[str, Any]: + """totalItems=1 → бисекция не делит, leaf сразу. Оффер уникален на бакет.""" + offer_id = f"y{price_min or 0}_{price_max or 0}" + return { + "response": { + "search": { + "offers": { + "entities": [{"offerId": offer_id, "price": {"value": 5_000_000}}], + "pager": {"totalItems": 1, "totalPages": 1, "page": 0}, + } + } + } + } + + +@pytest.mark.asyncio +async def test_yandex_full_load_saves_lot_list_from_real_provider() -> None: + """#3375: save_listings получает СПИСОК лотов, ровно один вызов на бакет.""" + recorder = _RunsRecorder() + save_mock = MagicMock(return_value=(1, 0)) + enrichment = MagicMock() + enrichment.record_yandex_price_history = MagicMock(return_value=0) + with ( + patch(f"{PFX}.YandexRealtyScraper", _StubbedYandexScraper), + patch(f"{PFX}.save_listings", save_mock), + patch(f"{PFX}.runs", recorder), + ): + counters = await run_yandex_full_load( + MagicMock(), + run_id=1, + config=_config(), + matcher=MagicMock(), + enrichment=enrichment, + request_delay_sec=0.0, + ) + + assert save_mock.call_count > 0, "save_listings не вызван — прогон ничего не сохранил" + for call in save_mock.call_args_list: + lots = call.args[1] + assert isinstance(lots, list), ( + f"в save_listings уехал {type(lots).__name__} вместо списка лотов — " + "провайдер и pipeline разошлись контрактом (#3375)" + ) + assert lots and all(isinstance(lot, ScrapedLot) for lot in lots), ( + f"бакет отдал {lots!r} — save_listings пишет не лоты" + ) + # По одному новому лоту на бакет (offer_id уникален на ценовой диапазон). + assert counters.unique_fetched == save_mock.call_count > 0 + assert counters.saved_inserted == save_mock.call_count + assert _normalize(recorder.calls)[-1][0] == "mark_done" + + # ── #2616: run_avito_full_load прокидывает proxy_provider в AvitoScraper ───── # # run_avito_full_load — единственное из мест создания AvitoScraper в pipeline.py, где diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py index 4da92747..03ea92ec 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/yandex/serp.py @@ -1139,7 +1139,8 @@ class YandexRealtyScraper(BaseScraper): Реализация: единый движок `scraper_kit.pricing.walk_price_range`. Yandex-spec (probe gate-JSON page=1, degraded paginate-until-empty, leaf с first-wins - дедупом и count-based on_bucket) — в callback'ах; пороги (cap, 500k min, + дедупом; on_bucket отдаёт ЛОТЫ бакета — новые в `seen`, — как cian/avito: + pipeline._on_bucket кладёт их в save_listings, #3375) — в callback'ах; пороги (cap, 500k min, depth<8) и политика DEGRADE — в BisectionConfig. """ @@ -1194,6 +1195,7 @@ class YandexRealtyScraper(BaseScraper): return page = 1 pages_fetched = 0 + bucket_lots: list[ScrapedLot] = [] while pages_fetched < max_pages_per_bucket: payload = await self._fetch_page_json(rooms, page, lo_param, phi) await asyncio.sleep(self.request_delay_sec) @@ -1205,6 +1207,7 @@ class YandexRealtyScraper(BaseScraper): for lot in lots: if lot.source_id and lot.source_id not in seen: seen[lot.source_id] = lot + bucket_lots.append(lot) page += 1 pages_fetched += 1 if on_bucket is not None: @@ -1213,13 +1216,14 @@ class YandexRealtyScraper(BaseScraper): # бакет НЕ пишется в done-леджер (как у cian), иначе его интервал # слился бы с соседними в containment-гейте (#3359) и резюм уже не # переобошёл бы недобранную полосу. - on_bucket(bucket_key, len(seen), False) + on_bucket(bucket_key, bucket_lots, False) async def _leaf(plo: int | None, phi: int | None, result: ProbeResult) -> None: total = result.count assert total is not None # DEGRADE-политика уводит None в _degraded lo_param = plo if plo and plo > 0 else None probe_lots: list[ScrapedLot] = result.payload or [] + bucket_lots: list[ScrapedLot] = [] bucket_key = _combo_label(rooms, plo, phi) if skip_buckets and bucket_key in skip_buckets: logger.debug("yandex gate: skip bucket %s (checkpoint)", bucket_key) @@ -1262,10 +1266,11 @@ class YandexRealtyScraper(BaseScraper): for lot in probe_lots: if lot.source_id and lot.source_id not in seen: seen[lot.source_id] = lot + bucket_lots.append(lot) if total_pages <= 1: if on_bucket is not None: - on_bucket(bucket_key, len(seen), not capped) + on_bucket(bucket_key, bucket_lots, not capped) return # Paginate pages 2..total_pages with concurrency @@ -1288,6 +1293,7 @@ class YandexRealtyScraper(BaseScraper): for lot in page_lots: if lot.source_id and lot.source_id not in seen: seen[lot.source_id] = lot + bucket_lots.append(lot) complete = not capped and dropped_pages == 0 if dropped_pages: @@ -1299,13 +1305,15 @@ class YandexRealtyScraper(BaseScraper): total_pages - 1, ) if on_bucket is not None: - on_bucket(bucket_key, len(seen), complete) + on_bucket(bucket_key, bucket_lots, complete) # #3359: гейт ПЕРЕД probe (как у avito, #3315). Ключи yandex'а — `_combo_label`, # т.е. «rooms:lo-hi» с «None» вместо открытого потолка: другой разделитель и # другой open-токен, чем у avito/cian, поэтому парсер параметризуется, а ключи # остаются как есть (живые чекпоинты не ломаем). - _covered = done_range_skipper(skip_buckets, rooms or "any", range_sep="-", open_token="None") + _covered = done_range_skipper( + skip_buckets, rooms or "any", range_sep="-", open_token="None" + ) def _skip_done(plo: int | None, phi: int | None) -> bool: if _covered is None or not _covered(plo, phi):