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(

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] = {}

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, где

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):