From d93e96fb0c7f210624e3511912d08bb7603bc21e Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 31 May 2026 21:34:23 +0000 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20checkpoint/resume=20for=20yand?= =?UTF-8?q?ex=20full-load=20=E2=80=94=20survive=20restart=20(#930)=20(#937?= =?UTF-8?q?)=20Co-authored-by:=20lekss361=20=20Co-?= =?UTF-8?q?committed-by:=20lekss361=20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/api/v1/admin.py | 12 ++ .../backend/app/services/scrape_pipeline.py | 43 +++++- .../app/services/scrapers/yandex_realty.py | 34 ++++- .../backend/tests/test_yandex_realty_serp.py | 123 ++++++++++++++++-- 4 files changed, 192 insertions(+), 20 deletions(-) diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index c6df173a..f587db5b 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/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) diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index ffc95615..25ad309c 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/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()) diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py index 52eb28fd..a73f4796 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py +++ b/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 diff --git a/tradein-mvp/backend/tests/test_yandex_realty_serp.py b/tradein-mvp/backend/tests/test_yandex_realty_serp.py index 3a7be582..22cbd824 100644 --- a/tradein-mvp/backend/tests/test_yandex_realty_serp.py +++ b/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: - bucket_calls.append(list(lots)) + def fake_on_bucket(bucket_key: str, lots: list) -> None: + 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) - for call_lots in bucket_calls: + # All calls contain (bucket_key, lots) + 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: - bucket_calls.append(list(lots)) + def fake_on_bucket(bucket_key: str, lots: list) -> None: + 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."""