diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index b0c58f31..66558735 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -49,6 +49,12 @@ from app.services.scrapers.yandex_realty import YandexRealtyScraper logger = logging.getLogger(__name__) +# Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает +# дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for +# прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом. +# Диапазон 180-300 с; 240 — разумный середина (4 мин > любой нормальный anchor). +ANCHOR_TIMEOUT_SEC: int = 240 + # Default anchors ЕКБ — 5 точек покрытия города EKB_ANCHORS: list[tuple[float, float, str]] = [ (56.8400, 60.6050, "Центр"), @@ -455,16 +461,19 @@ async def run_avito_city_sweep( lon, ) try: - result = await run_avito_pipeline( - db, - lat=lat, - lon=lon, - radius_m=radius_m, - enrich_houses=enrich_houses, - enrich_detail_top_n=detail_top_n, - pages=pages_per_anchor, - request_delay_sec=request_delay_sec, - shared_session=session, + result = await asyncio.wait_for( + run_avito_pipeline( + db, + lat=lat, + lon=lon, + radius_m=radius_m, + enrich_houses=enrich_houses, + enrich_detail_top_n=detail_top_n, + pages=pages_per_anchor, + request_delay_sec=request_delay_sec, + shared_session=session, + ), + timeout=ANCHOR_TIMEOUT_SEC, ) counters.lots_fetched += result.counters.lots_fetched counters.lots_inserted += result.counters.lots_inserted @@ -477,6 +486,19 @@ async def run_avito_city_sweep( counters.detail_failed += result.counters.detail_failed counters.errors_count += len(result.counters.errors) all_touched_house_ids.update(result.touched_house_ids) + except TimeoutError: + logger.warning( + "city-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " + "timed out after %ds — skipping", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + ANCHOR_TIMEOUT_SEC, + ) + counters.errors_count += 1 except (AvitoBlockedError, AvitoRateLimitedError) as e: logger.error( "city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s", @@ -672,13 +694,30 @@ async def run_yandex_city_sweep( lon, ) - # ── Phase 1+2: SERP + save ─────────────────────────────────────── + # ── Per-anchor phases под watchdog-таймаутом ──────────────────── + # Весь SERP + address-enrich одного anchor'а оборачиваем в wait_for. + # TimeoutError → log + errors_count++ + continue (sweep не прерывается). + anchor_lots: list[ScrapedLot] = [] - try: + _anchor_timed_out = False + # Capture loop variables in default args to satisfy B023 (flake8/ruff): + # без этого inner async def захватывает изменяемые переменные цикла + # по ссылке → неправильные значения при asyncio scheduling. + _lat, _lon, _name = lat, lon, name + + async def _yandex_anchor_phases( + _a_lat: float = _lat, + _a_lon: float = _lon, + _a_name: str = _name, + ) -> None: + """Все фазы одного anchor'а (SERP + address-enrich).""" + nonlocal anchor_lots, consecutive_failures + + # ── Phase 1+2: SERP + save ───────────────────────────────── async with YandexRealtyScraper() as scraper: anchor_lots = await scraper.fetch_around_multi_room( - lat, - lon, + _a_lat, + _a_lon, radius_m, max_pages=pages_per_anchor, ) @@ -691,11 +730,170 @@ async def run_yandex_city_sweep( logger.info( "yandex-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d", run_id, - name, + _a_name, len(anchor_lots), counters.lots_inserted, counters.lots_updated, ) + + # ── Phase 3: address-enrich ──────────────────────────────── + if not (enrich_address and anchor_lots): + return + source_ids = [lot.source_id for lot in anchor_lots if lot.source_id] + if not source_ids: + return + id_rows = ( + db.execute( + _text(""" + SELECT id, source_id, address, source_url + FROM listings + WHERE source = 'yandex' + AND source_id = ANY(CAST(:ids AS text[])) + AND address IS NOT NULL + AND NOT (address ~ ',\\s*\\d+') + AND source_url IS NOT NULL + """), + {"ids": source_ids}, + ) + .mappings() + .all() + ) + items = [dict(r) for r in id_rows] + counters.address_attempted += len(items) + + if items: + from curl_cffi.requests import AsyncSession as _AsyncSession + + async with _AsyncSession( + impersonate="chrome120", + timeout=30.0, + proxies=_proxies, + headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"}, + ) as enrich_session: + for eidx, item in enumerate(items): + lid: int = item["id"] + old_addr: str = item["address"] + url: str = item["source_url"] + try: + resp = await enrich_session.get(url, allow_redirects=True) + if resp.status_code != 200: + logger.warning( + "yandex-sweep run_id=%d address-enrich:" + " HTTP %d listing_id=%d", + run_id, + resp.status_code, + lid, + ) + counters.address_failed += 1 + else: + new_addr = _extract_address_from_title(resp.text) + if ( + new_addr is None + or new_addr == old_addr + or not _RE_HAS_HOUSE_NUMBER.search(new_addr) + ): + pass + else: + try: + db.execute( + _text(""" + UPDATE listings + SET address = :addr, + geocode_tried_at = NULL + WHERE id = CAST(:id AS bigint) + """), + {"addr": new_addr, "id": lid}, + ) + db.commit() + counters.address_enriched += 1 + logger.info( + "yandex-sweep run_id=%d: " + "address updated listing_id=%d " + "%r -> %r", + run_id, + lid, + old_addr, + new_addr, + ) + except Exception as upd_exc: + counters.address_failed += 1 + counters.errors_count += 1 + logger.warning( + "yandex-sweep run_id=%d: " + "address DB update failed " + "listing_id=%d: %s", + run_id, + lid, + upd_exc, + ) + try: + db.rollback() + except Exception: + pass + except Exception as fetch_exc: + counters.address_failed += 1 + logger.warning( + "yandex-sweep run_id=%d: address-enrich " + "fetch failed listing_id=%d: %s", + run_id, + lid, + fetch_exc, + ) + if eidx < len(items) - 1: + await asyncio.sleep(enrich_delay) + + logger.info( + "yandex-sweep run_id=%d anchor %s: " "address enrich=%d/%d failed=%d", + run_id, + _a_name, + counters.address_enriched, + counters.address_attempted, + counters.address_failed, + ) + + try: + await asyncio.wait_for(_yandex_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC) + except TimeoutError: + logger.warning( + "yandex-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " + "timed out after %ds — skipping", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + ANCHOR_TIMEOUT_SEC, + ) + counters.errors_count += 1 + consecutive_failures += 1 + _anchor_timed_out = True + if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive failures (last: timeout): " + "IP likely blocked or proxy hung", + run_id, + idx, + len(_anchors), + name, + consecutive_failures, + ) + 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: anchor timeout {ANCHOR_TIMEOUT_SEC}s)", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + if idx < len(_anchors): + await asyncio.sleep(inter_anchor_delay) + continue except Exception as e: # YandexRealtyScraper не propagate'ит block/rate exceptions — ловим broad. logger.exception("yandex-sweep run_id=%d: anchor %s SERP failed", run_id, name) @@ -729,122 +927,6 @@ async def run_yandex_city_sweep( await asyncio.sleep(inter_anchor_delay) continue - # ── Phase 3: address-enrich ───────────────────────────────────── - if enrich_address and anchor_lots: - # Собираем source_id из SERP-результатов этого anchor'а → ищем - # в DB только листинги, которые были только что сохранены/обновлены. - source_ids = [lot.source_id for lot in anchor_lots if lot.source_id] - if source_ids: - id_rows = ( - db.execute( - _text(""" - SELECT id, source_id, address, source_url - FROM listings - WHERE source = 'yandex' - AND source_id = ANY(CAST(:ids AS text[])) - AND address IS NOT NULL - AND NOT (address ~ ',\\s*\\d+') - AND source_url IS NOT NULL - """), - {"ids": source_ids}, - ) - .mappings() - .all() - ) - items = [dict(r) for r in id_rows] - counters.address_attempted += len(items) - - if items: - from curl_cffi.requests import AsyncSession as _AsyncSession - - async with _AsyncSession( - impersonate="chrome120", - timeout=30.0, - proxies=_proxies, - headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"}, - ) as enrich_session: - for eidx, item in enumerate(items): - lid: int = item["id"] - old_addr: str = item["address"] - url: str = item["source_url"] - try: - resp = await enrich_session.get(url, allow_redirects=True) - if resp.status_code != 200: - logger.warning( - "yandex-sweep run_id=%d address-enrich:" - " HTTP %d listing_id=%d", - run_id, - resp.status_code, - lid, - ) - counters.address_failed += 1 - else: - new_addr = _extract_address_from_title(resp.text) - if ( - new_addr is None - or new_addr == old_addr - or not _RE_HAS_HOUSE_NUMBER.search(new_addr) - ): - # Не удалось улучшить адрес — пропускаем (не ошибка). - pass - else: - try: - db.execute( - _text(""" - UPDATE listings - SET address = :addr, - geocode_tried_at = NULL - WHERE id = CAST(:id AS bigint) - """), - {"addr": new_addr, "id": lid}, - ) - db.commit() - counters.address_enriched += 1 - logger.info( - "yandex-sweep run_id=%d: " - "address updated listing_id=%d " - "%r -> %r", - run_id, - lid, - old_addr, - new_addr, - ) - except Exception as upd_exc: - counters.address_failed += 1 - counters.errors_count += 1 - logger.warning( - "yandex-sweep run_id=%d: " - "address DB update failed " - "listing_id=%d: %s", - run_id, - lid, - upd_exc, - ) - try: - db.rollback() - except Exception: - pass - except Exception as fetch_exc: - counters.address_failed += 1 - logger.warning( - "yandex-sweep run_id=%d: address-enrich " - "fetch failed listing_id=%d: %s", - run_id, - lid, - fetch_exc, - ) - if eidx < len(items) - 1: - await asyncio.sleep(enrich_delay) - - logger.info( - "yandex-sweep run_id=%d anchor %s: " "address enrich=%d/%d failed=%d", - run_id, - name, - counters.address_enriched, - counters.address_attempted, - counters.address_failed, - ) - counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) @@ -1125,13 +1207,29 @@ async def run_cian_city_sweep( lon, ) - # ── Phase 1+2: SERP + save ────────────────────────────────────── + # ── Per-anchor phases под watchdog-таймаутом ──────────────────── + # Весь SERP + detail + houses одного anchor'а оборачиваем в wait_for. + # TimeoutError → log + errors_count++ + continue (sweep не прерывается). + anchor_lots: list[ScrapedLot] = [] - try: + # Capture loop variables in default args (B023): prevents stale binding + # if the coroutine is scheduled after the loop variable changes. + _c_lat, _c_lon, _c_name = lat, lon, name + + async def _cian_anchor_phases( + _a_lat: float = _c_lat, + _a_lon: float = _c_lon, + _a_name: str = _c_name, + ) -> None: + """Все фазы одного cian anchor'а (SERP + detail + houses).""" + nonlocal anchor_lots, consecutive_failures + from sqlalchemy import text as _text + + # ── Phase 1+2: SERP + save ───────────────────────────────── async with CianScraper() as scraper: anchor_lots = await scraper.fetch_around_multi_room( - lat, - lon, + _a_lat, + _a_lon, radius_m, pages=pages_per_anchor, ) @@ -1144,11 +1242,194 @@ async def run_cian_city_sweep( logger.info( "cian-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d", run_id, - name, + _a_name, len(anchor_lots), counters.lots_inserted, counters.lots_updated, ) + + # ── Phase 3: detail enrichment (top-N) ──────────────────── + if detail_top_n > 0: + priority_rows = ( + db.execute( + _text(""" + SELECT id, source_url + FROM listings + WHERE source = 'cian' + AND source_url IS NOT NULL + AND detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '2 hours' + ORDER BY scraped_at DESC NULLS LAST + LIMIT :lim + """), + {"lim": detail_top_n}, + ) + .mappings() + .all() + ) + for didx, row in enumerate(priority_rows): + listing_id: int = row["id"] + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + enrichment = await fetch_detail(source_url) + if enrichment is not None: + save_detail_enrichment(db, listing_id, enrichment) + counters.detail_enriched += 1 + else: + counters.detail_failed += 1 + logger.warning( + "cian-sweep run_id=%d: detail returned None for " + "listing_id=%d url=%s", + run_id, + listing_id, + source_url, + ) + except Exception as exc: + counters.detail_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-sweep run_id=%d: detail failed listing_id=%d: %s", + run_id, + listing_id, + exc, + ) + if didx < len(priority_rows) - 1: + jitter = random.uniform(0.8, 1.2) + await asyncio.sleep(request_delay_sec * jitter) + logger.info( + "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d", + run_id, + _a_name, + counters.detail_enriched, + counters.detail_attempted, + counters.detail_failed, + ) + + # ── Phase 4: newbuilding/houses enrichment ───────────────── + if not (enrich_houses and anchor_lots): + return + cian_nb_ext_ids = { + lot.house_ext_id + for lot in anchor_lots + if lot.house_source == "cian_newbuilding" and lot.house_ext_id + } + if not cian_nb_ext_ids: + return + nb_id_list = list(cian_nb_ext_ids) + try: + house_rows = ( + db.execute( + _text(""" + SELECT h.id, h.cian_zhk_url + FROM houses h + JOIN house_sources hs ON hs.house_id = h.id + WHERE hs.ext_source = 'cian_newbuilding' + AND hs.ext_id = ANY(CAST(:ids AS text[])) + AND h.cian_zhk_url IS NOT NULL + """), + {"ids": nb_id_list}, + ) + .mappings() + .all() + ) + except Exception as exc: + logger.exception( + "cian-sweep run_id=%d anchor %s: houses DB query failed — " + "skipping houses phase (%d ids): %s", + run_id, + _a_name, + len(nb_id_list), + exc, + ) + counters.houses_failed += len(nb_id_list) + try: + db.rollback() + except Exception: + pass + return + for hidx, hrow in enumerate(house_rows): + house_id_val: int = hrow["id"] + zhk_url: str = hrow["cian_zhk_url"] + counters.houses_attempted += 1 + try: + nb_enrichment = await fetch_newbuilding(zhk_url) + if nb_enrichment is not None: + save_newbuilding_enrichment(db, house_id_val, nb_enrichment) + counters.houses_enriched += 1 + else: + counters.houses_failed += 1 + logger.warning( + "cian-sweep run_id=%d: newbuilding returned None " + "house_id=%d url=%s", + run_id, + house_id_val, + zhk_url, + ) + except Exception as exc: + counters.houses_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-sweep run_id=%d: houses failed house_id=%d: %s", + run_id, + house_id_val, + exc, + ) + try: + db.rollback() + except Exception: + pass + if hidx < len(house_rows) - 1: + await asyncio.sleep(request_delay_sec) + logger.info( + "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d", + run_id, + _a_name, + counters.houses_enriched, + counters.houses_attempted, + counters.houses_failed, + ) + + try: + await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC) + except TimeoutError: + logger.warning( + "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " + "timed out after %ds — skipping", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + ANCHOR_TIMEOUT_SEC, + ) + counters.errors_count += 1 + consecutive_failures += 1 + if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive failures (last: timeout): " + "IP likely blocked or proxy hung", + run_id, + idx, + len(_anchors), + name, + consecutive_failures, + ) + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_banned( + db, + run_id, + f"cian sweep aborted: {consecutive_failures} consecutive " + f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + continue except Exception as e: logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name) counters.errors_count += 1 @@ -1178,153 +1459,6 @@ async def run_cian_city_sweep( scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) continue - # ── Phase 3: detail enrichment (top-N) ───────────────────────── - if detail_top_n > 0: - from sqlalchemy import text as _text - - priority_rows = ( - db.execute( - _text(""" - SELECT id, source_url - FROM listings - WHERE source = 'cian' - AND source_url IS NOT NULL - AND detail_enriched_at IS NULL - AND scraped_at > NOW() - INTERVAL '2 hours' - ORDER BY scraped_at DESC NULLS LAST - LIMIT :lim - """), - {"lim": detail_top_n}, - ) - .mappings() - .all() - ) - for didx, row in enumerate(priority_rows): - listing_id: int = row["id"] - source_url: str = row["source_url"] - counters.detail_attempted += 1 - try: - enrichment = await fetch_detail(source_url) - if enrichment is not None: - save_detail_enrichment(db, listing_id, enrichment) - counters.detail_enriched += 1 - else: - counters.detail_failed += 1 - logger.warning( - "cian-sweep run_id=%d: detail returned None for " - "listing_id=%d url=%s", - run_id, - listing_id, - source_url, - ) - except Exception as exc: - counters.detail_failed += 1 - counters.errors_count += 1 - logger.warning( - "cian-sweep run_id=%d: detail failed listing_id=%d: %s", - run_id, - listing_id, - exc, - ) - if didx < len(priority_rows) - 1: - jitter = random.uniform(0.8, 1.2) - await asyncio.sleep(request_delay_sec * jitter) - logger.info( - "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d", - run_id, - name, - counters.detail_enriched, - counters.detail_attempted, - counters.detail_failed, - ) - - # ── Phase 4: newbuilding/houses enrichment ───────────────────── - if enrich_houses and anchor_lots: - from sqlalchemy import text as _text2 - - # Собираем cian_internal_house_ids из SERP-объявлений этого anchor'а. - # lot.house_ext_id → SERP предоставляет house_ext_id (newbuilding_id). - # Резолвим house_id через house_sources (ext_source, ext_id) → houses.id. - cian_nb_ext_ids = { - lot.house_ext_id - for lot in anchor_lots - if lot.house_source == "cian_newbuilding" and lot.house_ext_id - } - if cian_nb_ext_ids: - nb_id_list = list(cian_nb_ext_ids) - try: - house_rows = ( - db.execute( - _text2(""" - SELECT h.id, h.cian_zhk_url - FROM houses h - JOIN house_sources hs ON hs.house_id = h.id - WHERE hs.ext_source = 'cian_newbuilding' - AND hs.ext_id = ANY(CAST(:ids AS text[])) - AND h.cian_zhk_url IS NOT NULL - """), - {"ids": nb_id_list}, - ) - .mappings() - .all() - ) - except Exception as exc: - logger.exception( - "cian-sweep run_id=%d anchor %s: houses DB query failed — " - "skipping houses phase (%d ids): %s", - run_id, - name, - len(nb_id_list), - exc, - ) - counters.houses_failed += len(nb_id_list) - try: - db.rollback() - except Exception: - pass - house_rows = [] - for hidx, hrow in enumerate(house_rows): - house_id_val: int = hrow["id"] - zhk_url: str = hrow["cian_zhk_url"] - counters.houses_attempted += 1 - try: - nb_enrichment = await fetch_newbuilding(zhk_url) - if nb_enrichment is not None: - save_newbuilding_enrichment(db, house_id_val, nb_enrichment) - counters.houses_enriched += 1 - else: - counters.houses_failed += 1 - logger.warning( - "cian-sweep run_id=%d: newbuilding returned None " - "house_id=%d url=%s", - run_id, - house_id_val, - zhk_url, - ) - except Exception as exc: - counters.houses_failed += 1 - counters.errors_count += 1 - logger.warning( - "cian-sweep run_id=%d: houses failed house_id=%d: %s", - run_id, - house_id_val, - exc, - ) - try: - db.rollback() - except Exception: - pass - if hidx < len(house_rows) - 1: - await asyncio.sleep(request_delay_sec) - logger.info( - "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d", - run_id, - name, - counters.houses_enriched, - counters.houses_attempted, - counters.houses_failed, - ) - counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) @@ -1358,6 +1492,7 @@ async def run_cian_city_sweep( # ── Public re-exports ─────────────────────────────────────────── __all__ = [ + "ANCHOR_TIMEOUT_SEC", "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES", "EKB_ANCHORS", "CianCitySweepCounters", diff --git a/tradein-mvp/backend/tests/test_anchor_watchdog.py b/tradein-mvp/backend/tests/test_anchor_watchdog.py new file mode 100644 index 00000000..8d0c8060 --- /dev/null +++ b/tradein-mvp/backend/tests/test_anchor_watchdog.py @@ -0,0 +1,422 @@ +"""Unit tests for per-anchor watchdog timeout in city sweeps (#880). + +Verifies that ANCHOR_TIMEOUT_SEC is applied to all three sweeps (avito / cian / yandex): + (a) A hung anchor (asyncio.sleep(large)) doesn't freeze the whole sweep. + (b) TimeoutError branch: errors_count incremented, anchor logged as timed-out. + (c) Sweep proceeds to the next anchor after the timeout. + (d) mark_done is called when the sweep finishes all anchors. + +All tests are network-free — they monkeypatch the inner scrape call. +""" + +from __future__ import annotations + +import asyncio +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.services import scrape_pipeline +from app.services.scrapers.base import ScrapedLot +from app.services.scrapers.cian import CianScraper +from app.services.scrapers.yandex_realty import YandexRealtyScraper + +# Two anchors: first hangs (causes timeout), second succeeds. +TWO_ANCHORS: list[tuple[float, float, str]] = [ + (56.8400, 60.6050, "Hang"), + (56.7950, 60.5300, "OK"), +] + + +# ── Shared fake scrape_runs ───────────────────────────────────────────────── + + +class _FakeRuns: + def __init__(self) -> None: + 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: + return False + + 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)) + + +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) + + +def _fake_lot(source: str, offer_id: str) -> ScrapedLot: + return ScrapedLot( + source=source, + source_url=f"https://example.ru/{offer_id}/", + source_id=offer_id, + address="Екатеринбург, улица Тестовая, 1", + price_rub=4_000_000, + area_m2=42.0, + rooms=1, + ) + + +# ── CianScraper lifecycle stub ────────────────────────────────────────────── + + +@pytest.fixture(autouse=True) +def _stub_cian_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + def _init(self: CianScraper) -> None: + self.request_delay_sec = 0.0 + self._cffi = None + + async def _aenter(self: CianScraper) -> CianScraper: + return self + + async def _aexit(self: CianScraper, *_args: Any) -> None: + return None + + monkeypatch.setattr(CianScraper, "__init__", _init) + monkeypatch.setattr(CianScraper, "__aenter__", _aenter) + monkeypatch.setattr(CianScraper, "__aexit__", _aexit) + + +# ── YandexRealtyScraper lifecycle stub ───────────────────────────────────── + + +@pytest.fixture(autouse=True) +def _stub_yandex_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + def _init(self: YandexRealtyScraper, city: str = "ekaterinburg") -> None: + self.city = city + self.request_delay_sec = 0.0 + self._cffi_session = None + self._cookies = {} + + 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) + + +# ── ANCHOR_TIMEOUT_SEC is exported ───────────────────────────────────────── + + +def test_anchor_timeout_sec_exported() -> None: + from app.services.scrape_pipeline import ANCHOR_TIMEOUT_SEC + + assert isinstance(ANCHOR_TIMEOUT_SEC, int) + # Reasonable sane range: 60s–600s + assert ( + 60 <= ANCHOR_TIMEOUT_SEC <= 600 + ), f"ANCHOR_TIMEOUT_SEC={ANCHOR_TIMEOUT_SEC} out of sane range" + + +# ── Avito sweep watchdog ──────────────────────────────────────────────────── + + +async def test_avito_sweep_anchor_timeout_continues( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Первый anchor виснет → таймаут; второй anchor успешен; mark_done вызывается.""" + call_count = 0 + + # Monkeypatch asyncio.wait_for so we can inject TimeoutError on first call only, + # without actually waiting 240s in CI. + original_wait_for = asyncio.wait_for + wf_call = 0 + + async def fake_wait_for(coro: Any, *, timeout: float) -> Any: + nonlocal wf_call + wf_call += 1 + if wf_call == 1: + # Simulate the hung anchor: cancel the coro (mimic wait_for behaviour) + # and raise TimeoutError. + try: + coro.close() + except Exception: + pass + raise TimeoutError + return await original_wait_for(coro, timeout=timeout) + + monkeypatch.setattr(scrape_pipeline.asyncio, "wait_for", fake_wait_for) + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock()) + + # Stub run_avito_pipeline so the second anchor returns 3 lots quickly. + async def fake_pipeline(db: Any, *, lat: float, **_kw: Any) -> Any: + nonlocal call_count + call_count += 1 + from app.services.scrape_pipeline import PipelineCounters, PipelineResult + + cnt = PipelineCounters(lots_fetched=3, lots_inserted=3) + return PipelineResult(lat, 0.0, 1500, cnt, False, 0) + + monkeypatch.setattr(scrape_pipeline, "run_avito_pipeline", fake_pipeline) + + # Stub shared AsyncSession (used by avito sweep). + fake_session = AsyncMock() + fake_session.__aenter__ = AsyncMock(return_value=fake_session) + fake_session.__aexit__ = AsyncMock(return_value=None) + import curl_cffi.requests as _cffi_mod + + monkeypatch.setattr(_cffi_mod, "AsyncSession", lambda **_kw: fake_session) + + # Stub IMV phase (not relevant to this test). + async def fake_imv_batch(*_a: Any, **_kw: Any) -> Any: + result = MagicMock() + result.checked = 0 + result.saved = 0 + result.errors = 0 + return result + + monkeypatch.setattr( + "app.services.house_imv_backfill.process_houses_imv_batch", + fake_imv_batch, + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_avito_city_sweep( + db=MagicMock(), + run_id=1, + anchors=TWO_ANCHORS, + enrich_houses=False, + detail_top_n=0, + enrich_imv=False, + request_delay_sec=0.0, + ) + + # (a) sweep completes — doesn't hang + # (b) first anchor timed out → errors_count = 1 + assert counters.errors_count >= 1, "timeout должен увеличить errors_count" + # (c) second anchor reached: lots_fetched > 0 + assert counters.lots_fetched > 0, "второй anchor должен был выполниться" + # (d) mark_done вызван (sweep завершился) + assert fake.done is not None, "mark_done должен быть вызван" + assert fake.failed is None, "mark_failed не должен быть вызван" + assert counters.anchors_done == len(TWO_ANCHORS) + + +# ── Cian sweep watchdog ───────────────────────────────────────────────────── + + +async def test_cian_sweep_anchor_timeout_continues( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Первый anchor виснет → таймаут; второй anchor успешен; mark_done вызывается.""" + serp_calls: list[str] = [] + original_wait_for = asyncio.wait_for + wf_call = 0 + + async def fake_wait_for(coro: Any, *, timeout: float) -> Any: + nonlocal wf_call + wf_call += 1 + if wf_call == 1: + try: + coro.close() + except Exception: + pass + raise TimeoutError + return await original_wait_for(coro, timeout=timeout) + + monkeypatch.setattr(scrape_pipeline.asyncio, "wait_for", fake_wait_for) + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock()) + + async def fake_fetch_multi( + self: CianScraper, lat: float, lon: float, r: int, pages: int = 3 + ) -> list[ScrapedLot]: + serp_calls.append(f"{lat:.4f}") + return [_fake_lot("cian", f"{lat:.4f}-1")] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr( + scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0) + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + AsyncMock(return_value=None), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=2, + anchors=TWO_ANCHORS, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + # (a) sweep completes (doesn't hang at anchor 1) + # (b) errors_count >= 1 (timeout counted) + assert counters.errors_count >= 1 + # (c) second anchor was reached (SERP called once — first anchor timed out before SERP) + # wf_call==2 means the second wait_for ran → second anchor's coro executed. + assert wf_call == 2, f"wait_for should be called once per anchor; got {wf_call}" + # (d) mark_done called + assert fake.done is not None + assert fake.failed is None + assert counters.anchors_done == len(TWO_ANCHORS) + + +# ── Yandex sweep watchdog ─────────────────────────────────────────────────── + + +async def test_yandex_sweep_anchor_timeout_continues( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Первый anchor виснет → таймаут; второй anchor успешен; mark_done вызывается.""" + serp_calls: list[str] = [] + original_wait_for = asyncio.wait_for + wf_call = 0 + + async def fake_wait_for(coro: Any, *, timeout: float) -> Any: + nonlocal wf_call + wf_call += 1 + if wf_call == 1: + try: + coro.close() + except Exception: + pass + raise TimeoutError + return await original_wait_for(coro, timeout=timeout) + + monkeypatch.setattr(scrape_pipeline.asyncio, "wait_for", fake_wait_for) + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock()) + + async def fake_fetch_multi( + self: YandexRealtyScraper, + lat: float, + lon: float, + radius_m: int = 1500, + max_pages: int = 2, + **_: Any, + ) -> list[ScrapedLot]: + serp_calls.append(f"{lat:.4f}") + return [_fake_lot("yandex", f"{lat:.4f}-1")] + + monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi) + 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=MagicMock(), + run_id=3, + anchors=TWO_ANCHORS, + pages_per_anchor=1, + enrich_address=False, + request_delay_sec=0.0, + ) + + # (a) sweep completes + # (b) timeout counted + assert counters.errors_count >= 1 + # (c) second anchor reached + assert wf_call == 2 + # (d) mark_done called + assert fake.done is not None + assert fake.failed is None + assert counters.anchors_done == len(TWO_ANCHORS) + + +# ── All three: timeout increments errors_count and is logged, sweep reaches mark_done ── + + +async def test_cian_timeout_increments_counter_and_marks_done( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Детальная проверка: timeout anchor → errors_count+1, sweep продолжается, mark_done. + + Использует fake_wait_for (зеркало первых трёх тестов): первый вызов → TimeoutError + (имитирует зависший anchor), второй вызов → реальный wait_for (нормальный anchor). + Проверяет что: + - errors_count увеличивается + - второй anchor выполнился (SERP вызван) + - mark_done вызван (sweep завершился) + - mark_banned НЕ вызван (одиночный таймаут < порога consecutive failures) + """ + original_wait_for = asyncio.wait_for + wf_call = 0 + + async def fake_wait_for(coro: Any, *, timeout: float) -> Any: + nonlocal wf_call + wf_call += 1 + if wf_call == 1: + try: + coro.close() + except Exception: + pass + raise TimeoutError + return await original_wait_for(coro, timeout=timeout) + + monkeypatch.setattr(scrape_pipeline.asyncio, "wait_for", fake_wait_for) + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", AsyncMock()) + + serp_calls: list[str] = [] + + async def fake_fetch( + self: CianScraper, lat: float, lon: float, r: int, pages: int = 3 + ) -> list[ScrapedLot]: + serp_calls.append(f"{lat:.4f}") + return [_fake_lot("cian", f"{lat:.4f}-ok")] + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch) + monkeypatch.setattr( + scrape_pipeline, + "save_listings", + lambda _db, lots, *, run_id=None: (len(lots), 0), + ) + monkeypatch.setattr( + "app.services.scrapers.cian_detail.fetch_detail", + AsyncMock(return_value=None), + ) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=10, + anchors=TWO_ANCHORS, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + ) + + # (b) timeout branch: errors_count incremented + assert counters.errors_count >= 1, "timeout должен увеличить errors_count" + # (c) second anchor reached (wait_for was called twice — once per anchor) + assert wf_call == 2, f"wait_for должен быть вызван для каждого anchor; got {wf_call}" + # (d) sweep reached mark_done + assert fake.done is not None, "mark_done должен быть вызван" + assert fake.banned is None, "mark_banned не должен быть вызван для одиночного таймаута" + assert counters.anchors_done == len(TWO_ANCHORS)