From ef0b0d0b3a02b7a4490284a26fd2a38e87b48417 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Fri, 29 May 2026 13:49:28 +0500 Subject: [PATCH] feat(scheduler): add dormant Yandex Realty city-sweep (#561) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire a full Yandex.Недвижимость periodic city-sweep into the in-app scheduler, shipped DISABLED (enabled=false) by design — live behavior is intentionally unverified pending egress-IP rotation (#623). Mirrors the existing avito_city_sweep (also currently disabled in prod). - run_yandex_city_sweep in scrape_pipeline.py: mirrors run_avito_city_sweep but simpler (no houses/detail enrich) — search + save_listings per anchor, cooperative cancel, per-anchor exception isolation, defensive abort after 3 consecutive anchor failures (mark_banned; YandexRealtyScraper has no block/rate exception classes — fetch_around swallows errors → []). - trigger_yandex_city_sweep_run + dispatch branch in scheduler.py (faithful mirror of the avito trigger: claim run, fresh SessionLocal, create_task). - Migration 078 seeds scrape_schedules row enabled=false, window 02:00-05:00 UTC, conservative params {pages_per_anchor:2, request_delay_sec:9, radius_m:1500}, ON CONFLICT (source) DO NOTHING (idempotent). - 8 unit tests (mocked scraper + save_listings, no live network/DB): anchor iteration, cancel mid-sweep, per-anchor isolation, consecutive- failure abort + reset-on-success, empty-page pagination break. Enable later (after #623): UPDATE scrape_schedules SET enabled=true WHERE source='yandex_city_sweep'; Closes #561 --- tradein-mvp/backend/app/services/scheduler.py | 43 +- .../backend/app/services/scrape_pipeline.py | 154 ++++++++ ...078_scrape_schedules_seed_yandex_sweep.sql | 55 +++ .../backend/tests/test_yandex_city_sweep.py | 372 ++++++++++++++++++ 4 files changed, 623 insertions(+), 1 deletion(-) create mode 100644 tradein-mvp/backend/data/sql/078_scrape_schedules_seed_yandex_sweep.sql create mode 100644 tradein-mvp/backend/tests/test_yandex_city_sweep.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 68acd4ec..01c4759b 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -9,6 +9,7 @@ Sources: - avito_city_sweep → run_avito_city_sweep (scrape_pipeline.py) + - yandex_city_sweep → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT) - cian_history_backfill → run_cian_backfill (this module, #560) - rosreestr_dkp_import → import_rosreestr_dkp (this module, #563) """ @@ -26,7 +27,7 @@ from sqlalchemy.orm import Session from app.core.db import SessionLocal from app.services import scrape_runs as runs_mod -from app.services.scrape_pipeline import run_avito_city_sweep +from app.services.scrape_pipeline import run_avito_city_sweep, run_yandex_city_sweep logger = logging.getLogger(__name__) @@ -186,6 +187,44 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any] return run_id +async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать новый scrape_runs + launch run_yandex_city_sweep в asyncio.create_task. + + Зеркало trigger_avito_city_sweep_run, но проще (#561): у Yandex sweep нет + houses/detail-параметров. Shipped DORMANT — schedule seed enabled=false (078), + включается оператором вручную после #623 (egress-IP rotation). + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick) + async def _run() -> None: + run_db = SessionLocal() + try: + await run_yandex_city_sweep( + run_db, + run_id=run_id, + pages_per_anchor=int(params.get("pages_per_anchor", 2)), + request_delay_sec=float(params.get("request_delay_sec", 9.0)), + radius_m=int(params.get("radius_m", 1500)), + ) + except Exception: + logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + # Keep reference to avoid GC before task completes (RUF006) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered yandex_city_sweep run_id=%d", run_id) + return run_id + + async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch run_cian_backfill в asyncio.create_task. @@ -592,6 +631,8 @@ async def scheduler_loop() -> None: source = sch["source"] if source == "avito_city_sweep": await trigger_avito_city_sweep_run(db, sch) + elif source == "yandex_city_sweep": + await trigger_yandex_city_sweep_run(db, sch) elif source == "cian_history_backfill": await trigger_cian_backfill_run(db, sch) elif source == "rosreestr_dkp_import": diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index a6198a60..91b19878 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -40,6 +40,7 @@ from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichm from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment from app.services.scrapers.base import ScrapedLot, save_listings +from app.services.scrapers.yandex_realty import YandexRealtyScraper logger = logging.getLogger(__name__) @@ -424,12 +425,165 @@ async def run_avito_city_sweep( raise +# ── Yandex city sweep ─────────────────────────────────────────── + + +@dataclass +class YandexCitySweepCounters: + """Aggregate counters для Yandex.Недвижимость city sweep run. + + Проще avito CitySweepCounters: у Yandex SERP нет house-catalog / detail-enrich + шага (только search → save), поэтому houses/detail-полей нет. + """ + + anchors_total: int = 0 + anchors_done: int = 0 + pages_fetched: int = 0 + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +# Defensive abort threshold: у YandexRealtyScraper нет block/rate-exception классов +# (fetch_around глотает ошибки и возвращает []). Чтобы не молотить Yandex впустую при +# системном блоке/бане IP, прерываем sweep после N подряд anchor'ов, упавших с +# исключением. NB: пустой результат (0 lots) — это НЕ ошибка, счётчик не растёт. +YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 + + +async def run_yandex_city_sweep( + db: Session, + *, + run_id: int, + anchors: list[tuple[float, float, str]] | None = None, + radius_m: int = 1500, + pages_per_anchor: int = 2, + request_delay_sec: float | None = None, +) -> YandexCitySweepCounters: + """Yandex.Недвижимость city sweep: iterate anchors × pages → save. + + Проще avito sweep — у Yandex SERP нет house-catalog / detail-enrich шага: + per anchor: YandexRealtyScraper().fetch_around(lat, lon, radius_m, page=p) + → save_listings(db, lots) → accumulate counters. + + Anti-bot / cancel семантика (зеркало avito sweep, см. #561): + - Cooperative cancel: is_cancelled(db, run_id) проверяется перед каждым anchor; + при cancel — update_heartbeat + return (mark_cancelled делает caller-flow по статусу). + - YandexRealtyScraper НЕ определяет block/rate exception классов (fetch_around + ловит исключения внутри и возвращает []). Поэтому per-anchor ловим broad + Exception, логируем и continue; защитно прерываем sweep (early abort + mark_banned), + если N anchor'ов ПОДРЯД упали с исключением — это сигнал системного блока IP. + - request_delay_sec: YandexRealtyScraper.fetch_around НЕ принимает delay-override + (он сам спит self.request_delay_sec между страницами через sleep_between_requests). + Поэтому request_delay_sec применяется здесь как доп. asyncio.sleep МЕЖДУ anchor'ами. + + Возвращает YandexCitySweepCounters (агрегат по всем anchor'ам). + """ + _anchors = anchors if anchors is not None else EKB_ANCHORS + counters = YandexCitySweepCounters(anchors_total=len(_anchors)) + inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0 + consecutive_failures = 0 + + try: + for idx, (lat, lon, name) in enumerate(_anchors, start=1): + if scrape_runs.is_cancelled(db, run_id): + logger.info( + "yandex-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", + run_id, idx, len(_anchors), name, + ) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + return counters + + logger.info( + "yandex-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", + run_id, idx, len(_anchors), name, lat, lon, + ) + try: + # Fresh scraper per anchor — uses its own httpx.AsyncClient via the + # async-context-manager. fetch_around fetches ONE SERP page; loop pages. + anchor_lots: list[ScrapedLot] = [] + async with YandexRealtyScraper() as scraper: + for page in range(pages_per_anchor): + page_lots = await scraper.fetch_around( + lat, lon, radius_m, page=page + ) + counters.pages_fetched += 1 + if not page_lots: + # Пустая страница → дальше результатов нет, прекращаем пагинацию. + break + anchor_lots.extend(page_lots) + + counters.lots_fetched += len(anchor_lots) + if anchor_lots: + inserted, updated = save_listings(db, anchor_lots, run_id=run_id) + counters.lots_inserted += inserted + counters.lots_updated += updated + consecutive_failures = 0 + except Exception as e: + # YandexRealtyScraper не propagate'ит block/rate exceptions, поэтому здесь + # ловим broad Exception per-anchor: логируем и continue (изоляция anchor'а). + logger.exception( + "yandex-sweep run_id=%d: anchor %s failed", run_id, name + ) + counters.errors_count += 1 + consecutive_failures += 1 + if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive failures (IP likely blocked): %s", + run_id, idx, len(_anchors), name, consecutive_failures, e, + ) + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_banned( + db, + run_id, + f"yandex sweep aborted: {consecutive_failures} consecutive " + f"anchor failures (last: {e})", + counters.to_dict(), + ) + return counters + + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + + # Доп. пауза между anchor'ами (поверх per-page sleep внутри scraper'а). + if idx < len(_anchors): + await asyncio.sleep(inter_anchor_delay) + + scrape_runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "yandex-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d " + "(ins=%d/upd=%d) errors=%d", + run_id, + counters.anchors_done, + counters.anchors_total, + counters.pages_fetched, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("yandex-sweep run_id=%d: fatal error", run_id) + scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + # ── Public re-exports ─────────────────────────────────────────── __all__ = [ "EKB_ANCHORS", "CitySweepCounters", "PipelineCounters", "PipelineResult", + "YandexCitySweepCounters", "run_avito_city_sweep", "run_avito_pipeline", + "run_yandex_city_sweep", ] diff --git a/tradein-mvp/backend/data/sql/078_scrape_schedules_seed_yandex_sweep.sql b/tradein-mvp/backend/data/sql/078_scrape_schedules_seed_yandex_sweep.sql new file mode 100644 index 00000000..c83c23dd --- /dev/null +++ b/tradein-mvp/backend/data/sql/078_scrape_schedules_seed_yandex_sweep.sql @@ -0,0 +1,55 @@ +-- 078_scrape_schedules_seed_yandex_sweep.sql +-- Seed row для Yandex.Недвижимость периодического city-sweep (#561). +-- +-- !!! DORMANT BY DESIGN !!! enabled = false. +-- Sweep будет бить Yandex с prod-IP → anti-bot риск ("осторожнее с лимитами"). +-- Capability полностью wired (run_yandex_city_sweep + trigger + dispatch), +-- но schedule засеян ВЫКЛЮЧЕННЫМ. Live-поведение намеренно НЕ verified. +-- ВКЛЮЧАТЬ ВРУЧНУЮ оператором ПОСЛЕ #623 (egress-IP rotation): +-- UPDATE scrape_schedules SET enabled = true WHERE source = 'yandex_city_sweep'; +-- Зеркалит existing avito_city_sweep (тоже сейчас enabled = f в prod). +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)). +-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно. +-- +-- yandex_city_sweep — окно 02:00-05:00 UTC (05:00-08:00 МСК), как avito_city_sweep. +-- default_params (консервативно, под анти-бот): +-- pages_per_anchor — страниц SERP на anchor (мало, чтобы не отсвечивать). +-- request_delay_sec — пауза между anchor'ами (поверх per-page sleep scraper'а). +-- radius_m — прокидывается в fetch_around (Yandex его игнорит: path-based URL, +-- держим для совместимости сигнатуры avito-shape sweep). +-- У Yandex sweep НЕТ detail/houses-enrich шага — поэтому detail_top_n/enrich_houses +-- в params отсутствуют (в отличие от avito_city_sweep). + +BEGIN; + +-- next_run_at = следующее наступление night-window (tomorrow + window_start_hour). +-- При enabled=false get_due_schedules() всё равно не подхватит строку, но заполняем +-- по образцу 072: когда оператор включит schedule, next_run_at уже валидный и +-- запуск не выстрелит мгновенно (вне окна). AT TIME ZONE 'UTC' фиксирует wall-clock +-- час в UTC независимо от session TimeZone GUC. +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'yandex_city_sweep', + false, -- DORMANT (#561) — enable manually after #623 + 2, + 5, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC', + '{"pages_per_anchor": 2, "request_delay_sec": 9, "radius_m": 1500}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import.'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_yandex_city_sweep.py b/tradein-mvp/backend/tests/test_yandex_city_sweep.py new file mode 100644 index 00000000..6aae9ae0 --- /dev/null +++ b/tradein-mvp/backend/tests/test_yandex_city_sweep.py @@ -0,0 +1,372 @@ +"""Offline unit tests for run_yandex_city_sweep (#561, shipped DORMANT). + +Pure & fast: NO live network, NO DB. We monkeypatch: + - YandexRealtyScraper.__init__ / __aenter__ / __aexit__ — avoid get_scraper_delay + DB read + real httpx.AsyncClient creation. + - YandexRealtyScraper.fetch_around — return a fixed list of fake ScrapedLot. + - scrape_pipeline.save_listings — counter/no-op (no real listings table). + - scrape_runs.* (is_cancelled / update_heartbeat / mark_*) — in-memory fakes. + +These tests assert orchestration semantics only; live Yandex behavior is intentionally +unverified (sweep is dormant until egress-IP rotation #623). +""" +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from typing import Any + +import pytest + +from app.services import scrape_pipeline +from app.services.scrapers.base import ScrapedLot +from app.services.scrapers.yandex_realty import YandexRealtyScraper + +# 3 anchors keep tests fast and let us probe "anchor k cancels/fails, others proceed". +TEST_ANCHORS: list[tuple[float, float, str]] = [ + (56.8400, 60.6050, "A1"), + (56.7950, 60.5300, "A2"), + (56.8970, 60.6100, "A3"), +] + + +def _fake_lot(offer_id: str) -> ScrapedLot: + """Minimal valid ScrapedLot (source/source_url required, price_rub > 0).""" + return ScrapedLot( + source="yandex", + source_url=f"https://realty.yandex.ru/offer/{offer_id}/", + source_id=offer_id, + address="Екатеринбург, улица Тестовая, 1", + price_rub=5_000_000, + area_m2=50.0, + rooms=2, + ) + + +class _FakeRuns: + """In-memory stand-in for app.services.scrape_runs (no DB).""" + + def __init__(self, *, cancel_at_anchor: int | None = None) -> None: + # cancel_at_anchor: 1-based anchor index at which is_cancelled flips True. + self.cancel_at_anchor = cancel_at_anchor + self.is_cancelled_calls = 0 + self.heartbeats: list[dict[str, int]] = [] + self.done: dict[str, int] | None = None + self.failed: tuple[str, dict[str, int]] | None = None + self.banned: tuple[str, dict[str, int]] | None = None + + def is_cancelled(self, _db: Any, _run_id: int) -> bool: + self.is_cancelled_calls += 1 + if self.cancel_at_anchor is None: + return False + return self.is_cancelled_calls >= self.cancel_at_anchor + + def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.heartbeats.append(dict(counters)) + + def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.done = dict(counters) + + def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.failed = (error, dict(counters)) + + def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.banned = (error, dict(counters)) + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None: + """Kill inter-anchor asyncio.sleep so tests are instant.""" + async def _instant(_secs: float) -> None: + return None + + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant) + + +@pytest.fixture(autouse=True) +def _stub_scraper_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + """Neutralize __init__/__aenter__/__aexit__ — no get_scraper_delay DB, no httpx.""" + + def _init(self: YandexRealtyScraper, city: str = "ekaterinburg") -> None: + self.city = city + self.request_delay_sec = 0.0 + self._client = None + + async def _aenter(self: YandexRealtyScraper) -> YandexRealtyScraper: + return self + + async def _aexit(self: YandexRealtyScraper, *_args: Any) -> None: + return None + + monkeypatch.setattr(YandexRealtyScraper, "__init__", _init) + monkeypatch.setattr(YandexRealtyScraper, "__aenter__", _aenter) + monkeypatch.setattr(YandexRealtyScraper, "__aexit__", _aexit) + + +def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None: + monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned) + + +# ── Happy path: all anchors iterated, counters aggregated ─────────────────── + + +async def test_sweep_iterates_all_anchors_and_aggregates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fetch_calls: list[tuple[float, float, int]] = [] + save_calls: list[int] = [] + + async def fake_fetch( + self: YandexRealtyScraper, + lat: float, + lon: float, + radius_m: int = 1000, + page: int = 0, + ) -> list[ScrapedLot]: + fetch_calls.append((lat, lon, page)) + # 2 lots per page; let pagination run the full pages_per_anchor. + return [_fake_lot(f"{lat}-{page}-{i}") for i in range(2)] + + def fake_save( + _db: Any, lots: list[ScrapedLot], *, run_id: int | None = None + ) -> tuple[int, int]: + save_calls.append(len(lots)) + # Pretend everything is a fresh insert. + return len(lots), 0 + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_yandex_city_sweep( + db=object(), + run_id=1, + anchors=TEST_ANCHORS, + pages_per_anchor=2, + request_delay_sec=0.0, + ) + + # fetch_around called pages_per_anchor times per anchor (no empty-page short-circuit). + assert len(fetch_calls) == len(TEST_ANCHORS) * 2 + # save_listings called once per anchor (anchor-level batch). + assert len(save_calls) == len(TEST_ANCHORS) + # Counters: 3 anchors × 2 pages × 2 lots = 12. + assert counters.anchors_total == len(TEST_ANCHORS) + assert counters.anchors_done == len(TEST_ANCHORS) + assert counters.pages_fetched == len(TEST_ANCHORS) * 2 + assert counters.lots_fetched == 12 + assert counters.lots_inserted == 12 + assert counters.lots_updated == 0 + assert counters.errors_count == 0 + # mark_done called with final counters; no failure/ban. + assert fake.done is not None + assert fake.done["lots_fetched"] == 12 + assert fake.failed is None + assert fake.banned is None + + +async def test_save_listings_receives_run_id(monkeypatch: pytest.MonkeyPatch) -> None: + """run_id must be threaded into save_listings for snapshot provenance.""" + seen_run_ids: list[int | None] = [] + + async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0): + return [_fake_lot(f"{lat}-{page}")] + + def fake_save(_db, lots, *, run_id=None): + seen_run_ids.append(run_id) + return len(lots), 0 + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + _install_runs(monkeypatch, _FakeRuns()) + + await scrape_pipeline.run_yandex_city_sweep( + db=object(), run_id=777, anchors=TEST_ANCHORS, pages_per_anchor=1, + ) + assert seen_run_ids == [777, 777, 777] + + +# ── Empty page short-circuits pagination ──────────────────────────────────── + + +async def test_empty_page_stops_pagination(monkeypatch: pytest.MonkeyPatch) -> None: + """An empty SERP page ends pagination for that anchor (no further pages).""" + pages_seen: list[int] = [] + + async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0): + pages_seen.append(page) + if page == 0: + return [_fake_lot(f"{lat}-0")] + return [] # page 1 empty → stop + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_yandex_city_sweep( + db=object(), run_id=2, anchors=TEST_ANCHORS, pages_per_anchor=5, + ) + # Per anchor: page 0 (1 lot) then page 1 (empty, break) → 2 pages each. + assert pages_seen == [0, 1, 0, 1, 0, 1] + assert counters.pages_fetched == 6 # 2 per anchor × 3 + assert counters.lots_fetched == 3 # 1 per anchor + assert counters.anchors_done == 3 + + +# ── Mid-sweep cancel stops further anchors ────────────────────────────────── + + +async def test_cancel_midway_stops_remaining_anchors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fetch_calls: list[str] = [] + + async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0): + fetch_calls.append(f"{lat}") + return [_fake_lot(f"{lat}-{page}")] + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + # is_cancelled flips True on the 2nd check (before anchor #2) → only anchor #1 runs. + fake = _FakeRuns(cancel_at_anchor=2) + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_yandex_city_sweep( + db=object(), run_id=3, anchors=TEST_ANCHORS, pages_per_anchor=1, + ) + # Only the first anchor fetched before cancel. + assert len(fetch_calls) == 1 + assert counters.anchors_done == 1 + # Cancel path returns early: no mark_done, no ban/fail. + assert fake.done is None + assert fake.failed is None + assert fake.banned is None + + +# ── Per-anchor exception is isolated (single failure → continue) ──────────── + + +async def test_single_anchor_failure_is_isolated( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Anchor #2 raises → anchors #1 and #3 still processed; sweep completes.""" + processed: list[str] = [] + + async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0): + if name_of(lat) == "A2": + raise RuntimeError("boom on anchor 2") + processed.append(name_of(lat)) + return [_fake_lot(f"{lat}-{page}")] + + def name_of(lat: float) -> str: + for a_lat, _lon, nm in TEST_ANCHORS: + if a_lat == lat: + return nm + return "?" + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_yandex_city_sweep( + db=object(), run_id=4, anchors=TEST_ANCHORS, pages_per_anchor=1, + ) + # A1 and A3 processed; A2 raised and was skipped. + assert processed == ["A1", "A3"] + assert counters.errors_count == 1 + assert counters.anchors_done == 3 # loop advanced past all 3 anchors + assert counters.lots_fetched == 2 # only A1 + A3 contributed lots + # Not an abort — sweep marked done despite the isolated failure. + assert fake.done is not None + assert fake.banned is None + + +# ── Defensive abort: N consecutive failures → mark_banned, stop early ─────── + + +async def test_consecutive_failures_abort_sweep( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """All anchors raise → after MAX_CONSECUTIVE_FAILURES the sweep aborts + bans.""" + attempts: list[str] = [] + + async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0): + attempts.append(f"{lat}") + raise RuntimeError("yandex blocked") + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + # 5 anchors so the 3-consecutive-failure threshold trips before the list ends. + anchors = [ + *TEST_ANCHORS, + (56.7700, 60.5500, "A4"), + (56.8650, 60.6200, "A5"), + ] + counters = await scrape_pipeline.run_yandex_city_sweep( + db=object(), run_id=5, anchors=anchors, pages_per_anchor=1, + ) + # Abort exactly at the threshold (3 consecutive failures), not all 5 anchors. + assert len(attempts) == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES + assert counters.errors_count == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES + assert counters.anchors_done == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES + assert fake.banned is not None + assert "consecutive" in fake.banned[0] + assert fake.done is None # aborted, never marked done + + +async def test_non_consecutive_failures_do_not_abort( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A success between two failures resets the consecutive counter → no abort.""" + def name_of(lat: float, anchors: list[tuple[float, float, str]]) -> str: + for a_lat, _lon, nm in anchors: + if a_lat == lat: + return nm + return "?" + + anchors = [ + *TEST_ANCHORS, + (56.7700, 60.5500, "A4"), + (56.8650, 60.6200, "A5"), + ] + # Fail A1, succeed A2, fail A3, succeed A4, fail A5 → never 3-in-a-row. + failing = {"A1", "A3", "A5"} + + async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0): + if name_of(lat, anchors) in failing: + raise RuntimeError("transient") + return [_fake_lot(f"{lat}-{page}")] + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_yandex_city_sweep( + db=object(), run_id=6, anchors=anchors, pages_per_anchor=1, + ) + assert counters.errors_count == 3 # A1, A3, A5 + assert counters.anchors_done == 5 # walked the whole list, no early abort + assert counters.lots_fetched == 2 # A2 + A4 contributed + assert fake.banned is None + assert fake.done is not None -- 2.45.3