feat(tradein): checkpoint/resume for yandex full-load — survive restart (#930)

tradein-mvp/backend/app/api/v1/admin.py

@@ -1081,6 +1081,14 @@ class YandexFullLoadRequest(BaseModel):
         le=8,
         description="Параллельных page-фетчей внутри одного leaf-бакета.",
     )
+    resume_run_id: int | None = Field(
+        default=None,
+        description=(
+            "Если задан — читает done_buckets из counters прошлого run и пропускает "
+            "уже завершённые бакеты. Передать run_id предыдущего (прерванного) "
+            "yandex_full_load прогона для resume. Без этого поля — full walk с нуля."
+        ),
+    )
 
 
 @router.post("/scrape/yandex-full-load", response_model=CitySweepStartResponse)
@@ -1098,6 +1106,9 @@ async def start_yandex_full_load(
     Один региональный проход БЕЗ anchor'ов (Yandex SERP path-based по городу — anchors
     избыточны). IP-ротация через changeip при captcha.
 
+    Checkpoint/resume: задай resume_run_id=<предыдущий run_id> чтобы пропустить
+    уже завершённые бакеты и продолжить с места остановки.
+
     Coop cancel: POST /scrape/yandex-full-load/{run_id}/cancel.
     Статус прогона: смотреть через GET /scrape/cian-city-sweep/runs
     (source='yandex_full_load').
@@ -1113,6 +1124,7 @@ async def start_yandex_full_load(
                 price_cap_per_bucket=payload.price_cap_per_bucket,
                 request_delay_sec=payload.request_delay_sec,
                 concurrency=payload.concurrency,
+                resume_run_id=payload.resume_run_id,
             )
         except Exception:
             logger.exception("yandex-full-load background task run_id=%d crashed", run_id)

tradein-mvp/backend/app/services/scrape_pipeline.py

@@ -1780,6 +1780,7 @@ async def run_yandex_full_load(
     price_cap_per_bucket: int = 500,
     request_delay_sec: float = 2.0,
     concurrency: int = 4,
+    resume_run_id: int | None = None,
 ) -> YandexFullLoadCounters:
     """Exhaustive региональный сбор Yandex ЕКБ вторички (БЕЗ anchor'ов).
 
@@ -1792,6 +1793,9 @@ async def run_yandex_full_load(
       price_cap_per_bucket: максимум офферов на price-бакет (< Yandex SERP-cap ~575).
       request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
       concurrency: параллельных page-фетчей в leaf-бакете.
+      resume_run_id: если задан — читает done_buckets из counters прошлого run и
+        пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного)
+        yandex_full_load прогона для resume. Без этого поля — full walk с нуля.
 
     Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
     Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
@@ -1800,22 +1804,50 @@ async def run_yandex_full_load(
 
     counters = YandexFullLoadCounters()
 
-    def _on_bucket(lots: list) -> None:  # type: ignore[type-arg]
+    skip_set: set[str] = set()
+    if resume_run_id is not None:
+        _prev_row = db.execute(
+            text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"),
+            {"rid": resume_run_id},
+        ).fetchone()
+        if _prev_row is not None and _prev_row.counters:
+            _prev_counters: dict = (
+                _prev_row.counters if isinstance(_prev_row.counters, dict) else {}
+            )
+            skip_set = set(_prev_counters.get("done_buckets", []))
+        logger.info(
+            "yandex-full-load run_id=%d: resuming from run %d — %d buckets already done",
+            run_id,
+            resume_run_id,
+            len(skip_set),
+        )
+    done: set[str] = set(skip_set)
+
+    def _on_bucket(bucket_key: str, lots: list) -> None:  # type: ignore[type-arg]
         """Инкрементальный save после каждого leaf-бакета."""
+        nonlocal done
         if scrape_runs.is_cancelled(db, run_id):
             logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id)
             raise RuntimeError("cancelled")
         if not lots:
+            done.add(bucket_key)
+            scrape_runs.update_heartbeat(
+                db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
+            )
             return
         inserted, updated = save_listings(db, lots, run_id=run_id)
         # save_listings вызывает db.commit() внутри — данные в БД сразу
         counters.saved_inserted += inserted
         counters.saved_updated += updated
         counters.unique_fetched += len(lots)
-        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+        done.add(bucket_key)
+        scrape_runs.update_heartbeat(
+            db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
+        )
         logger.info(
-            "yandex-full-load run_id=%d: bucket saved ins=%d upd=%d total_unique=%d",
+            "yandex-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d",
             run_id,
+            bucket_key,
             inserted,
             updated,
             counters.unique_fetched,
@@ -1834,6 +1866,7 @@ async def run_yandex_full_load(
                 concurrency=concurrency,
                 on_bucket=_on_bucket,
                 on_progress=_on_progress,
+                skip_buckets=skip_set if skip_set else None,
             )
 
         logger.info(
@@ -1844,7 +1877,7 @@ async def run_yandex_full_load(
             counters.saved_updated,
         )
         scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
-        scrape_runs.mark_done(db, run_id, counters.to_dict())
+        scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
         logger.info(
             "yandex-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d",
             run_id,
@@ -1865,7 +1898,7 @@ async def run_yandex_full_load(
                 counters.saved_inserted,
                 counters.saved_updated,
             )
-            scrape_runs.mark_done(db, run_id, counters.to_dict())
+            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
             return counters
         logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
         scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())

tradein-mvp/backend/app/services/scrapers/yandex_realty.py

@@ -429,6 +429,7 @@ class YandexRealtyScraper(BaseScraper):
         concurrency: int = 4,
         on_bucket: Any = None,
         on_progress: Any = None,
+        skip_buckets: set[str] | None = None,
     ) -> list[ScrapedLot]:
         """Exhaustive-загрузка Yandex ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА.
 
@@ -442,9 +443,13 @@ class YandexRealtyScraper(BaseScraper):
           price_cap_per_bucket: максимум офферов в бакете перед делением (< Yandex ~575 cap).
           max_pages_per_bucket: жёсткий cap страниц; 100 × 30 ≈ 3000 > 575.
           concurrency: максимум параллельных page-фетчей в leaf-бакете.
-          on_bucket: опциональный callback(list[ScrapedLot]) после каждого leaf-бакета.
+          on_bucket: опциональный callback(bucket_key: str, list[ScrapedLot]) после каждого
+            leaf-бакета. bucket_key имеет формат "{rooms}:{lo}:{hi}".
             Может быть async или sync. Исключение прерывает прогон.
           on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket).
+          skip_buckets: множество bucket_key уже завершённых бакетов (resume). Бакеты из
+            этого множества пропускаются — пагинация и on_bucket не вызываются (probe
+            всё равно выполняется для split-решения).
 
         Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
         """
@@ -468,6 +473,7 @@ class YandexRealtyScraper(BaseScraper):
                 max_pages_per_bucket=max_pages_per_bucket,
                 concurrency=concurrency,
                 on_bucket=on_bucket,
+                skip_buckets=skip_buckets,
             )
             room_collected = len(seen) - before
             logger.info(
@@ -493,6 +499,7 @@ class YandexRealtyScraper(BaseScraper):
         max_pages_per_bucket: int,
         concurrency: int = 4,
         on_bucket: Any = None,
+        skip_buckets: set[str] | None = None,
         _depth: int = 0,
     ) -> None:
         """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
@@ -504,6 +511,10 @@ class YandexRealtyScraper(BaseScraper):
           Guard: hi - lo < _YANDEX_MIN_BRACKET → пагинировать как есть.
           Fallback: если totalItems=None → paginate-until-empty (безопасная деградация).
 
+        skip_buckets: если задан, leaf-бакеты с ключом "{rooms}:{lo}:{hi}" из этого
+          множества пропускаются (пагинация + on_bucket не вызываются). Probe всё равно
+          выполняется для принятия split-решения.
+
         # TODO(yandex-total): totalItems not reliably extractable from hydrated DOM
         # — using paginate-until-empty fallback when state extraction fails.
         """
@@ -544,6 +555,12 @@ class YandexRealtyScraper(BaseScraper):
                 hi,
                 _depth,
             )
+            # Compute bucket_key AFTER probe (skip check is post-probe, pre-pagination)
+            bucket_key = f"{rooms}:{lo}:{hi}"
+            if skip_buckets and bucket_key in skip_buckets:
+                logger.info("yandex: skip bucket %s — already done (resume)", bucket_key)
+                return
+
             bucket_lots: list[ScrapedLot] = []
             # page=0 probe already fetched — reuse its HTML if available
             if html is not None and not _is_captcha(html):
@@ -579,7 +596,7 @@ class YandexRealtyScraper(BaseScraper):
                     seen[key] = lot
 
             if on_bucket and bucket_lots:
-                res = on_bucket(bucket_lots)
+                res = on_bucket(bucket_key, bucket_lots)
                 if inspect.isawaitable(res):
                     await res
             return
@@ -627,6 +644,7 @@ class YandexRealtyScraper(BaseScraper):
                 max_pages_per_bucket=max_pages_per_bucket,
                 concurrency=concurrency,
                 on_bucket=on_bucket,
+                skip_buckets=skip_buckets,
                 _depth=_depth + 1,
             )
             await self._walk_price_range(
@@ -638,11 +656,18 @@ class YandexRealtyScraper(BaseScraper):
                 max_pages_per_bucket=max_pages_per_bucket,
                 concurrency=concurrency,
                 on_bucket=on_bucket,
+                skip_buckets=skip_buckets,
                 _depth=_depth + 1,
             )
             return
 
         # ── Параллельная пагинация leaf-бакета ─────────────────────────────────
+        # Compute bucket_key AFTER split decision (skip check is post-split, pre-pagination)
+        bucket_key = f"{rooms}:{lo}:{hi}"
+        if skip_buckets and bucket_key in skip_buckets:
+            logger.info("yandex: skip bucket %s — already done (resume)", bucket_key)
+            return
+
         # Yandex SERP pages are 0-indexed (unlike Cian which is 1-indexed)
         max_pages = min(
             math.ceil(total / _YANDEX_OFFERS_PER_PAGE),
@@ -694,17 +719,18 @@ class YandexRealtyScraper(BaseScraper):
                 seen[key] = lot
 
         logger.info(
-            "yandex: rooms=%s [%d, %d] paginated=%d pages collected=%d unique_total=%d",
+            "yandex: rooms=%s [%d, %d] bucket=%s paginated=%d pages collected=%d unique_total=%d",
             rooms,
             lo,
             hi,
+            bucket_key,
             max_pages,
             len(bucket_lots_leaf),
             len(seen),
         )
 
         if on_bucket and bucket_lots_leaf:
-            res_cb = on_bucket(bucket_lots_leaf)
+            res_cb = on_bucket(bucket_key, bucket_lots_leaf)
             if inspect.isawaitable(res_cb):
                 await res_cb
 

tradein-mvp/backend/tests/test_yandex_realty_serp.py

@@ -705,14 +705,14 @@ async def test_walk_price_range_splits_when_total_exceeds_cap():
 
 @pytest.mark.asyncio
 async def test_walk_price_range_fires_on_bucket_per_leaf():
-    """on_bucket callback is called once per leaf bucket with the bucket's lots."""
+    """on_bucket callback is called once per leaf bucket with (bucket_key, lots)."""
     s = YandexRealtyScraper()
     s._cffi_session = AsyncMock()
 
-    bucket_calls: list[list] = []
+    bucket_calls: list[tuple[str, list]] = []
 
-    def fake_on_bucket(lots: list) -> None:
+    def fake_on_bucket(bucket_key: str, lots: list) -> None:
-        bucket_calls.append(list(lots))
+        bucket_calls.append((bucket_key, list(lots)))
 
     # total = 10 (< cap=500) → paginate leaf immediately, 1 page (page=0)
     async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str:
@@ -740,22 +740,23 @@ async def test_walk_price_range_fires_on_bucket_per_leaf():
 
     # on_bucket must have been called at least once
     assert len(bucket_calls) >= 1
-    # All calls contain ScrapedLot-like objects (the actual card from SINGLE_CARD_HTML)
+    # All calls contain (bucket_key, lots)
-    for call_lots in bucket_calls:
+    for key, call_lots in bucket_calls:
+        assert isinstance(key, str), f"bucket_key must be str, got {type(key)}"
         assert len(call_lots) >= 0  # may be 0 if card parse yields empty — still fires
 
 
 @pytest.mark.asyncio
 async def test_walk_price_range_fallback_when_total_none():
     """When totalItems=None (state unavailable), falls back to paginate-until-empty
-    and still fires on_bucket with whatever was collected."""
+    and still fires on_bucket(bucket_key, lots)."""
     s = YandexRealtyScraper()
     s._cffi_session = AsyncMock()
 
-    bucket_calls: list[list] = []
+    bucket_calls: list[tuple[str, list]] = []
 
-    def fake_on_bucket(lots: list) -> None:
+    def fake_on_bucket(bucket_key: str, lots: list) -> None:
-        bucket_calls.append(list(lots))
+        bucket_calls.append((bucket_key, list(lots)))
 
     page_htmls = {
         0: SINGLE_CARD_HTML,  # page=0: 1 card
@@ -785,10 +786,110 @@ async def test_walk_price_range_fallback_when_total_none():
     # Fallback path must fire on_bucket with the collected lots
     assert len(bucket_calls) >= 1, "on_bucket must be called in fallback path"
     # The card from SINGLE_CARD_HTML should have been collected
-    all_lots = [lot for call in bucket_calls for lot in call]
+    all_lots = [lot for _key, call in bucket_calls for lot in call]
     assert len(all_lots) >= 1
 
 
+# ── PART F: checkpoint / resume — skip_buckets ───────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_skip_buckets_skips_pagination_and_on_bucket():
+    """When a leaf bucket_key is in skip_buckets, pagination (page>0) and on_bucket
+    are NOT called for it. The probe fetch (page=0) may still run."""
+    s = YandexRealtyScraper()
+    s._cffi_session = AsyncMock()
+
+    on_bucket_calls: list[str] = []
+    pagination_fetches: list[int] = []
+
+    def fake_on_bucket(bucket_key: str, lots: list) -> None:
+        on_bucket_calls.append(bucket_key)
+
+    async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str:
+        if page > 0:
+            pagination_fetches.append(page)
+        return SINGLE_CARD_HTML
+
+    # bucket_key for rooms="1", lo=0, hi=5_000_000 is "1:0:5000000"
+    target_bucket_key = "1:0:5000000"
+    skip_set = {target_bucket_key}
+
+    seen: dict = {}
+    with (
+        patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(s, "_extract_total_count", return_value=10),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        await s._walk_price_range(
+            rooms="1",
+            lo=0,
+            hi=5_000_000,
+            seen=seen,
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=5,
+            concurrency=2,
+            on_bucket=fake_on_bucket,
+            skip_buckets=skip_set,
+        )
+
+    # on_bucket must NOT have been called for the skipped bucket
+    assert (
+        target_bucket_key not in on_bucket_calls
+    ), f"on_bucket must not fire for skipped bucket, but got: {on_bucket_calls}"
+    # No pagination pages (page > 0) should have been fetched for the skipped bucket
+    assert (
+        pagination_fetches == []
+    ), f"pagination must not run for skipped bucket, but page fetches: {pagination_fetches}"
+
+
+@pytest.mark.asyncio
+async def test_on_bucket_receives_key_and_lots():
+    """on_bucket receives (bucket_key: str, lots: list) where bucket_key matches
+    '{rooms}:{lo}:{hi}' shape for the normal-leaf path."""
+    s = YandexRealtyScraper()
+    s._cffi_session = AsyncMock()
+
+    received_keys: list[str] = []
+    received_lots: list[list] = []
+
+    def fake_on_bucket(bucket_key: str, lots: list) -> None:
+        received_keys.append(bucket_key)
+        received_lots.append(list(lots))
+
+    async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str:
+        if page == 0:
+            return SINGLE_CARD_HTML
+        return EMPTY_PAGE_HTML
+
+    seen: dict = {}
+    with (
+        patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html),
+        patch.object(s, "_extract_total_count", return_value=5),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        await s._walk_price_range(
+            rooms="3",
+            lo=1_000_000,
+            hi=8_000_000,
+            seen=seen,
+            price_cap_per_bucket=500,
+            max_pages_per_bucket=5,
+            concurrency=2,
+            on_bucket=fake_on_bucket,
+        )
+
+    assert len(received_keys) >= 1, "on_bucket must have been called"
+    # bucket_key must match the expected format
+    for key in received_keys:
+        assert (
+            key == "3:1000000:8000000"
+        ), f"bucket_key format mismatch: expected '3:1000000:8000000', got {key!r}"
+    # lots must be a list
+    for lots in received_lots:
+        assert isinstance(lots, list)
+
+
 @pytest.mark.asyncio
 async def test_fetch_all_secondary_deduplicates_across_rooms():
     """fetch_all_secondary deduplicates offers that appear in multiple room buckets."""