feat(tradein): per-anchor watchdog timeout in city sweeps (#880) (#881)

Co-authored-by: bot-backend <bot-backend@gendsgn.local>
Co-committed-by: bot-backend <bot-backend@gendsgn.local>

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(
+                    result = await asyncio.wait_for(
-                        db,
+                        run_avito_pipeline(
-                        lat=lat,
+                            db,
-                        lon=lon,
+                            lat=lat,
-                        radius_m=radius_m,
+                            lon=lon,
-                        enrich_houses=enrich_houses,
+                            radius_m=radius_m,
-                        enrich_detail_top_n=detail_top_n,
+                            enrich_houses=enrich_houses,
-                        pages=pages_per_anchor,
+                            enrich_detail_top_n=detail_top_n,
-                        request_delay_sec=request_delay_sec,
+                            pages=pages_per_anchor,
-                        shared_session=session,
+                            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,
+                        _a_lat,
-                        lon,
+                        _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,
+                        _a_lat,
-                        lon,
+                        _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",

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)