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..aaa1db08 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,139 @@ 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 → _probe/_leaf/ +# _degraded), подменён только gate-JSON транспорт. +# +# on_bucket в serp.py зовётся из ТРЁХ мест, и стаб обязан доходить до каждого, +# иначе фальсификация двух из них останется зелёной (ревью PR #3378): +# * _degraded — брекет `_DEGRADED_LO` (probe отдаёт None, _rotate_ip=False); +# * leaf одностраничный — все прочие брекеты (totalItems=1); +# * leaf многостраничный — брекет `_MULTIPAGE_LO` (totalItems требует 3 страницы). +# Фальсификация: вернуть в serp.py len(seen) — тест краснеет TypeError'ом на +# `counters.unique_fetched += len(lots)` (save_listings здесь мок и int проглатывает; +# в проде падает сам save_listings). + +# Сид-брекеты берутся из get_price_seed_brackets() — эти два существуют в сетке ЕКБ. +_MULTIPAGE_LO = 4_000_000 # totalItems=45 при _GATE_PAGE_SIZE=20 → 3 страницы +_DEGRADED_LO = 8_000_000 # probe этого брекета проваливается → DEGRADE-политика +_MULTIPAGE_TOTAL = 45 +_MULTIPAGE_PAGES = 3 + + +class _StubbedYandexScraper(YandexRealtyScraper): + """Настоящий скрапер яндекса, у которого замокан ТОЛЬКО gate-JSON транспорт.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + # Брекеты, чей probe уже провалился: повторный запрос (он приходит уже из + # _degraded) отдаёт данные. Иначе degraded-бакет пришёл бы пустым и + # pipeline._on_bucket вернулся бы на `if not lots` до save_listings. + self._probe_failed: set[int | None] = set() + + async def __aenter__(self) -> _StubbedYandexScraper: + return self # без camoufox/BrowserFetcher + + async def __aexit__(self, *_exc: Any) -> bool: + return False + + async def _rotate_ip(self) -> bool: + """Ротация не спасает → probe-fail доходит до DEGRADE, а не до retry-успеха.""" + 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] | None: + """Ответ гейта, зависящий от брекета — см. комментарий над классом.""" + if price_min == _DEGRADED_LO and price_min not in self._probe_failed: + self._probe_failed.add(price_min) + return None # probe провалился; _degraded ниже уже пагинирует по данным + multipage = price_min == _MULTIPAGE_LO + total = _MULTIPAGE_TOTAL if multipage else 1 + pages = _MULTIPAGE_PAGES if multipage else 1 + # Degraded-ветка пагинирует до пустоты — вторая страница обрывает цикл. + entities: list[dict[str, Any]] = [] + if page <= pages: + offer_id = f"y{price_min or 0}_{price_max or 0}_p{page}" + entities = [{"offerId": offer_id, "price": {"value": 5_000_000}}] + return { + "response": { + "search": { + "offers": { + "entities": entities, + "pager": {"totalItems": total, "totalPages": pages, "page": page - 1}, + } + } + } + } + + +@pytest.mark.asyncio +async def test_yandex_full_load_saves_lot_list_from_real_provider() -> None: + """#3375: save_listings получает СПИСОК лотов из всех трёх веток on_bucket.""" + 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 не вызван — прогон ничего не сохранил" + sizes: list[int] = [] + 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 пишет не лоты" + ) + sizes.append(len(lots)) + # Обе leaf-ветки реально пройдены: одностраничная (1 лот) и многостраничная + # (probe + страницы 2..3). Без этого фальсификация многостраничного вызова + # on_bucket осталась бы зелёной — стаб до неё просто не доходил. + assert _MULTIPAGE_PAGES in sizes, ( + f"ни один бакет не собрал {_MULTIPAGE_PAGES} лота — многостраничный leaf " + f"не пройден, размеры бакетов: {sizes}" + ) + assert 1 in sizes, f"одностраничный leaf не пройден, размеры бакетов: {sizes}" + # То же для _degraded: единственный неполный бакет прогона — тот, чей probe + # провалился (потолком страниц здесь никого не обрезает). + assert counters.partial_buckets == 1, ( + f"partial_buckets={counters.partial_buckets} — degraded-ветка не пройдена " + "(probe-fail не доехал до _degraded)" + ) + assert counters.unique_fetched == sum(sizes) > 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):