From dc793e8701e67cf419bbda88e031de91b578e5ed Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 28 Aug 2026 00:48:55 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein/yandex):=20=D1=87=D0=B5=D0=BA=D0=BF?= =?UTF-8?q?=D0=BE=D0=B8=D0=BD=D1=82=20combo=20=D1=81=D1=82=D0=B0=D0=B2?= =?UTF-8?q?=D0=B8=D0=BB=D1=81=D1=8F=20=D0=B4=D0=BE=20save=5Flistings,=20?= =?UTF-8?q?=D0=BD=D0=B5=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _on_combo в run_yandex_city_sweep делал done_combos.add(combo_label) ДО вызова save_listings. Отказ save_listings перехватывается (осознанно — одна упавшая единица не должна ронять весь sweep) и логируется, но combo уже был отмечен пройденным и уходил в heartbeat done_buckets. Следующий resume брал skip_combos из done_buckets (членство в множестве — само по себе корректно, не трогал) и пропускал этот combo навсегда: молча, прогон завершался штатно, просто сегмент выдачи не собирался никогда. Тот же инвариант "отмечаем пройденным только после успешного save", что уже есть у страницы в run_avito_newbuilding_sweep (_saved_ok), якоря в run_avito_city_sweep (_anchor_ok) и бакета в run_cian_full_load (_mark_bucket) — применил к combo. Heartbeat пишется в любом случае (и при отказе save тоже), иначе reap_zombies посчитает живой прогон мёртвым. Тесты: test_3170_yandex_combo_checkpoint.py — combo с упавшим save не попадает в done_buckets, успешный (включая пустую выдачу) — попадает. Обратимость проверена: с возвращённым дефектом (git stash) первый тест красный, со снятым — зелёный вместе с существующим test_3074_yandex_ sweep_checkpoint.py (6/6). --- .../test_3170_yandex_combo_checkpoint.py | 168 ++++++++++++++++++ .../src/scraper_kit/orchestration/pipeline.py | 81 +++++---- 2 files changed, 212 insertions(+), 37 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_3170_yandex_combo_checkpoint.py diff --git a/tradein-mvp/backend/tests/test_3170_yandex_combo_checkpoint.py b/tradein-mvp/backend/tests/test_3170_yandex_combo_checkpoint.py new file mode 100644 index 00000000..661db802 --- /dev/null +++ b/tradein-mvp/backend/tests/test_3170_yandex_combo_checkpoint.py @@ -0,0 +1,168 @@ +"""#3170: чекпоинт combo в yandex_city_sweep обязан ставиться ПОСЛЕ save, не до. + +На origin/main `_on_combo` (run_yandex_city_sweep) делал `done_combos.add(combo_label)` +ДО вызова save_listings. Отказ save_listings перехватывался (`except Exception`), +логировался, шёл db.rollback() — и прогон продолжался, но combo уже был отмечен +пройденным и уходил в heartbeat done_buckets. Следующий resume брал skip_combos +из done_buckets (членство в множестве, само по себе корректно) и пропускал этот +combo НАВСЕГДА — молча, прогон завершался штатно, просто сегмент выдачи (сегмент +× комнатность × ценовой диапазон) не собирался никогда. + +Тот же инвариант уже был закреплён для страницы в run_avito_newbuilding_sweep +(_saved_ok, см. test_3074_avito_newbuilding_checkpoint.py), якоря в +run_avito_city_sweep (_anchor_ok) и бакета в run_cian_full_load (_mark_bucket) — +здесь применяем его к combo в yandex-свипе. + +ИНВАРИАНТЫ, РАДИ КОТОРЫХ ТЕСТ: +1. Combo, чей save_listings упал, НЕ попадает в done_buckets (чекпоинт). +2. Combo, пройденный успешно (включая пустую выдачу — #3074), попадает. +3. Heartbeat пишется в любом случае (и при отказе save тоже) — иначе + reap_zombies посчитает живой прогон мёртвым. +""" + +from __future__ import annotations + +import os + +# Settings собирается автофикстурой conftest'а и требует database_url. Выставляем +# до остальных импортов — так же, как в test_3074_yandex_sweep_checkpoint.py. +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 + + +class _SweepFakeDb: + """Резюм-SELECT отдаёт counters предшественника; heartbeat'ы записываются.""" + + def __init__(self, prev_counters: dict[str, Any] | None = None) -> None: + self.prev_counters = prev_counters or {} + 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: ... + def rollback(self) -> None: ... + + +def _lot(tag: str) -> MagicMock: + m = MagicMock() + m.source_id = tag + return m + + +class _FakeSweepScraper: + """Двойник YandexRealtyScraper: прогоняет заданный список combo через on_combo.""" + + combos: list[tuple[str, list[Any]]] = [] # noqa: RUF012 — тестовый сборник + 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: + on_combo = kw.get("on_combo") + if on_combo is not None: + for label, lots in _FakeSweepScraper.combos: + on_combo(label, lots) + return [] + + +def _make_save(fail_labels: set[str]): + """save_listings, падающий для лотов указанных combo-меток (тег лота == label).""" + + def _save(_db, new_lots, **_kw): + tags = {lot.source_id for lot in new_lots} + if tags & fail_labels: + raise RuntimeError(f"save_listings упал: {tags & fail_labels}") + return (len(new_lots), 0) + + return _save + + +async def _run(db: _SweepFakeDb, *, fail_labels: set[str]) -> _SweepFakeDb: + from scraper_kit.orchestration import pipeline as pl + + # record_yandex_price_history складывается в counters.price_history_rows (int) — + # дефолтный MagicMock() возврата не годится для json.dumps в heartbeat. + enrichment = MagicMock() + enrichment.record_yandex_price_history.return_value = 0 + + with ( + patch.object(pl, "YandexRealtyScraper", _FakeSweepScraper), + patch.object(pl.runs, "is_cancelled", lambda *_a: False), + patch.object(pl, "save_listings", _make_save(fail_labels)), + ): + await pl.run_yandex_city_sweep( + db, # type: ignore[arg-type] + run_id=9001, + config=types.SimpleNamespace(scraper_proxy_url=None), + matcher=MagicMock(), + enrichment=enrichment, + enrich_address=False, + resume_run_id=None, + ) + return db + + +def _last_checkpoint(db: _SweepFakeDb) -> list[str]: + with_ckpt = [hb for hb in db.heartbeats if "done_buckets" in hb] + assert with_ckpt, "ни один heartbeat не унёс done_buckets — чекпоинт не персистится" + return with_ckpt[-1]["done_buckets"] + + +@pytest.mark.asyncio +async def test_combo_with_failed_save_does_not_enter_checkpoint() -> None: + """Combo, чей save_listings упал, НЕ должен попасть в done_buckets. + + Иначе следующий resume возьмёт его из skip_combos и пропустит навсегда — + молча, потому что прогон завершится штатно. + """ + label_ok = "vtorichka/room_1:None-3000000" + label_fail = "vtorichka/room_2:None-3000000" + _FakeSweepScraper.combos = [ + (label_ok, [_lot(label_ok)]), + (label_fail, [_lot(label_fail)]), + ] + + db = await _run(_SweepFakeDb(), fail_labels={label_fail}) + + ckpt = _last_checkpoint(db) + assert label_fail not in ckpt, "combo с упавшим save попал в чекпоинт — покрытие потеряно молча" + assert label_ok in ckpt + + +@pytest.mark.asyncio +async def test_combo_saved_successfully_enters_checkpoint() -> None: + """Combo, пройденный до конца (save успешен, включая пустую выдачу), + обязан попасть в done_buckets — иначе следующий resume перечитывает + уже собранное вечно.""" + label_nonempty = "vtorichka/room_1:None-3000000" + label_empty = "vtorichka/room_2:None-3000000" + _FakeSweepScraper.combos = [ + (label_nonempty, [_lot(label_nonempty)]), + (label_empty, []), # #3074: пустой, но полностью пройденный combo + ] + + db = await _run(_SweepFakeDb(), fail_labels=set()) + + ckpt = _last_checkpoint(db) + assert label_nonempty in ckpt + assert label_empty in ckpt, "пустой, но пройденный 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 97e70169..9e92f8eb 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 @@ -2382,46 +2382,53 @@ async def run_yandex_city_sweep( ) -> None: """Callback: после каждого ПРОЙДЕННОГО combo (и пустого — #3074). - Вызов означает «combo пройден до конца» — фиксируем его в - чекпоинт done_combos и пишем в heartbeat (мерж jsonb: другие - heartbeat'ы без ключа его не затирают). Пустой new_lots не - гоняет save_listings. + В чекпоинт done_combos (и, соответственно, в heartbeat + done_buckets, откуда resume берёт skip_combos) попадает + ТОЛЬКО combo, чьи лоты реально сохранены (#3170). Отказ + save_listings перехвачен ниже и прогон продолжается — но + отметить combo пройденным ДО успешного save значило бы, что + следующий прогон пропустит его НАВСЕГДА, причём молча: прогон + завершится штатно. Тот же инвариант, что у страницы в + run_avito_newbuilding_sweep (_saved_ok), якоря в + run_avito_city_sweep (_anchor_ok) и бакета в run_cian_full_load + (_mark_bucket). Heartbeat пишем в любом случае (мерж jsonb: + другие heartbeat'ы без ключа его не затирают) — иначе + reap_zombies посчитает живой прогон мёртвым. """ - done_combos.add(combo_label) + _saved_ok = False 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: - ins, upd = save_listings( - db, - new_lots, - matcher=matcher, - region_code=region_code, - run_id=run_id, - city=_city_name, - city_anchor=_city_anchor_point, - city_radius_km=_city_radius_km, - ) - counters.lots_inserted += ins - counters.lots_updated += upd - except Exception as save_exc: - counters.errors_count += 1 - logger.warning( - "yandex-sweep run_id=%d combo %s: save_listings failed: %s", - run_id, - combo_label, - save_exc, - ) + _saved_ok = True # пустая выдача — тоже валидно пройденный combo + else: + _accumulator.extend(new_lots) + counters.lots_fetched += len(new_lots) try: - db.rollback() - except Exception: - pass + ins, upd = save_listings( + db, + new_lots, + matcher=matcher, + region_code=region_code, + run_id=run_id, + city=_city_name, + city_anchor=_city_anchor_point, + city_radius_km=_city_radius_km, + ) + counters.lots_inserted += ins + counters.lots_updated += upd + _saved_ok = True + except Exception as save_exc: + counters.errors_count += 1 + logger.warning( + "yandex-sweep run_id=%d combo %s: save_listings failed: %s", + run_id, + combo_label, + save_exc, + ) + try: + db.rollback() + except Exception: + pass + if _saved_ok: + done_combos.add(combo_label) runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done_combos)} )