From 47afe402c1aeff4f9ba66c50d44ed0ea0d84c214 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Mon, 1 Jun 2026 00:03:09 +0300 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20checkpoint/resume=20for=20cian?= =?UTF-8?q?=20full-load=20=E2=80=94=20survive=20restart=20(#930)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit done_buckets (set of "room:lo:hi") в scrape_runs.counters, дописывается per-bucket. Endpoint принимает resume_run_id → новый run читает done_buckets прошлого и пропускает уже завершённые бакеты (skip пагинации; probe-запросы остаются). Без resume_run_id — full walk как раньше. on_bucket сигнатура → (key, lots). --- tradein-mvp/backend/app/api/v1/admin.py | 9 ++ .../backend/app/services/scrape_pipeline.py | 44 ++++++- .../backend/app/services/scrapers/cian.py | 29 ++++- .../tests/scrapers/test_cian_exhaustive.py | 116 ++++++++++++++++-- 4 files changed, 181 insertions(+), 17 deletions(-) diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 9372f918..57436a95 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -982,6 +982,14 @@ class CianFullLoadRequest(BaseModel): le=50, description="Сколько свежих листингов обогатить detail (актуально при enrich_detail=True).", ) + resume_run_id: int | None = Field( + default=None, + description=( + "Если задан — читает done_buckets из counters прошлого run и пропускает " + "уже завершённые бакеты. Передать run_id предыдущего (прерванного) " + "cian_full_load прогона для resume. Без этого поля — full walk с нуля." + ), + ) @router.post("/scrape/cian-full-load", response_model=CitySweepStartResponse) @@ -1016,6 +1024,7 @@ async def start_cian_full_load( concurrency=payload.concurrency, enrich_detail=payload.enrich_detail, detail_top_n=payload.detail_top_n, + resume_run_id=payload.resume_run_id, ) except Exception: logger.exception("cian-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 12b7f171..6c0f26ed 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -1558,6 +1558,7 @@ async def run_cian_full_load( request_delay_sec: float = 4.0, enrich_detail: bool = False, concurrency: int = 5, + resume_run_id: int | None = None, ) -> CianFullLoadCounters: """Exhaustive региональный сбор Cian ЕКБ вторички (БЕЗ anchor'ов). @@ -1581,22 +1582,52 @@ async def run_cian_full_load( counters = CianFullLoadCounters() - def _on_bucket(lots: list) -> None: # type: ignore[type-arg] - """Инкрементальный save после каждого leaf-бакета.""" + # ── Checkpoint/resume: читаем done_buckets из прошлого run ─────────────── + 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( + "cian-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-бакета. Дописывает bucket_key в done.""" + nonlocal done if scrape_runs.is_cancelled(db, run_id): logger.info("cian-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( - "cian-full-load run_id=%d: bucket saved ins=%d upd=%d total_unique=%d", + "cian-full-load run_id=%d: bucket %s saved ins=%d upd=%d total_unique=%d", run_id, + bucket_key, inserted, updated, counters.unique_fetched, @@ -1615,6 +1646,7 @@ async def run_cian_full_load( concurrency=concurrency, on_bucket=_on_bucket, on_progress=_on_progress, + skip_buckets=skip_set if skip_set else None, ) logger.info( @@ -1687,7 +1719,7 @@ async def run_cian_full_load( ) 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( "cian-full-load run_id=%d done: unique=%d ins=%d upd=%d detail=%d/%d errors=%d", run_id, @@ -1710,7 +1742,7 @@ async def run_cian_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("cian-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/cian.py b/tradein-mvp/backend/app/services/scrapers/cian.py index 876d3c7a..c12e161c 100644 --- a/tradein-mvp/backend/app/services/scrapers/cian.py +++ b/tradein-mvp/backend/app/services/scrapers/cian.py @@ -320,6 +320,7 @@ class CianScraper(BaseScraper): concurrency: int = 5, on_bucket: Callable[..., Any] | None = None, on_progress: Callable[[int], None] | None = None, + skip_buckets: set[str] | None = None, ) -> list[ScrapedLot]: """Exhaustive-загрузка Cian ЕКБ вторички через партиционирование КОМНАТНОСТЬ × ЦЕНА. @@ -334,9 +335,12 @@ class CianScraper(BaseScraper): price_cap_per_bucket: максимум офферов в бакете перед делением (< 1500). max_pages_per_bucket: Cian hard cap ~54; не превышать. concurrency: максимум параллельных page-фетчей в leaf-бакете (default=5). - on_bucket: опциональный callback(list[ScrapedLot]) после каждого leaf-бакета. - Может быть async или sync. Если кидает исключение — прерывает прогон. + on_bucket: опциональный callback(bucket_key, list[ScrapedLot]) после каждого + leaf-бакета. Может быть async или sync. Если кидает исключение — прерывает прогон. on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket). + skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов — + пагинация и on_bucket для них пропускаются. Probe-запросы (для split-решения) + всё равно выполняются (их мало по сравнению с пагинацией). Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url). """ @@ -361,6 +365,7 @@ class CianScraper(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( @@ -386,6 +391,7 @@ class CianScraper(BaseScraper): max_pages_per_bucket: int, concurrency: int = 5, on_bucket: Callable[..., Any] | None = None, + skip_buckets: set[str] | None = None, _depth: int = 0, ) -> None: """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi]. @@ -396,8 +402,10 @@ class CianScraper(BaseScraper): 3. Если totalOffers > cap → разбить бакет пополам (рекурсия). Guard: hi - lo < _MIN_BRACKET → пагинировать как есть (логируем WARNING). - После пагинации leaf-бакета вызывает on_bucket(bucket_lots) если задан. + После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots) если задан. + bucket_key = "room_label:lo:hi" (checkpoint-ключ для resume). on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон. + skip_buckets: если bucket_key в skip_buckets — пагинацию и on_bucket пропускаем. """ effective_hi = hi @@ -481,6 +489,7 @@ class CianScraper(BaseScraper): max_pages_per_bucket=max_pages_per_bucket, concurrency=concurrency, on_bucket=on_bucket, + skip_buckets=skip_buckets, _depth=_depth + 1, ) # [mid+1, hi] @@ -493,10 +502,22 @@ class CianScraper(BaseScraper): max_pages_per_bucket=max_pages_per_bucket, concurrency=concurrency, on_bucket=on_bucket, + skip_buckets=skip_buckets, _depth=_depth + 1, ) return + # ── Checkpoint-ключ leaf-бакета ────────────────────────────────────── + bucket_key = f"{room_label}:{lo}:{hi}" + + # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket. + if skip_buckets and bucket_key in skip_buckets: + logger.info( + "cian: skip bucket %s — already done (resume)", + bucket_key, + ) + return + # ── Параллельная пагинация leaf-бакета ──────────────────────────────── max_pages = min( math.ceil(total / _CIAN_OFFERS_PER_PAGE), @@ -563,7 +584,7 @@ class CianScraper(BaseScraper): # ── on_bucket callback: инкрементальный save ────────────────────────── if on_bucket is not None and bucket_lots: - res_cb = on_bucket(bucket_lots) + res_cb = on_bucket(bucket_key, bucket_lots) if inspect.isawaitable(res_cb): await res_cb diff --git a/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py b/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py index 694cdbd2..79391ed5 100644 --- a/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py +++ b/tradein-mvp/backend/tests/scrapers/test_cian_exhaustive.py @@ -6,8 +6,12 @@ from __future__ import annotations +import os from unittest.mock import AsyncMock, MagicMock, patch +# Settings requires DATABASE_URL at import time — set dummy DSN before any app import. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + import pytest from app.services.scrapers.base import ScrapedLot @@ -213,10 +217,10 @@ async def test_fetch_all_secondary_min_bracket_guard(scraper: CianScraper) -> No async def test_fetch_all_secondary_on_bucket_called_per_leaf(scraper: CianScraper) -> None: """on_bucket вызывается после каждого leaf-бакета с лотами бакета.""" # totalOffers=56 ≤ cap → leaf-бакет, пагинируется параллельно 2 страницы - bucket_calls: list[list[ScrapedLot]] = [] + bucket_calls: list[tuple[str, list[ScrapedLot]]] = [] - def fake_on_bucket(lots: list[ScrapedLot]) -> None: - bucket_calls.append(list(lots)) + def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None: + bucket_calls.append((bucket_key, list(lots))) async def fake_fetch_page_html( rooms: tuple, page: int, min_price: int | None, max_price: int | None @@ -249,17 +253,18 @@ async def test_fetch_all_secondary_on_bucket_called_per_leaf(scraper: CianScrape # on_bucket вызван ровно 1 раз (один leaf-бакет для одной комнатности) assert len(bucket_calls) == 1, f"Ожидался 1 вызов on_bucket, получено {len(bucket_calls)}" + key, lots = bucket_calls[0] + # Ключ бакета содержит room_label:lo:hi + assert "room1" in key, f"bucket_key должен содержать room1, got {key!r}" # Лоты из обеих страниц переданы в on_bucket - assert ( - len(bucket_calls[0]) == 20 - ), f"Ожидалось 20 лотов в on_bucket (2 стр × 10), получено {len(bucket_calls[0])}" + assert len(lots) == 20, f"Ожидалось 20 лотов в on_bucket (2 стр × 10), получено {len(lots)}" @pytest.mark.asyncio async def test_fetch_all_secondary_on_bucket_cancel_stops_run(scraper: CianScraper) -> None: """Если on_bucket кидает RuntimeError('cancelled') — прогон прерывается.""" - def cancel_on_bucket(lots: list[ScrapedLot]) -> None: + def cancel_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None: raise RuntimeError("cancelled") async def fake_fetch_page_html( @@ -331,3 +336,100 @@ async def test_fetch_all_secondary_concurrent_pages_deduped(scraper: CianScraper assert 3 in fetch_pages or 3 in [ p for p in fetch_pages ], f"Страница 3 не была запрошена, fetch_pages={fetch_pages}" + + +@pytest.mark.asyncio +async def test_skip_buckets_skips_pagination_and_on_bucket(scraper: CianScraper) -> None: + """skip_buckets: для скипнутого бакета пагинация не вызывается, on_bucket не вызывается.""" + fetch_calls: list[tuple] = [] + on_bucket_calls: list[str] = [] + + async def fake_fetch_page_html( + rooms: tuple, page: int, min_price: int | None, max_price: int | None + ) -> str: + fetch_calls.append((rooms, page, min_price, max_price)) + return f"page={page}" + + def fake_extract_total_offers(html: str) -> int | None: + return 28 # один leaf-бакет, 1 страница + + def fake_parse_serp_html(html: str) -> list[ScrapedLot]: + return [_make_lot("lot_1")] + + def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None: + on_bucket_calls.append(bucket_key) + + # Сформируем ключ того же бакета, который получится при walk (room1:0:200000000) + bucket_key_to_skip = f"room1:0:{_MAX_PRICE}" + skip_set = {bucket_key_to_skip} + + with ( + patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), + patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), + patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), + patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + lots = await scraper.fetch_all_secondary( + rooms_buckets=[(1,)], + price_cap_per_bucket=1400, + on_bucket=fake_on_bucket, + skip_buckets=skip_set, + ) + + # Probe (page=1) ВСЁ РАВНО вызывается (нужен для split-решения), + # но НЕ должно быть страниц пагинации (т.к. бакет в skip_buckets). + # Поскольку probe = page=1 и мы пропускаем ДО пагинации — fetch_page_html не должен вызываться. + assert ( + len(fetch_calls) == 1 + ), f"Ожидался ровно 1 вызов (probe), получено {len(fetch_calls)}: {fetch_calls}" + assert fetch_calls[0][1] == 1, "Probe должен быть page=1" + # on_bucket не вызывается для skip-бакета + assert ( + len(on_bucket_calls) == 0 + ), f"on_bucket не должен вызываться для skip-бакета, got: {on_bucket_calls}" + # lots пустой — ничего не собрали (пагинация пропущена) + assert len(lots) == 0, f"Ожидалось 0 лотов для skip-бакета, got {len(lots)}" + + +@pytest.mark.asyncio +async def test_on_bucket_receives_key_and_lots(scraper: CianScraper) -> None: + """on_bucket получает (bucket_key, lots) — bucket_key в формате room_label:lo:hi.""" + received: list[tuple[str, int]] = [] # (key, len(lots)) + + def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None: + received.append((bucket_key, len(lots))) + + async def fake_fetch_page_html( + rooms: tuple, page: int, min_price: int | None, max_price: int | None + ) -> str: + return f"page={page}" + + def fake_extract_total_offers(html: str) -> int | None: + return 28 # 1 страница + + def fake_parse_serp_html(html: str) -> list[ScrapedLot]: + import re + + page_m = re.search(r"page=(\d+)", html) + page = int(page_m.group(1)) if page_m else 1 + return [_make_lot(f"lot_p{page}_{i}") for i in range(5)] + + with ( + patch.object(scraper, "_fetch_page_html", side_effect=fake_fetch_page_html), + patch.object(scraper, "_extract_total_offers", side_effect=fake_extract_total_offers), + patch.object(scraper, "_parse_serp_html", side_effect=fake_parse_serp_html), + patch.object(scraper, "sleep_between_requests", new_callable=AsyncMock), + patch.object(scraper, "_rotate_ip", return_value=False), + ): + await scraper.fetch_all_secondary( + rooms_buckets=[(2,)], + price_cap_per_bucket=1400, + on_bucket=fake_on_bucket, + ) + + assert len(received) == 1, f"Ожидался 1 вызов on_bucket, получено {len(received)}" + key, count = received[0] + # Ключ должен быть «room2:0:» + assert key == f"room2:0:{_MAX_PRICE}", f"Неверный bucket_key: {key!r}" + assert count == 5, f"Ожидалось 5 лотов, получено {count}"