From e4b3c6cc2b3e2adbcb36d5f306cbe7e08e1c89f5 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 26 Aug 2026 12:35:44 +0500 Subject: [PATCH 1/2] =?UTF-8?q?fix(tradein/scheduler):=20=D1=87=D0=B5?= =?UTF-8?q?=D0=BA=D0=BF=D0=BE=D0=B8=D0=BD=D1=82=20=D0=BF=D1=80=D0=B5=D0=B4?= =?UTF-8?q?=D1=88=D0=B5=D1=81=D1=82=D0=B2=D0=B5=D0=BD=D0=BD=D0=B8=D0=BA?= =?UTF-8?q?=D0=B0=20=D0=BF=D0=B5=D1=80=D0=B5=D0=B6=D0=B8=D0=B2=D0=B0=D0=B5?= =?UTF-8?q?=D1=82=20=D0=BE=D0=B1=D1=80=D1=8B=D0=B2=20=D0=B4=D0=BE=20=D0=BF?= =?UTF-8?q?=D0=B5=D1=80=D0=B2=D0=BE=D0=B9=20=D0=BD=D0=BE=D0=B2=D0=BE=D0=B9?= =?UTF-8?q?=20=D0=BA=D0=BE=D1=80=D0=B7=D0=B8=D0=BD=D1=8B=20(#3074)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Прод-факт: прогон 4707 (23.08) подхватил 42 корзины у 4117, был убит деплоем на 26-й минуте до завершения первой НОВОЙ корзины — и не успел ни разу написать heartbeat с done_buckets. Его собственный чекпоинт пуст: следующий кандидат увидел бы no_checkpoint, цепочка оборвалась бы с потерей 42 корзин. _resume_decision при вердикте 'ok' теперь кладёт done_buckets предшественника в counters-заготовку нового прогона — она персистится при claim, до старта пайплайна. Heartbeat мержит jsonb: первый настоящий bucket-heartbeat перезапишет ключ надмножеством, двойной записи нет. Отказные вердикты чекпоинт не наследуют (прогон с нуля не должен врать о собранном). Co-Authored-By: Claude Opus 5 --- .../test_3074_checkpoint_survives_claim.py | 63 +++++++++++++++++++ .../scraper_kit/orchestration/scheduler.py | 11 ++++ 2 files changed, 74 insertions(+) create mode 100644 tradein-mvp/backend/tests/test_3074_checkpoint_survives_claim.py diff --git a/tradein-mvp/backend/tests/test_3074_checkpoint_survives_claim.py b/tradein-mvp/backend/tests/test_3074_checkpoint_survives_claim.py new file mode 100644 index 00000000..04ed87af --- /dev/null +++ b/tradein-mvp/backend/tests/test_3074_checkpoint_survives_claim.py @@ -0,0 +1,63 @@ +"""#3074: чекпоинт предшественника наследуется в counters НОВОГО прогона при claim. + +Прод-факт (Beget, 23.08): прогон 4707 подхватил чекпоинт 4117 (42 корзины, +resume_reason='ok'), был убит деплоем на 26-й минуте — ДО завершения первой +НОВОЙ корзины — и не успел ни разу написать heartbeat с done_buckets. Его +собственный чекпоинт остался пуст: следующий кандидат увидел бы no_checkpoint, +и 42 корзины пропали бы, хотя цепочка (resume_chain=2) ещё позволяла подхват. + +Фикс: _resume_decision при вердикте 'ok' кладёт done_buckets предшественника в +counters-заготовку нового прогона — она пишется в БД прямо при claim +(_pick_resume → update_heartbeat), до старта пайплайна. Heartbeat мержит jsonb, +первый настоящий bucket-heartbeat перезапишет ключ надмножеством. + +Красный на origin/main по ЗНАЧЕНИЮ: 'done_buckets' отсутствует в вердикте. +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from types import SimpleNamespace +from typing import Any + +from scraper_kit.orchestration import scheduler as sched + + +def _candidate(**over: Any) -> SimpleNamespace: + """Кандидат из _RESUME_CANDIDATE_SQL: прогон 4117 как он лежал на проде 23.08.""" + base = { + "prev_id": 4117, + "prev_status": "banned", + "prev_counters": { + "resume_chain": 1, + "done_buckets": [f"room_1_komn:{i}:0" for i in range(42)], + }, + "same_params": True, + "age_h": 164.8, + "interval_days": "7", + } + base.update(over) + return SimpleNamespace(**base) + + +def test_ok_verdict_carries_predecessor_checkpoint() -> None: + """Вердикт 'ok' несёт done_buckets предшественника — чекпоинт персистится при + claim и переживает обрыв до первой новой корзины (кейс 4707).""" + resume_id, verdict = sched._resume_decision(_candidate()) + assert resume_id == 4117 + assert verdict["resume_reason"] == "ok" + assert verdict.get("done_buckets") == sorted(f"room_1_komn:{i}:0" for i in range(42)), ( + "counters-заготовка нового прогона не содержит чекпоинт предшественника — " + "обрыв до первой новой корзины снова потеряет всю цепочку" + ) + + +def test_refusal_verdict_carries_no_checkpoint() -> None: + """Отказ от подхвата чекпоинт не наследует: прогон честно идёт с нуля, и его + counters не должны врать, будто корзины предшественника уже собраны.""" + _id, verdict = sched._resume_decision(_candidate(same_params=False)) + assert verdict["resume_reason"] == "params_changed" + assert "done_buckets" not in verdict diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py index 8b0b5009..d3e0446b 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py @@ -599,6 +599,17 @@ def _resume_decision(row: Any) -> tuple[int | None, dict[str, Any]]: verdict["resume_from"] = int(row.prev_id) verdict["resume_reason"] = "ok" verdict["resume_chain"] = prev_chain + 1 + # Наследуем чекпоинт в counters нового прогона ПРЯМО ПРИ CLAIM (#3074). + # Прод-факт (run 4707, 23.08): возобновлённый прогон, убитый деплоем на + # 26-й минуте — ДО завершения первой новой корзины, — не успел ни разу + # написать heartbeat с done_buckets. Его собственный чекпоинт остался + # пуст, кандидат следующей субботы увидел no_checkpoint, и 42 корзины + # предшественника (4117) пропали. Пайплайн сеет `done = set(skip_set)` + # у себя в памяти, но до первого _on_bucket это знание нигде не + # персистится; heartbeat мержит jsonb (`counters || :counters`), так что + # первый же настоящий bucket-heartbeat перезапишет ключ тем же множеством + # плюс новое — двойной записи не возникает. + verdict["done_buckets"] = sorted(done_buckets) if done_buckets else [] return int(row.prev_id), verdict return None, verdict -- 2.45.3 From 307055232ec1c4843ee7326c2ab4fa2c49b2ad5c Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 26 Aug 2026 12:43:58 +0500 Subject: [PATCH 2/2] =?UTF-8?q?feat(tradein/yandex):=20=D1=87=D0=B5=D0=BA?= =?UTF-8?q?=D0=BF=D0=BE=D0=B8=D0=BD=D1=82=D1=8B=20=D0=B4=D0=BB=D1=8F=20yan?= =?UTF-8?q?dex=5Fcity=5Fsweep=20=E2=80=94=20combo=20=D0=BA=D0=B0=D0=BA=20?= =?UTF-8?q?=D0=B5=D0=B4=D0=B8=D0=BD=D0=B8=D1=86=D0=B0=20=D0=B2=D0=BE=D0=B7?= =?UTF-8?q?=D0=BE=D0=B1=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D1=8F=20?= =?UTF-8?q?(#3074)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Из таблицы убитых деплоем: yandex_city_sweep 15.08 прожил 65 мин, 12.08 — 2ч33м; оба потеряны целиком — у свипа не было чекпоинтов вовсе. Единица чекпоинта — combo (сегмент × комнатность × ценовой диапазон), ровно то, чем цикл обхода уже итерируется. Три слоя: - провайдер: skip_combos (ни одного HTTP по собранным) + on_combo для каждого ПРОЙДЕННОГО combo, включая пустые — иначе пустой combo не попадал бы в чекпоинт и перечитывался бы вечно; оборванный отказом combo (gate failure) on_combo по-прежнему не вызывает; - пайплайн: done_combos → heartbeat с done_buckets (мерж jsonb, финализаторы не затирают); подхват гейтится единственным якорем — combo_label не содержит якоря, multi-anchor подхват пропускал бы чужие якоря; - планировщик: resume_run_id=_pick_resume(...) в диспатче (generic-механизм #2845 — params-идентичность, свежесть точки, потолок цепочки — бесплатно). Co-Authored-By: Claude Opus 5 --- .../test_3074_yandex_sweep_checkpoint.py | 241 ++++++++++++++++++ .../src/scraper_kit/orchestration/pipeline.py | 55 +++- .../scraper_kit/orchestration/scheduler.py | 1 + .../src/scraper_kit/providers/yandex/serp.py | 22 +- 4 files changed, 312 insertions(+), 7 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_3074_yandex_sweep_checkpoint.py diff --git a/tradein-mvp/backend/tests/test_3074_yandex_sweep_checkpoint.py b/tradein-mvp/backend/tests/test_3074_yandex_sweep_checkpoint.py new file mode 100644 index 00000000..45dbe87c --- /dev/null +++ b/tradein-mvp/backend/tests/test_3074_yandex_sweep_checkpoint.py @@ -0,0 +1,241 @@ +"""#3074: чекпоинты для yandex_city_sweep — combo как единица возобновления. + +Из таблицы убитых деплоем прогонов (#3074): yandex_city_sweep 15.08 прожил 65 мин, +12.08 — 2 ч 33 мин; оба потеряны целиком, потому что у свипа не было чекпоинтов +вовсе (avito/cian full-load обзавелись ими в #930/#2845). + +Единица чекпоинта — combo (сегмент × комнатность × ценовой диапазон): ровно то, +чем цикл обхода уже итерируется, метка та же, что у бакетов avito +("vtorichka/room_1:0-3000000"). Три слоя: + провайдер — skip_combos (ни одного HTTP по собранным) + on_combo для каждого + ПРОЙДЕННОГО combo, включая пустые (иначе пустой combo не попадал + бы в чекпоинт и перечитывался бы вечно); + пайплайн — done_combos → heartbeat c done_buckets (мерж jsonb); + планировщик — resume_run_id=_pick_resume(...) в диспатче. + +Красные на origin/main: планировщик передаёт None литералом (по значению), +on_combo не вызывается для пустых combo (по значению). +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import json +import types +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +from scraper_kit.orchestration import scheduler as sched +from scraper_kit.providers.yandex.serp import YandexRealtyScraper + +# ── провайдер: заглушка gate-API ───────────────────────────────────────────── + + +def _gate_payload(entities: list[dict[str, Any]], total_pages: int = 1) -> str: + offers = {"entities": entities, "pager": {"totalPages": total_pages}} + return json.dumps({"response": {"search": {"offers": offers}}}) + + +class _RecordingBrowser: + """BrowserFetcher-заглушка: отвечает одним и тем же телом, считает вызовы.""" + + def __init__(self, body: str) -> None: + self.body = body + self.urls: list[str] = [] + + async def fetch(self, url: str) -> str: + self.urls.append(url) + return self.body + + +def _scraper(body: str) -> YandexRealtyScraper: + s = YandexRealtyScraper(types.SimpleNamespace()) + s._browser = _RecordingBrowser(body) # type: ignore[assignment] + s.sleep_between_requests = _no_sleep # type: ignore[method-assign] + return s + + +async def _no_sleep() -> None: + pass + + +@pytest.mark.asyncio +async def test_on_combo_fires_for_complete_empty_combo() -> None: + """Пройденный до конца combo с ПУСТОЙ выдачей обязан дойти до on_combo — + иначе он не попадёт в чекпоинт и будет перечитываться каждым продолжением. + На origin/main on_combo вызывался только при непустых new_lots.""" + scraper = _scraper(_gate_payload([])) + seen_labels: list[str] = [] + + await scraper.fetch_around_multi_room( + 56.84, + 60.60, + 1000, + max_pages=1, + rooms_list=["room_1"], + price_ranges=[(None, 3_000_000)], + segments=["NO"], + on_combo=lambda label, lots: seen_labels.append(label), + ) + + assert seen_labels == ["vtorichka/room_1:None-3000000"], ( + "пустой, но полностью пройденный combo не дошёл до on_combo — " + "в чекпоинт он не попадёт никогда" + ) + + +@pytest.mark.asyncio +async def test_skip_combos_makes_zero_http_requests() -> None: + """Combo из чекпоинта не порождает ни одного HTTP-запроса и не зовёт on_combo.""" + scraper = _scraper(_gate_payload([])) + seen_labels: list[str] = [] + + await scraper.fetch_around_multi_room( + 56.84, + 60.60, + 1000, + max_pages=1, + rooms_list=["room_1", "room_2"], + price_ranges=[(None, 3_000_000)], + segments=["NO"], + on_combo=lambda label, lots: seen_labels.append(label), + skip_combos={"vtorichka/room_1:None-3000000"}, + ) + + browser: _RecordingBrowser = scraper._browser # type: ignore[assignment] + assert seen_labels == ["vtorichka/room_2:None-3000000"] + assert len(browser.urls) == 1, f"скипнутый combo всё равно ходил в сеть: {browser.urls}" + + +# ── планировщик: диспатч отдаёт точку в пайплайн (зеркало test_930) ───────── + + +def _candidate(**over: Any) -> types.SimpleNamespace: + base = { + "prev_id": 4117, + "prev_status": "banned", + "prev_counters": {"done_buckets": ["vtorichka/room_1:None-3000000"]}, + "same_params": True, + "age_h": 20.0, + "interval_days": "1", + } + base.update(over) + return types.SimpleNamespace(**base) + + +class _FakeDb: + def __init__(self, row: Any) -> None: + self.row = row + + def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any: + if params and "counters" in params: + return MagicMock() + return MagicMock(fetchone=lambda: self.row) + + def commit(self) -> None: + pass + + +async def test_scheduler_hands_checkpoint_to_yandex_sweep() -> None: + """_job_yandex_city_sweep передаёт resume_run_id, а не None литералом. + Красный на origin/main по значению (ключа в kwargs нет → None != 4117).""" + db = _FakeDb(_candidate()) + captured: dict[str, Any] = {} + + async def _spy(*_a: Any, **kw: Any) -> None: + captured.update(kw) + + with patch.object(sched, "run_yandex_city_sweep", _spy): + await sched._job_yandex_city_sweep(db, 5000, {}, MagicMock()) + + assert captured.get("resume_run_id") == 4117, ( + "планировщик не отдал чекпоинт яндекс-свипу — прогон пойдёт с нуля" + ) + + +# ── пайплайн: проводка skip→scraper и done_buckets→heartbeat ──────────────── + + +class _SweepFakeDb: + """Резюм-SELECT отдаёт counters предшественника; UPDATE'ы записываются.""" + + def __init__(self, prev_counters: dict[str, Any]) -> None: + self.prev_counters = prev_counters + self.heartbeats: list[dict[str, Any]] = [] + + def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any: + if params and "counters" in params: + self.heartbeats.append(json.loads(params["counters"])) + return MagicMock() + if params and "rid" in params: + return MagicMock(fetchone=lambda: types.SimpleNamespace(counters=self.prev_counters)) + return MagicMock() # is_cancelled/сторожа — best-effort, глотаем + + def commit(self) -> None: + pass + + def rollback(self) -> None: + pass + + +class _FakeSweepScraper: + """Двойник YandexRealtyScraper для пайплайна: фиксирует kwargs fetch'а, + отдаёт один пройденный пустой combo через on_combo.""" + + captured: dict[str, Any] = {} # noqa: RUF012 — тестовый сборник kwargs + gate_fetch_attempts = 1 + gate_fetch_failures = 0 + + def __init__(self, *_a: Any, **_kw: Any) -> None: + pass + + async def __aenter__(self) -> _FakeSweepScraper: + return self + + async def __aexit__(self, *_exc: Any) -> None: + return None + + async def fetch_around_multi_room(self, *_a: Any, **kw: Any) -> list: + _FakeSweepScraper.captured = dict(kw) + on_combo = kw.get("on_combo") + if on_combo is not None: + on_combo("vtorichka/room_2:None-3000000", []) + return [] + + +async def test_pipeline_resumes_and_checkpoints() -> None: + """run_yandex_city_sweep: чекпоинт предшественника уезжает в scraper как + skip_combos, пройденный combo дописывается в done_buckets heartbeat'а.""" + from scraper_kit.orchestration import pipeline as pl + + prev = {"done_buckets": ["vtorichka/room_1:None-3000000"]} + db = _SweepFakeDb(prev) + + with ( + patch.object(pl, "YandexRealtyScraper", _FakeSweepScraper), + patch.object(pl.runs, "is_cancelled", lambda *_a: False), + ): + await pl.run_yandex_city_sweep( + db, # type: ignore[arg-type] + run_id=5001, + config=types.SimpleNamespace(scraper_proxy_url=None), + matcher=MagicMock(), + enrichment=MagicMock(), + enrich_address=False, + resume_run_id=4999, + ) + + assert _FakeSweepScraper.captured.get("skip_combos") == {"vtorichka/room_1:None-3000000"}, ( + "чекпоинт предшественника не доехал до scraper'а" + ) + + with_ckpt = [hb for hb in db.heartbeats if "done_buckets" in hb] + assert with_ckpt, "ни один heartbeat не унёс done_buckets — чекпоинт не персистится" + assert with_ckpt[-1]["done_buckets"] == [ + "vtorichka/room_1:None-3000000", + "vtorichka/room_2:None-3000000", + ], "done_buckets не аккумулирует пройденный combo поверх унаследованных" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index c0dd2bf2..c8520c19 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -2078,6 +2078,7 @@ async def run_yandex_city_sweep( price_ranges: list[tuple[int | None, int | None]] | None = None, segments: list[str] | None = None, region_code: int = DEFAULT_REGION_CODE, + resume_run_id: int | None = None, ) -> YandexCitySweepCounters: """Yandex.Недвижимость city sweep: rooms × price combos от центра ЕКБ → save → address-enrich. @@ -2169,6 +2170,39 @@ async def run_yandex_city_sweep( _proxy_url = config.scraper_proxy_url _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None + # ── Checkpoint/resume (#3074): combo — естественная единица обхода ──────── + # Ключ чекпоинта — combo_label ("сегмент/комнатность:lo:hi"), якоря в нём нет, + # поэтому подхват включается ТОЛЬКО при единственном якоре (штатный прод-режим: + # и ЕКБ-центр, и city-свипы — ровно один anchor). Multi-anchor — легаси/ручной + # режим, у него один combo_label повторяется на каждом якоре и пропуск был бы + # пропуском ЧУЖИХ якорей. + skip_combos: set[str] = set() + if resume_run_id is not None and len(_anchors) == 1: + _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_combos = set(_prev_counters.get("done_buckets", [])) + logger.info( + "yandex-sweep run_id=%d: resuming from run %s — %d combos already done", + run_id, + resume_run_id, + len(skip_combos), + ) + elif resume_run_id is not None: + logger.warning( + "yandex-sweep run_id=%d: resume от run %s ОТКЛОНЁН — %d якорей (>1), " + "combo-чекпоинт применим только к единственному якорю", + run_id, + resume_run_id, + len(_anchors), + ) + done_combos: set[str] = set(skip_combos) + try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): # ── Cooperative cancel ─────────────────────────────────────────── @@ -2231,7 +2265,21 @@ async def run_yandex_city_sweep( new_lots: list[ScrapedLot], _accumulator: list[ScrapedLot] = _al, ) -> None: - """Callback: вызывается scraper'ом после каждого combo с новыми лотами.""" + """Callback: после каждого ПРОЙДЕННОГО combo (и пустого — #3074). + + Вызов означает «combo пройден до конца» — фиксируем его в + чекпоинт done_combos и пишем в heartbeat (мерж jsonb: другие + heartbeat'ы без ключа его не затирают). Пустой new_lots не + гоняет save_listings. + """ + done_combos.add(combo_label) + if not new_lots: + runs.update_heartbeat( + db, + run_id, + {**counters.to_dict(), "done_buckets": sorted(done_combos)}, + ) + return _accumulator.extend(new_lots) counters.lots_fetched += len(new_lots) try: @@ -2259,7 +2307,9 @@ async def run_yandex_city_sweep( db.rollback() except Exception: pass - runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.update_heartbeat( + db, run_id, {**counters.to_dict(), "done_buckets": sorted(done_combos)} + ) logger.debug( "yandex-sweep run_id=%d combo %s: saved %d lots " "(ins=%d upd=%d total_fetched=%d)", @@ -2285,6 +2335,7 @@ async def run_yandex_city_sweep( price_ranges=_price_ranges, segments=_segments, on_combo=_on_combo, + skip_combos=skip_combos or None, ) # #2625: аккумулируем run-level gate-API attempts/failures. yandex_gate_attempts += scraper.gate_fetch_attempts diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py index d3e0446b..3a65a109 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py @@ -891,6 +891,7 @@ async def _job_yandex_city_sweep( radius_m=int(params.get("radius_m", 1500)), enrich_address=bool(params.get("enrich_address", True)), segments=params.get("segments"), + resume_run_id=_pick_resume(db, run_id), ) 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 10802773..c4f97bd0 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 @@ -848,6 +848,7 @@ class YandexRealtyScraper(BaseScraper): price_ranges: list[tuple[int | None, int | None]] | None = None, segments: list[str] | None = None, on_combo: Any = None, + skip_combos: Any = None, **_legacy_kwargs: Any, ) -> list[ScrapedLot]: """Fetch via segment x rooms x price-range combos; paginate each combo to totalPages. @@ -860,9 +861,16 @@ class YandexRealtyScraper(BaseScraper): ["NO", "YES"] to sweep both vtorichka and novostroyki in one call. on_combo: опциональный callback(combo_label: str, new_lots: list[ScrapedLot]) -> None. - Вызывается после каждого combo с новыми (не повторными) лотами из него. - Позволяет инкрементальный save: вызывающий код сохраняет new_lots сразу, + Вызывается после каждого НЕ скипнутого combo — в том числе с пустым + new_lots (#3074: полностью пройденный combo без новых лотов обязан + попасть в чекпоинт, иначе его перечитывали бы вечно). Позволяет + инкрементальный save: вызывающий код сохраняет new_lots сразу, не дожидаясь конца sweep'а. Дубли между combo НЕ передаются повторно. + Combo, оборванный отказом (gate/extraction failure), on_combo НЕ + вызывает — вызов означает «combo пройден до конца». + skip_combos: опциональная коллекция combo_label, которые уже собраны + предыдущим оборванным прогоном (#3074, чекпоинт done_buckets) — по + ним не делается ни одного HTTP-запроса и on_combo не вызывается. """ seen: dict[str, ScrapedLot] = {} _segments = segments or ["NO"] @@ -879,6 +887,9 @@ class YandexRealtyScraper(BaseScraper): _seg = _segment_for(new_flat) for rooms, price_min, price_max in combos: combo_label = f"{_seg}/{_combo_label(rooms, price_min, price_max)}" + if skip_combos and combo_label in skip_combos: + # #3074: собрано предыдущим оборванным прогоном — ни одного запроса. + continue total_pages: int | None = None combo_new_lots: list[ScrapedLot] = [] combo_skipped = False @@ -1012,9 +1023,10 @@ class YandexRealtyScraper(BaseScraper): seen[key] = lot combo_new_lots.append(lot) - # Инкрементальный save: вызываем on_combo если есть новые лоты. - # Скипнутые combo (combo_skipped=True) не вызывают on_combo. - if on_combo is not None and combo_new_lots and not combo_skipped: + # Инкрементальный save + чекпоинт: on_combo для КАЖДОГО пройденного + # combo, включая пустые (#3074) — вызов означает «combo пройден до + # конца». Оборванные отказом (combo_skipped=True) не вызывают. + if on_combo is not None and not combo_skipped: try: on_combo(combo_label, combo_new_lots) except Exception: -- 2.45.3