gendesign/tradein-mvp/backend/app/services/scrape_pipeline.py

"""scrape_pipeline.py — Avito full orchestrator (Stage 2e + city sweep).

Координирует 5 шагов одного pipeline run:
  1. SEARCH   — AvitoScraper().fetch_around(lat, lon, radius_m, pages=N)
  2. SAVE     — save_listings(db, lots) → (inserted, updated)
  3. GROUP    — dedupe house_url из lots → unique set
  4. ENRICH_HOUSES — fetch_house_catalog + save_house_catalog_enrichment per unique house
  5. ENRICH_DETAIL — top-N listings (detail_enriched_at IS NULL) → fetch_detail + save

City sweep: run_avito_city_sweep — итерирует EKB_ANCHORS × pages_per_anchor,
с cooperative cancel через scrape_runs.status и heartbeat update каждый anchor.

Anti-bot hardening (2023-05-23):
  - Shared AsyncSession на весь sweep (один TLS handshake)
  - asyncio.sleep с random +-20% jitter между detail requests
  - AvitoBlockedError/AvitoRateLimitedError propagation → early abort
  - После 3 consecutive blocks — abort sweep + mark_banned (status='banned')

Graceful degradation на каждом step: exception в одной house/listing
не валит весь pipeline (кроме blocked exceptions — они abort весь sweep).

IMV-фаза (финальная, после всех anchor'ов): если enrich_imv=True —
для домов, тронутых в этом sweep'е (touched house_id из houses-enrichment),
вызывается house_imv_backfill.process_houses_imv_batch. Ошибки per-house
логируются и считаются, но не валят sweep.
"""

from __future__ import annotations

import asyncio
import logging
import random
from contextlib import AsyncExitStack
from dataclasses import dataclass, field, fields
from datetime import date, timedelta
from typing import Any
from urllib.parse import urlparse

from curl_cffi.requests import AsyncSession
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.shutdown import shutdown_requested
from app.services import scrape_runs
from app.services.scrapers.avito import AvitoScraper
from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
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.browser_fetcher import BrowserFetcher
from app.services.scrapers.yandex_realty import YandexRealtyScraper
from app.services.yandex_price_history import record_yandex_price_history

logger = logging.getLogger(__name__)


async def _rotate_proxy_ip(*, reason: str, rotations_done: int, source: str = "avito") -> bool:
    """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790/#1848).

    Provider-aware: выбирает rotate_url и max_rotations по параметру `source`
    (avito / cian / yandex). Fallback-цепочка для rotate_url:
      avito: avito_proxy_rotate_url
      cian:  cian_proxy_rotate_url → avito_proxy_rotate_url
      yandex: yandex_proxy_rotate_url → avito_proxy_rotate_url

    Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в
    enrichment-фазах где нет доступа к экземпляру scraper'а.

    Returns True при успешной смене IP, False если rotate_url не задан или ошибка.
    Логирует каждую ротацию (причина, порядковый номер, source, остаток лимита).
    """
    if source == "cian":
        rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url
        max_rot = settings.cian_proxy_max_rotations
    elif source == "yandex":
        rotate_url = settings.yandex_proxy_rotate_url or settings.avito_proxy_rotate_url
        max_rot = settings.yandex_proxy_max_rotations
    else:
        # avito (default) — backward compat
        rotate_url = settings.avito_proxy_rotate_url
        max_rot = settings.avito_proxy_max_rotations
    if not rotate_url:
        return False
    sep = "&" if "?" in rotate_url else "?"
    # #1950: retry changeip-GET короткими попытками вместо одношотного 30s-timeout.
    # Если changeip-сервер завис на 30s → весь прогон ждёт зря + ротация считается
    # успешной (False возвращался). Теперь: proxy_rotate_attempts попыток по
    # proxy_rotate_attempt_timeout_s каждая; возвращаем True при первом успехе.
    _attempts = settings.proxy_rotate_attempts
    _attempt_timeout = settings.proxy_rotate_attempt_timeout_s
    for attempt in range(_attempts):
        try:
            async with AsyncSession(timeout=_attempt_timeout) as rot:
                resp = await rot.get(f"{rotate_url}{sep}format=json")
            new_ip: str = ""
            try:
                data = resp.json()
                new_ip = str(data.get("new_ip", ""))
            except Exception:
                pass
            await asyncio.sleep(settings.avito_proxy_rotate_settle_s)
            logger.info(
                "pipeline: IP rotated via changeip — source=%s reason=%s "
                "rotation=#%d attempt=%d/%d remaining=%d new_ip=%s",
                source,
                reason,
                rotations_done + 1,
                attempt + 1,
                _attempts,
                max_rot - rotations_done - 1,
                new_ip or "unknown",
            )
            return True
        except Exception:
            logger.warning(
                "pipeline: IP rotation attempt %d/%d failed " "(source=%s reason=%s rotation=#%d)",
                attempt + 1,
                _attempts,
                source,
                reason,
                rotations_done + 1,
                exc_info=True,
            )
            if attempt < _attempts - 1:
                await asyncio.sleep(1.0)
    logger.error(
        "pipeline: all %d changeip attempts failed " "(source=%s reason=%s rotation=#%d)",
        _attempts,
        source,
        reason,
        rotations_done + 1,
    )
    return False


# Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает
# дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for
# прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом.
# Диапазон 180-300 с; 240 — разумный середина (4 мин > любой нормальный anchor).
ANCHOR_TIMEOUT_SEC: int = 240

# Константы для расчёта watchdog-таймаута combos-режима Yandex sweep.
# В combos-режиме единственный "anchor" делает num_combos × max_pages fetch'ей;
# каждый fetch занимает request_delay_sec + сетевой overhead. ADDRESS_ENRICH_BUDGET_S
# добавляет буфер на address-enrich фазу (per-listing HTTP + sleep × N листингов).
# Пример: 30 combos × 3 pages × (9+12) + 300 ≈ 2190s (~37 мин) — надёжно укладывается
# в окно 02:00-05:00 без риска перекрытия следующего sweep'а.
_YANDEX_COMBOS_PER_FETCH_S: float = 12.0  # network + parse margin (секунды на страницу)
_YANDEX_ADDRESS_ENRICH_BUDGET_S: float = 300.0  # бюджет на address-enrich фазу

# Константы для расчёта watchdog-таймаута Avito city sweep.
# Avito detail-fetch идёт через браузерный сервис с page.goto timeout 60s, и
# anti-bot страницы нередко доходят до этого лимита. Фиксированный ANCHOR_TIMEOUT_SEC=240
# guillotine'ит anchor ещё в detail-фазе → SERP-счётчики теряются, run показывает
# lots_fetched=0 при реально сохранённых листингах. Таймаут масштабируется:
#   _avito_anchor_timeout = ANCHOR_TIMEOUT_SEC
#                           + detail_top_n × _AVITO_PER_DETAIL_S
#                           + (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0)
# Пример: detail_top_n=20 + houses → 240 + 1000 + 180 = 1420s/anchor (worst-case).
# 5 anchor'ов × 1420s ≈ 2ч worst-case (нормальный прогон быстрее: не каждый detail
# достигает 60s timeout). Окно лишь ограничивает старт следующего sweep'а — scheduler
# ждёт финиша задачи и не запустит параллельный run.
_AVITO_PER_DETAIL_S: float = 50.0  # секунд на один detail-fetch (incl. occasional 60s timeout)
_AVITO_HOUSES_BUDGET_S: float = 180.0  # бюджет на house-enrich фазу
# Fail-fast: если detail phase накапливает подряд N timeout/ошибок, прерываем её
# (аналог consecutive_blocks для AvitoBlockedError в step 5). Срабатывает только при
# явной деградации browser-сервиса, не мешает нормальным transient-ошибкам.
_AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5

# #1820: порог ПОСЛЕДОВАТЕЛЬНЫХ блоков (AvitoBlockedError/AvitoRateLimitedError) внутри
# enrichment-фаз run_avito_pipeline. Один заблокированный фетч (house/detail) больше НЕ
# роняет весь прогон — блок ловится пер-item (как обычная ошибка), и лишь N подряд
# прерывают enrichment-фазу. Listings из SEARCH+SAVE сохраняются всегда. Зеркалит
# паттерн avito_detail_backfill (max_consecutive_blocks).
_AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3

# Default anchors ЕКБ — 5 точек покрытия города
EKB_ANCHORS: list[tuple[float, float, str]] = [
    (56.8400, 60.6050, "Центр"),
    (56.7950, 60.5300, "ЮЗ"),
    (56.8970, 60.6100, "Уралмаш"),
    (56.7700, 60.5500, "Академический"),
    (56.8650, 60.6200, "Пионерский"),
]

_CHROME_HEADERS = {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
    "Cache-Control": "max-age=0",
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
    "Upgrade-Insecure-Requests": "1",
}


def _avito_proxies() -> dict[str, str] | None:
    """#623/#806: mobile-proxy egress для shared-сессии sweep'а.

    КРИТИЧНО: sweep создаёт собственную shared AsyncSession и присваивает её
    `scraper._cffi`, перезатирая проксированную сессию из `AvitoScraper.__aenter__`.
    Через эту же сессию идёт и SERP, и detail-энричмент (fetch_detail). Без прокси
    detail-страницы летят с datacenter-IP → HTTP 429 → весь run mark_banned.
    Пусто (env не задан) → прямое подключение (dev/no-op).

    Читает settings.scraper_proxy_url (SCRAPER_PROXY_URL > AVITO_PROXY_URL fallback).
    """
    url = settings.scraper_proxy_url
    return {"http": url, "https": url} if url else None


# ── Result dataclasses ──────────────────────────────────────────


@dataclass
class PipelineCounters:
    """Counters одного pipeline run для логов/админки."""

    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: int = 0
    unique_houses: int = 0
    houses_enriched: int = 0
    houses_failed: int = 0
    detail_attempted: int = 0
    detail_enriched: int = 0
    detail_failed: int = 0
    errors: list[str] = field(default_factory=list)


@dataclass
class PipelineResult:
    """Полный результат full Avito pipeline."""

    anchor_lat: float
    anchor_lon: float
    radius_m: int
    counters: PipelineCounters
    enrich_houses: bool
    enrich_detail_top_n: int
    touched_house_ids: set[int] = field(default_factory=set)


# ── Main orchestrator ───────────────────────────────────────────


async def run_avito_pipeline(
    db: Session,
    lat: float,
    lon: float,
    radius_m: int = 1500,
    *,
    enrich_houses: bool = True,
    enrich_detail_top_n: int = 10,
    pages: int = 1,
    request_delay_sec: float | None = None,
    shared_session: AsyncSession | None = None,
    shared_browser: BrowserFetcher | None = None,
) -> PipelineResult:
    """Full Avito search → houses → detail enrichment pipeline.

    Steps:
      1. SEARCH: AvitoScraper().fetch_around(lat, lon, radius_m, pages=pages)
      2. SAVE: save_listings(db, lots) → (inserted, updated)
      3. GROUP: dedupe house_url из новых lots → unique house path set
      4. ENRICH_HOUSES: fetch_house_catalog + save per house (with sleep between)
      5. ENRICH_DETAIL: top-N listings → fetch_detail + save (abort on 3 blocks)

    shared_session — если передан, используется без закрытия (lifecycle у вызывающего).
    shared_browser — если передан в browser-mode, используется без закрытия.

    Блокировки (AvitoBlockedError / AvitoRateLimitedError) в SEARCH-фазе (шаг 1)
    по-прежнему propagate наружу (без листингов прогон бессмыслен). В enrichment-фазах
    (houses шаг 4 + detail шаг 5 + house-detail шаг 5b) блок ловится ПЕР-ITEM как обычная
    ошибка (#1820): один заблокированный фетч НЕ роняет весь прогон. Лишь
    _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT блоков ПОДРЯД прерывают enrichment — но
    PipelineResult всё равно возвращается с уже собранными listings/обогащениями.
    """
    counters = PipelineCounters()
    lots: list[ScrapedLot] = []
    detail_delay = request_delay_sec if request_delay_sec is not None else 7.0
    # #1820: общий счётчик ПОСЛЕДОВАТЕЛЬНЫХ блоков на все enrichment-фазы (4/5/5b).
    # Сбрасывается на любом успешном fetch; при достижении порога — ротация IP (#1790)
    # или abort если бюджет ротаций исчерпан.
    enrich_consecutive_blocks = 0
    enrich_rotations_done = 0  # #1790: сколько раз уже сменили IP в этом run
    enrichment_aborted = False

    browser_mode = settings.scraper_fetch_mode == "browser"
    session: AsyncSession | None = None
    browser_fetcher: BrowserFetcher | None = None
    own_session = False
    own_browser = False

    scraper = AvitoScraper()
    if browser_mode:
        browser_fetcher = shared_browser
        if browser_fetcher is None:
            browser_fetcher = BrowserFetcher(source="avito")
            await browser_fetcher.__aenter__()
            own_browser = True
        scraper._browser = browser_fetcher
    else:
        own_session = shared_session is None
        session = shared_session or AsyncSession(
            impersonate="chrome120",
            timeout=25,
            headers=_CHROME_HEADERS,
            proxies=_avito_proxies(),
        )
        scraper._cffi = session

    try:
        # ── Step 1: search ──────────────────────────────────────
        try:
            lots = await scraper.fetch_around(
                lat,
                lon,
                radius_m,
                pages=pages,
                delay_override_sec=request_delay_sec,
            )
            counters.lots_fetched = len(lots)
            logger.info(
                "pipeline:search anchor=(%.4f,%.4f) r=%dm fetched=%d",
                lat,
                lon,
                radius_m,
                len(lots),
            )
        except (AvitoBlockedError, AvitoRateLimitedError):
            logger.error("pipeline:search BLOCKED by Avito at anchor=(%.4f,%.4f)", lat, lon)
            raise
        except Exception as e:
            logger.exception("pipeline:search failed")
            counters.errors.append(f"search: {e}")
            return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)

        # ── Step 2: save listings ───────────────────────────────
        if lots:
            try:
                counters.lots_inserted, counters.lots_updated = save_listings(db, lots)
            except Exception as e:
                logger.exception("pipeline:save_listings failed")
                counters.errors.append(f"save_listings: {e}")

        # ── Step 3: group by house ──────────────────────────────
        unique_house_paths: set[str] = set()
        for lot in lots:
            if lot.house_url:
                try:
                    parsed = urlparse(lot.house_url)
                    path = parsed.path if parsed.path else lot.house_url
                    unique_house_paths.add(path)
                except Exception:
                    continue

        counters.unique_houses = len(unique_house_paths)
        logger.info("pipeline:group_by_house unique=%d", counters.unique_houses)

        # ── Step 4: enrich houses с shared session + sleep ──────
        touched_house_ids: set[int] = set()
        if enrich_houses and unique_house_paths and not enrichment_aborted:
            house_paths_list = list(unique_house_paths)
            for idx, house_path in enumerate(house_paths_list):
                try:
                    enrichment = await fetch_house_catalog(
                        house_path, cffi_session=session, browser_fetcher=browser_fetcher
                    )
                    house_counters = save_house_catalog_enrichment(db, enrichment)
                    counters.houses_enriched += 1
                    enrich_consecutive_blocks = 0
                    hid = house_counters.get("house_id")
                    if hid:
                        touched_house_ids.add(int(hid))
                except (AvitoBlockedError, AvitoRateLimitedError) as e:
                    # #1820: блок ловим пер-item как обычную ошибку — НЕ роняем прогон.
                    enrich_consecutive_blocks += 1
                    counters.houses_failed += 1
                    counters.errors.append(f"BLOCKED_HOUSE {house_path}: {e}")
                    logger.warning(
                        "pipeline:house BLOCKED at %s (consecutive=%d)",
                        house_path,
                        enrich_consecutive_blocks,
                    )
                    if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
                        # #1790: Перед abort'ом пробуем сменить IP (changeip).
                        # После ротации сбрасываем счётчик и продолжаем — НЕ abort.
                        # Если ротации исчерпаны — abort enrichment (listings уже сохранены).
                        if enrich_rotations_done < settings.avito_proxy_max_rotations:
                            rotated = await _rotate_proxy_ip(
                                reason=f"house consecutive={enrich_consecutive_blocks}",
                                rotations_done=enrich_rotations_done,
                            )
                            enrich_rotations_done += 1
                            if rotated:
                                enrich_consecutive_blocks = 0
                                logger.info(
                                    "pipeline:enrich ROTATE — IP changed, "
                                    "reset consecutive blocks (rotation #%d/%d)",
                                    enrich_rotations_done,
                                    settings.avito_proxy_max_rotations,
                                )
                                continue
                        logger.error(
                            "pipeline:enrich ABORT — %d consecutive blocks (IP rate-limited), "
                            "listings preserved",
                            enrich_consecutive_blocks,
                        )
                        counters.errors.append(
                            f"AVITO_BLOCKED — enrichment aborted after "
                            f"{enrich_consecutive_blocks} consecutive blocks (house phase)"
                        )
                        enrichment_aborted = True
                        break
                except Exception as e:
                    logger.warning("pipeline:house_enrich failed for %s: %s", house_path, e)
                    counters.houses_failed += 1
                    counters.errors.append(f"house {house_path}: {e}")
                    # #1368: save_house_catalog_enrichment не делает внутренний rollback,
                    # поэтому упавший промежуточный statement оставляет общий Session в
                    # InFailedSqlTransaction. Без сброса — каскад «current transaction is
                    # aborted» на все оставшиеся дома/listings этого якоря.
                    try:
                        db.rollback()
                    except Exception:
                        pass

                if idx < len(house_paths_list) - 1:
                    await asyncio.sleep(detail_delay)

        # ── Step 5: enrich detail для top-N priority listings ───
        if enrich_detail_top_n > 0 and not enrichment_aborted:
            priority_rows = (
                db.execute(
                    text("""
                    SELECT source_url
                    FROM listings
                    WHERE source = 'avito'
                      AND source_url IS NOT NULL
                      AND (
                        (
                          detail_enriched_at IS NULL
                          AND price_rub > 0
                          AND ST_DWithin(
                            geom::geography,
                            ST_MakePoint(:lon, :lat)::geography,
                            :radius
                          )
                        )
                        OR (
                          detail_enriched_at IS NULL
                          AND scraped_at > NOW() - INTERVAL '2 hours'
                        )
                      )
                    ORDER BY scraped_at DESC NULLS LAST
                    LIMIT :limit
                """),
                    {
                        "lat": lat,
                        "lon": lon,
                        "radius": radius_m * 2,
                        "limit": enrich_detail_top_n,
                    },
                )
                .mappings()
                .all()
            )

            detail_house_paths: set[str] = set()
            for idx, row in enumerate(priority_rows):
                source_url: str = row["source_url"]
                counters.detail_attempted += 1
                try:
                    item_url = (
                        urlparse(source_url).path if source_url.startswith("http") else source_url
                    )
                    enrichment_detail = await fetch_detail(
                        item_url, cffi_session=session, browser_fetcher=browser_fetcher
                    )
                    if save_detail_enrichment(db, enrichment_detail):
                        counters.detail_enriched += 1
                    # #871: house-link есть в detail SSR (в SERP-карточке — JS-only).
                    # Копим catalog-пути для house-enrich ниже (Step 5b).
                    if enrichment_detail.house_catalog_url:
                        hp = urlparse(enrichment_detail.house_catalog_url).path
                        if hp:
                            detail_house_paths.add(hp)
                    enrich_consecutive_blocks = 0
                except (AvitoBlockedError, AvitoRateLimitedError) as e:
                    # #1820: блок ловим пер-item, НЕ роняем прогон. Счётчик общий с house-фазой.
                    enrich_consecutive_blocks += 1
                    counters.detail_failed += 1
                    counters.errors.append(f"BLOCKED_DETAIL #{idx}: {e}")
                    logger.warning(
                        "pipeline:detail BLOCKED #%d/%d (consecutive=%d): %s",
                        idx + 1,
                        len(priority_rows),
                        enrich_consecutive_blocks,
                        e,
                    )
                    if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
                        # #1790: Перед abort'ом пробуем сменить IP (changeip).
                        # После ротации сбрасываем счётчик — продолжаем detail-обход.
                        # Если ротации исчерпаны — abort detail-фазы (listings уже сохранены).
                        if enrich_rotations_done < settings.avito_proxy_max_rotations:
                            rotated = await _rotate_proxy_ip(
                                reason=f"detail consecutive={enrich_consecutive_blocks}",
                                rotations_done=enrich_rotations_done,
                            )
                            enrich_rotations_done += 1
                            if rotated:
                                enrich_consecutive_blocks = 0
                                logger.info(
                                    "pipeline:enrich ROTATE — IP changed, "
                                    "reset consecutive blocks (rotation #%d/%d)",
                                    enrich_rotations_done,
                                    settings.avito_proxy_max_rotations,
                                )
                                continue
                        logger.error(
                            "pipeline:detail ABORT — %d consecutive blocks (IP rate-limited), "
                            "listings preserved",
                            enrich_consecutive_blocks,
                        )
                        counters.errors.append(
                            f"AVITO_BLOCKED — enrichment aborted after "
                            f"{idx + 1}/{len(priority_rows)} details (consecutive blocks)"
                        )
                        # #1820: НЕ raise — прерываем только detail-фазу (и Step 5b),
                        # PipelineResult возвращается с уже сохранёнными listings/houses.
                        enrichment_aborted = True
                        break
                except Exception as e:
                    counters.detail_failed += 1
                    counters.errors.append(f"detail {source_url}: {e}")
                    logger.warning("pipeline:detail_enrich failed for %s: %s", source_url, e)
                    # #1386: НЕ сбрасываем consecutive_blocks здесь. Транзиентная
                    # сетевая/parse-ошибка (5xx, timeout, connection reset) — правдоподобный
                    # побочный эффект активного IP-блока (Avito не отдаёт 403/429 единообразно).
                    # Сброс маскировал бы продолжающийся блок (block, block, timeout→reset, ...)
                    # и не давал бы сработать early-abort на 3 consecutive blocks.
                    # Сброс на успехе (см. выше) остаётся.
                    # #1368: save_detail_enrichment без внутреннего rollback мог оставить
                    # общий Session в aborted-state — сбрасываем, иначе каскад на оставшиеся
                    # detail-строки и Step 5b.
                    try:
                        db.rollback()
                    except Exception:
                        pass

                if idx < len(priority_rows) - 1:
                    jitter = random.uniform(0.8, 1.2)
                    await asyncio.sleep(detail_delay * jitter)

            # ── Step 5b: enrich houses, найденные через detail-страницы (#871) ──
            # SERP-карточки больше не несут house-link (JS-render) → дома берём из
            # detail SSR. Дедуп против обработанных в Step 4 (unique_house_paths).
            new_house_paths = detail_house_paths - unique_house_paths
            if enrich_houses and new_house_paths and not enrichment_aborted:
                counters.unique_houses += len(new_house_paths)
                nh_list = list(new_house_paths)
                for h_idx, house_path in enumerate(nh_list):
                    try:
                        enrichment = await fetch_house_catalog(
                            house_path, cffi_session=session, browser_fetcher=browser_fetcher
                        )
                        house_counters = save_house_catalog_enrichment(db, enrichment)
                        counters.houses_enriched += 1
                        enrich_consecutive_blocks = 0
                        hid = house_counters.get("house_id")
                        if hid:
                            touched_house_ids.add(int(hid))
                    except (AvitoBlockedError, AvitoRateLimitedError) as e:
                        # #1820: блок пер-item, общий счётчик; не роняем прогон.
                        enrich_consecutive_blocks += 1
                        counters.houses_failed += 1
                        counters.errors.append(f"BLOCKED_HOUSE(detail) {house_path}: {e}")
                        logger.warning(
                            "pipeline:house(detail) BLOCKED at %s (consecutive=%d)",
                            house_path,
                            enrich_consecutive_blocks,
                        )
                        if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT:
                            # #1790: попытка ротации перед abort'ом (общий бюджет run).
                            if enrich_rotations_done < settings.avito_proxy_max_rotations:
                                rotated = await _rotate_proxy_ip(
                                    reason=f"house-detail consecutive={enrich_consecutive_blocks}",
                                    rotations_done=enrich_rotations_done,
                                )
                                enrich_rotations_done += 1
                                if rotated:
                                    enrich_consecutive_blocks = 0
                                    logger.info(
                                        "pipeline:enrich ROTATE — IP changed, "
                                        "reset consecutive blocks (rotation #%d/%d)",
                                        enrich_rotations_done,
                                        settings.avito_proxy_max_rotations,
                                    )
                                    continue
                            logger.error(
                                "pipeline:enrich ABORT — %d consecutive blocks "
                                "(IP rate-limited), listings preserved",
                                enrich_consecutive_blocks,
                            )
                            counters.errors.append(
                                f"AVITO_BLOCKED — enrichment aborted after "
                                f"{enrich_consecutive_blocks} consecutive blocks "
                                f"(house-detail phase)"
                            )
                            enrichment_aborted = True
                            break
                    except Exception as e:
                        logger.warning(
                            "pipeline:house(detail) enrich failed for %s: %s", house_path, e
                        )
                        counters.houses_failed += 1
                        counters.errors.append(f"house(detail) {house_path}: {e}")
                        # #1368: см. Step 4 — сбрасываем aborted-state общего Session,
                        # иначе каскад на оставшиеся дома detail-фазы.
                        try:
                            db.rollback()
                        except Exception:
                            pass
                    if h_idx < len(nh_list) - 1:
                        await asyncio.sleep(detail_delay)

        logger.info(
            "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) "
            "houses=%d/%d detail=%d/%d touched_house_ids=%d errors=%d",
            lat,
            lon,
            counters.lots_fetched,
            counters.lots_inserted,
            counters.lots_updated,
            counters.houses_enriched,
            counters.unique_houses,
            counters.detail_enriched,
            counters.detail_attempted,
            len(touched_house_ids),
            len(counters.errors),
        )
        return PipelineResult(
            lat,
            lon,
            radius_m,
            counters,
            enrich_houses,
            enrich_detail_top_n,
            touched_house_ids=touched_house_ids,
        )

    finally:
        if own_session and session is not None:
            await session.close()
        if own_browser and browser_fetcher is not None:
            await browser_fetcher.__aexit__(None, None, None)


@dataclass
class CitySweepCounters:
    """Aggregate counters для full city sweep run."""

    anchors_total: int = 0
    anchors_done: int = 0
    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: int = 0
    unique_houses: int = 0
    houses_enriched: int = 0
    houses_failed: int = 0
    detail_attempted: int = 0
    detail_enriched: int = 0
    detail_failed: int = 0
    imv_attempted: int = 0
    imv_enriched: int = 0
    imv_failed: 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)}


async def run_avito_city_sweep(
    db: Session,
    *,
    run_id: int,
    anchors: list[tuple[float, float, str]] | None = None,
    radius_m: int = 1500,
    pages_per_anchor: int = 3,
    enrich_houses: bool = True,
    detail_top_n: int = 10,
    request_delay_sec: float = 7.0,
    enrich_imv: bool = True,
) -> CitySweepCounters:
    """Full city sweep: iterate anchors × pages → save → enrich houses + detail → IMV.

    - Single shared AsyncSession на весь sweep (один TLS fingerprint)
    - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
    - Прочие errors per-anchor логируются, не валят весь sweep
    - После всех anchor'ов: если enrich_imv=True — IMV-оценка тронутых домов
      (cooperative cancel + per-house graceful error handling)

    SERP-счётчики (lots_fetched/inserted/updated) записываются в sweep-level counters
    НЕМЕДЛЕННО после save_listings — до начала detail/houses фазы. Даже если anchor
    превышает watchdog-таймаут в detail-фазе, SERP-результаты не теряются.
    """
    from app.services.house_imv_backfill import process_houses_imv_batch

    _anchors = anchors if anchors is not None else EKB_ANCHORS
    counters = CitySweepCounters(anchors_total=len(_anchors))
    all_touched_house_ids: set[int] = set()

    # Масштабируемый watchdog-таймаут для Avito anchor'а. В отличие от Yandex/Cian,
    # Avito detail-fetch идёт через browser service с page.goto timeout 60s. При
    # detail_top_n=20 detail-фаза может занять до N×60s — фиксированный ANCHOR_TIMEOUT_SEC
    # обрезает её раньше SERP, теряя счётчики. Масштабирование по detail_top_n + houses
    # гарантирует, что SERP+save всегда успевает до watchdog.
    _avito_anchor_timeout = (
        float(ANCHOR_TIMEOUT_SEC)
        + detail_top_n * _AVITO_PER_DETAIL_S
        + (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0)
    )
    logger.info(
        "city-sweep run_id=%d: avito anchor_timeout=%.0fs "
        "(base=%d + detail_top_n=%d×%.0fs + houses=%.0fs)",
        run_id,
        _avito_anchor_timeout,
        ANCHOR_TIMEOUT_SEC,
        detail_top_n,
        _AVITO_PER_DETAIL_S,
        _AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0,
    )

    browser_mode = settings.scraper_fetch_mode == "browser"
    # #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам).
    sweep_rotations_done = 0

    async with AsyncExitStack() as stack:
        session: AsyncSession | None = None
        shared_bf: BrowserFetcher | None = None
        if browser_mode:
            shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito"))
        else:
            session = await stack.enter_async_context(
                AsyncSession(
                    impersonate="chrome120",
                    timeout=25,
                    headers=_CHROME_HEADERS,
                    proxies=_avito_proxies(),
                )
            )
        try:
            for idx, (lat, lon, name) in enumerate(_anchors, start=1):
                if scrape_runs.is_cancelled(db, run_id):
                    logger.info(
                        "city-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
                elif shutdown_requested():
                    # Кооперативный SIGTERM-drain (#1182 Phase 3a): останавливаемся
                    # на границе anchor'а и финализируем run как mark_done(partial) —
                    # это НЕ user-cancel, поэтому НЕ mark_cancelled. Дальнейшие anchor'ы
                    # и IMV-фаза не выполняются.
                    logger.info(
                        "city-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)",
                        run_id,
                        idx,
                        len(_anchors),
                        name,
                    )
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                    scrape_runs.mark_done(db, run_id, counters.to_dict())
                    return counters

                logger.info(
                    "city-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
                    run_id,
                    idx,
                    len(_anchors),
                    name,
                    lat,
                    lon,
                )

                # Capture loop variables in default args (B023): prevents stale binding
                # if the coroutine is scheduled after the loop variable changes.
                _a_lat, _a_lon, _a_name = lat, lon, name

                async def _avito_anchor_phases(
                    _lat: float = _a_lat,
                    _lon: float = _a_lon,
                    _name: str = _a_name,
                ) -> None:
                    """Все фазы одного Avito anchor'а (SERP+save → houses → detail).

                    Обновляет sweep-level counters и all_touched_house_ids через nonlocal
                    НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что
                    TimeoutError из wait_for теряет только detail-счётчики, но не SERP.
                    """
                    nonlocal all_touched_house_ids, sweep_rotations_done

                    # ── Phase 1+2: SERP + save ─────────────────────────────────
                    scraper = AvitoScraper()
                    if browser_mode:
                        scraper._browser = shared_bf
                        # Shared-browser режим: _cffi=None → curl_cffi-fallback на
                        # firewall пропускается (by design). Логируем для observability.
                        logger.debug("avito pipeline-mode: shared browser only, no cffi fallback")
                    else:
                        scraper._cffi = session

                    try:
                        anchor_lots: list[ScrapedLot] = await scraper.fetch_around(
                            _lat,
                            _lon,
                            radius_m,
                            pages=pages_per_anchor,
                            delay_override_sec=request_delay_sec,
                        )
                    except (AvitoBlockedError, AvitoRateLimitedError):
                        logger.error(
                            "city-sweep run_id=%d: SERP BLOCKED at anchor (%s, %.4f, %.4f)",
                            run_id,
                            _name,
                            _lat,
                            _lon,
                        )
                        raise

                    counters.lots_fetched += len(anchor_lots)
                    if anchor_lots:
                        try:
                            ins, upd = save_listings(db, anchor_lots)
                            counters.lots_inserted += ins
                            counters.lots_updated += upd
                        except Exception as save_exc:
                            logger.exception(
                                "city-sweep run_id=%d: save_listings failed at anchor %s: %s",
                                run_id,
                                _name,
                                save_exc,
                            )
                            counters.errors_count += 1

                    logger.info(
                        "city-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d",
                        run_id,
                        _name,
                        len(anchor_lots),
                        counters.lots_inserted,
                        counters.lots_updated,
                    )
                    # SERP-счётчики зафиксированы в sweep-level counters.
                    # Дальнейший TimeoutError из wait_for их не сотрёт.

                    # ── Phase 3: group by house ────────────────────────────────
                    unique_house_paths: set[str] = set()
                    for lot in anchor_lots:
                        if lot.house_url:
                            try:
                                parsed = urlparse(lot.house_url)
                                path = parsed.path if parsed.path else lot.house_url
                                unique_house_paths.add(path)
                            except Exception:
                                continue
                    if unique_house_paths:
                        counters.unique_houses += len(unique_house_paths)

                    # ── Phase 4: enrich houses ─────────────────────────────────
                    touched_house_ids: set[int] = set()
                    sweep_consecutive_house_blocks = 0  # #1790: per-anchor счётчик
                    if enrich_houses and unique_house_paths:
                        house_paths_list = list(unique_house_paths)
                        h_idx = 0
                        while h_idx < len(house_paths_list):
                            house_path = house_paths_list[h_idx]
                            try:
                                enrichment = await fetch_house_catalog(
                                    house_path,
                                    cffi_session=session,
                                    browser_fetcher=shared_bf,
                                )
                                hc = save_house_catalog_enrichment(db, enrichment)
                                counters.houses_enriched += 1
                                sweep_consecutive_house_blocks = 0
                                hid = hc.get("house_id")
                                if hid:
                                    touched_house_ids.add(int(hid))
                                if h_idx < len(house_paths_list) - 1:
                                    await asyncio.sleep(request_delay_sec)
                                h_idx += 1
                            except (AvitoBlockedError, AvitoRateLimitedError) as be:
                                # #1790: пробуем сменить IP перед re-raise.
                                sweep_consecutive_house_blocks += 1
                                counters.houses_failed += 1
                                counters.errors_count += 1
                                logger.warning(
                                    "city-sweep run_id=%d: house BLOCKED at %s "
                                    "(consecutive=%d) — %s",
                                    run_id,
                                    house_path,
                                    sweep_consecutive_house_blocks,
                                    be,
                                )
                                if (
                                    sweep_consecutive_house_blocks
                                    >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT
                                    and sweep_rotations_done < settings.avito_proxy_max_rotations
                                ):
                                    rotated = await _rotate_proxy_ip(
                                        reason=(
                                            f"city-sweep house "
                                            f"consecutive={sweep_consecutive_house_blocks}"
                                        ),
                                        rotations_done=sweep_rotations_done,
                                    )
                                    sweep_rotations_done += 1
                                    if rotated:
                                        sweep_consecutive_house_blocks = 0
                                        logger.info(
                                            "city-sweep run_id=%d: house ROTATE — "
                                            "IP changed, reset consecutive "
                                            "(rotation #%d/%d)",
                                            run_id,
                                            sweep_rotations_done,
                                            settings.avito_proxy_max_rotations,
                                        )
                                        # Не инкрементируем h_idx — retry текущего дома
                                        continue
                                elif sweep_consecutive_house_blocks >= 3:
                                    logger.error(
                                        "city-sweep run_id=%d: house ABORT — "
                                        "%d consecutive blocks, propagating",
                                        run_id,
                                        sweep_consecutive_house_blocks,
                                    )
                                    raise
                                h_idx += 1
                            except Exception as he:
                                logger.warning(
                                    "city-sweep run_id=%d: house_enrich failed %s: %s",
                                    run_id,
                                    house_path,
                                    he,
                                )
                                counters.houses_failed += 1
                                counters.errors_count += 1
                                try:
                                    db.rollback()
                                except Exception:
                                    pass
                                h_idx += 1

                    # ── Phase 5: enrich detail (top-N) ────────────────────────
                    if detail_top_n > 0 and anchor_lots:
                        priority_rows = (
                            db.execute(
                                text("""
                                    SELECT source_url
                                    FROM listings
                                    WHERE source = 'avito'
                                      AND source_url IS NOT NULL
                                      AND (
                                        (
                                          detail_enriched_at IS NULL
                                          AND price_rub > 0
                                          AND ST_DWithin(
                                            geom::geography,
                                            ST_MakePoint(:lon, :lat)::geography,
                                            :radius
                                          )
                                        )
                                        OR (
                                          detail_enriched_at IS NULL
                                          AND scraped_at > NOW() - INTERVAL '2 hours'
                                        )
                                      )
                                    ORDER BY scraped_at DESC NULLS LAST
                                    LIMIT :limit
                                """),
                                {
                                    "lat": _lat,
                                    "lon": _lon,
                                    "radius": radius_m * 2,
                                    "limit": detail_top_n,
                                },
                            )
                            .mappings()
                            .all()
                        )

                        consecutive_blocks = 0
                        consecutive_timeouts = 0
                        detail_house_paths: set[str] = set()
                        for d_idx, row in enumerate(priority_rows):
                            source_url: str = row["source_url"]
                            counters.detail_attempted += 1
                            try:
                                item_url = (
                                    urlparse(source_url).path
                                    if source_url.startswith("http")
                                    else source_url
                                )
                                enrichment_detail = await fetch_detail(
                                    item_url,
                                    cffi_session=session,
                                    browser_fetcher=shared_bf,
                                )
                                if save_detail_enrichment(db, enrichment_detail):
                                    counters.detail_enriched += 1
                                if enrichment_detail.house_catalog_url:
                                    hp = urlparse(enrichment_detail.house_catalog_url).path
                                    if hp:
                                        detail_house_paths.add(hp)
                                consecutive_blocks = 0
                                consecutive_timeouts = 0
                            except (AvitoBlockedError, AvitoRateLimitedError) as be:
                                consecutive_blocks += 1
                                counters.detail_failed += 1
                                counters.errors_count += 1
                                logger.warning(
                                    "city-sweep run_id=%d: detail BLOCKED #%d/%d "
                                    "(consecutive=%d): %s",
                                    run_id,
                                    d_idx + 1,
                                    len(priority_rows),
                                    consecutive_blocks,
                                    be,
                                )
                                if consecutive_blocks >= 3:
                                    # #1790: попытка ротации перед abort'ом.
                                    if sweep_rotations_done < settings.avito_proxy_max_rotations:
                                        rotated = await _rotate_proxy_ip(
                                            reason=(
                                                f"city-sweep detail "
                                                f"consecutive={consecutive_blocks}"
                                            ),
                                            rotations_done=sweep_rotations_done,
                                        )
                                        sweep_rotations_done += 1
                                        if rotated:
                                            consecutive_blocks = 0
                                            logger.info(
                                                "city-sweep run_id=%d: detail ROTATE — "
                                                "IP changed, reset consecutive "
                                                "(rotation #%d/%d)",
                                                run_id,
                                                sweep_rotations_done,
                                                settings.avito_proxy_max_rotations,
                                            )
                                            continue
                                    logger.error(
                                        "city-sweep run_id=%d: detail ABORT — "
                                        "%d consecutive blocks (IP rate-limited)",
                                        run_id,
                                        consecutive_blocks,
                                    )
                                    raise
                            except Exception as de:
                                counters.detail_failed += 1
                                counters.errors_count += 1
                                consecutive_timeouts += 1
                                logger.warning(
                                    "city-sweep run_id=%d: detail failed #%d/%d "
                                    "(consecutive_timeouts=%d): %s",
                                    run_id,
                                    d_idx + 1,
                                    len(priority_rows),
                                    consecutive_timeouts,
                                    de,
                                )
                                # (C) Fail-fast: браузер явно деградирует — прерываем
                                # detail-фазу, не тратим время на оставшиеся запросы.
                                if consecutive_timeouts >= _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT:
                                    logger.error(
                                        "city-sweep run_id=%d: detail ABORT — "
                                        "%d consecutive timeouts/errors, browser degraded",
                                        run_id,
                                        consecutive_timeouts,
                                    )
                                    break
                                try:
                                    db.rollback()
                                except Exception:
                                    pass

                            if d_idx < len(priority_rows) - 1:
                                jitter = random.uniform(0.8, 1.2)
                                await asyncio.sleep(request_delay_sec * jitter)

                        # ── Phase 5b: houses найденные через detail-страницы ──
                        new_house_paths = detail_house_paths - unique_house_paths
                        if enrich_houses and new_house_paths:
                            counters.unique_houses += len(new_house_paths)
                            nh_list = list(new_house_paths)
                            nh_idx = 0
                            sweep_consecutive_detail_house_blocks = 0  # #1790
                            while nh_idx < len(nh_list):
                                house_path = nh_list[nh_idx]
                                try:
                                    enrichment = await fetch_house_catalog(
                                        house_path,
                                        cffi_session=session,
                                        browser_fetcher=shared_bf,
                                    )
                                    hc = save_house_catalog_enrichment(db, enrichment)
                                    counters.houses_enriched += 1
                                    sweep_consecutive_detail_house_blocks = 0
                                    hid = hc.get("house_id")
                                    if hid:
                                        touched_house_ids.add(int(hid))
                                    if nh_idx < len(nh_list) - 1:
                                        await asyncio.sleep(request_delay_sec)
                                    nh_idx += 1
                                except (AvitoBlockedError, AvitoRateLimitedError) as beh:
                                    # #1790: ротация перед re-raise.
                                    sweep_consecutive_detail_house_blocks += 1
                                    counters.houses_failed += 1
                                    counters.errors_count += 1
                                    logger.warning(
                                        "city-sweep run_id=%d: house(detail) BLOCKED %s"
                                        " (consecutive=%d) — %s",
                                        run_id,
                                        house_path,
                                        sweep_consecutive_detail_house_blocks,
                                        beh,
                                    )
                                    if (
                                        sweep_consecutive_detail_house_blocks >= 3
                                        and sweep_rotations_done
                                        < settings.avito_proxy_max_rotations
                                    ):
                                        rotated = await _rotate_proxy_ip(
                                            reason=(
                                                "city-sweep house(detail) "
                                                f"consecutive="
                                                f"{sweep_consecutive_detail_house_blocks}"
                                            ),
                                            rotations_done=sweep_rotations_done,
                                        )
                                        sweep_rotations_done += 1
                                        if rotated:
                                            sweep_consecutive_detail_house_blocks = 0
                                            logger.info(
                                                "city-sweep run_id=%d: house(detail) "
                                                "ROTATE — IP changed (rotation #%d/%d)",
                                                run_id,
                                                sweep_rotations_done,
                                                settings.avito_proxy_max_rotations,
                                            )
                                            continue
                                    elif sweep_consecutive_detail_house_blocks >= 3:
                                        logger.error(
                                            "city-sweep run_id=%d: house(detail) ABORT"
                                            " — %d consecutive blocks, propagating",
                                            run_id,
                                            sweep_consecutive_detail_house_blocks,
                                        )
                                        raise
                                    nh_idx += 1
                                except Exception as he:
                                    logger.warning(
                                        "city-sweep run_id=%d: house(detail) failed %s: %s",
                                        run_id,
                                        house_path,
                                        he,
                                    )
                                    counters.houses_failed += 1
                                    counters.errors_count += 1
                                    try:
                                        db.rollback()
                                    except Exception:
                                        pass
                                    nh_idx += 1

                    all_touched_house_ids.update(touched_house_ids)
                    logger.info(
                        "city-sweep run_id=%d anchor %s done: "
                        "lots=%d houses=%d/%d detail=%d/%d touched=%d errors=%d",
                        run_id,
                        _name,
                        len(anchor_lots),
                        counters.houses_enriched,
                        counters.unique_houses,
                        counters.detail_enriched,
                        counters.detail_attempted,
                        len(touched_house_ids),
                        counters.errors_count,
                    )

                try:
                    await asyncio.wait_for(_avito_anchor_phases(), timeout=_avito_anchor_timeout)
                except TimeoutError:
                    logger.warning(
                        "city-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
                        "timed out after %.0fs — SERP counters preserved, "
                        "detail phase incomplete",
                        run_id,
                        idx,
                        len(_anchors),
                        name,
                        lat,
                        lon,
                        _avito_anchor_timeout,
                    )
                    counters.errors_count += 1
                except (AvitoBlockedError, AvitoRateLimitedError) as e:
                    logger.error(
                        "city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s",
                        run_id,
                        idx,
                        len(_anchors),
                        name,
                        e,
                    )
                    counters.errors_count += 1
                    counters.anchors_done = idx
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                    # #1950: если SERP уже собрал лоты и заблокировало только detail/houses,
                    # ставим 'done' (не 'banned') — partial intake сохранён.
                    # За флагом avito_serp_ok_not_banned (default True).
                    _serp_ok = (counters.lots_inserted + counters.lots_updated) > 0
                    if settings.avito_serp_ok_not_banned and _serp_ok:
                        _note = (
                            f"detail enrichment aborted ({e}) но SERP intake завершён: "
                            f"{counters.lots_inserted} ins {counters.lots_updated} upd"
                        )
                        logger.warning(
                            "city-sweep run_id=%d: SERP OK (ins=%d upd=%d), "
                            "detail blocked — DONE not banned. %s",
                            run_id,
                            counters.lots_inserted,
                            counters.lots_updated,
                            _note,
                        )
                        scrape_runs.mark_done(
                            db,
                            run_id,
                            {**counters.to_dict(), "enrichment_abort_note": _note},  # type: ignore[arg-type]
                        )
                    else:
                        scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict())
                    return counters
                except Exception:
                    logger.exception("city-sweep run_id=%d: anchor %s failed", run_id, name)
                    counters.errors_count += 1

                counters.anchors_done = idx
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

            # ── IMV-фаза: финальный обход тронутых домов ──────────
            if enrich_imv and all_touched_house_ids:
                if scrape_runs.is_cancelled(db, run_id):
                    logger.info(
                        "city-sweep run_id=%d: cancelled before IMV phase (%d houses skipped)",
                        run_id,
                        len(all_touched_house_ids),
                    )
                    scrape_runs.mark_done(db, run_id, counters.to_dict())
                    return counters
                elif shutdown_requested():
                    # SIGTERM-drain до IMV-фазы: финализируем без дорогой IMV-оценки.
                    logger.info(
                        "city-sweep run_id=%d: SIGTERM-drain — skipping IMV phase "
                        "(%d houses skipped)",
                        run_id,
                        len(all_touched_house_ids),
                    )
                    scrape_runs.mark_done(db, run_id, counters.to_dict())
                    return counters

                logger.info(
                    "city-sweep run_id=%d: IMV phase — %d touched houses",
                    run_id,
                    len(all_touched_house_ids),
                )
                try:
                    # Периодический heartbeat внутри длинной IMV-фазы (#1363):
                    # без него десятки/сотни домов × request_delay_sec проходят
                    # без update_heartbeat, и reap_zombies помечает живой run
                    # 'zombie' → последующий mark_done становится no-op (дубль-sweep).
                    def _imv_heartbeat() -> None:
                        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

                    imv_result = await process_houses_imv_batch(
                        db,
                        all_touched_house_ids,
                        request_delay_sec=request_delay_sec,
                        heartbeat=_imv_heartbeat,
                    )
                    counters.imv_attempted += imv_result.checked
                    counters.imv_enriched += imv_result.saved
                    counters.imv_failed += imv_result.errors
                    counters.errors_count += imv_result.errors
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                    logger.info(
                        "city-sweep run_id=%d: IMV phase done — attempted=%d enriched=%d failed=%d",
                        run_id,
                        imv_result.checked,
                        imv_result.saved,
                        imv_result.errors,
                    )
                except Exception as exc:
                    logger.error(
                        "city-sweep run_id=%d: IMV phase crashed — %r (sweep still done)",
                        run_id,
                        exc,
                    )
                    counters.errors_count += 1
                    # #1521: IMV-фаза могла оставить Session в aborted-state
                    # (необёрнутый eligibility SELECT / _mark_status на отравленной
                    # сессии). update_heartbeat и mark_done — единственные финализаторы
                    # БЕЗ defensive rollback, поэтому сбрасываем txn-state сами. Иначе
                    # PendingRollbackError пролетает мимо внутреннего try во внешний
                    # except (mark_failed) → успешный sweep ложно помечается 'failed'.
                    try:
                        db.rollback()
                    except Exception:
                        pass
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

            scrape_runs.mark_done(db, run_id, counters.to_dict())
            logger.info(
                "city-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
                "houses=%d/%d detail=%d/%d imv=%d/%d errors=%d",
                run_id,
                counters.anchors_done,
                counters.anchors_total,
                counters.lots_fetched,
                counters.lots_inserted,
                counters.lots_updated,
                counters.houses_enriched,
                counters.unique_houses,
                counters.detail_enriched,
                counters.detail_attempted,
                counters.imv_enriched,
                counters.imv_attempted,
                counters.errors_count,
            )
            return counters

        except Exception as exc:
            logger.exception("city-sweep run_id=%d: fatal error", run_id)
            scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
            raise


# ── Avito newbuilding (novostroyka) citywide sweep ────────────────


@dataclass
class NewbuildingSweepCounters:
    """Aggregate counters для Avito novostroyka citywide sweep run."""

    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)}


async def run_avito_newbuilding_sweep(
    db: Session,
    *,
    run_id: int,
    pages: int = 20,
    request_delay_sec: float = 7.0,
) -> NewbuildingSweepCounters:
    """Citywide-обход ЕКБ-выборки только новостроек (novostroyka-filter) → save.

    В отличие от run_avito_city_sweep (anchor-based fetch_around + houses/detail/IMV),
    это простой citywide paginated-обход через AvitoScraper.fetch_newbuildings:
    dedicated novostroyka-SERP отдаёт 100% new-build карточки (listing_segment=
    "novostroyki", newbuilding_id/newbuilding_url заполнены парсером). Сохраняем через
    стандартный save_listings, ведём SERP-счётчики через scrape_runs.

    - Single shared AsyncSession/BrowserFetcher на весь sweep (один TLS fingerprint)
    - AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned')
    - is_cancelled-чек до старта SERP-фазы
    """
    counters = NewbuildingSweepCounters()

    browser_mode = settings.scraper_fetch_mode == "browser"
    async with AsyncExitStack() as stack:
        session: AsyncSession | None = None
        shared_bf: BrowserFetcher | None = None
        if browser_mode:
            shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito"))
        else:
            session = await stack.enter_async_context(
                AsyncSession(
                    impersonate="chrome120",
                    timeout=25,
                    headers=_CHROME_HEADERS,
                    proxies=_avito_proxies(),
                )
            )
        try:
            if scrape_runs.is_cancelled(db, run_id):
                logger.info("nb-sweep run_id=%d: cancelled before SERP phase", run_id)
                scrape_runs.mark_done(db, run_id, counters.to_dict())
                return counters
            elif shutdown_requested():
                logger.info(
                    "nb-sweep run_id=%d: SIGTERM-drain — stopping before SERP phase", run_id
                )
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                scrape_runs.mark_done(db, run_id, counters.to_dict())
                return counters

            scraper = AvitoScraper()
            if browser_mode:
                scraper._browser = shared_bf
                # Shared-browser режим: _cffi=None → curl_cffi-fallback на firewall
                # пропускается (by design). Логируем для observability.
                logger.debug("avito pipeline-mode: shared browser only, no cffi fallback")
            else:
                scraper._cffi = session

            try:
                lots: list[ScrapedLot] = await scraper.fetch_newbuildings(
                    pages=pages,
                    delay_override_sec=request_delay_sec,
                )
            except (AvitoBlockedError, AvitoRateLimitedError) as e:
                logger.error("nb-sweep run_id=%d: SERP BLOCKED — %s", run_id, e)
                scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict())
                return counters

            counters.lots_fetched += len(lots)
            if lots:
                try:
                    ins, upd = save_listings(db, lots)
                    counters.lots_inserted += ins
                    counters.lots_updated += upd
                except Exception as save_exc:
                    logger.exception(
                        "nb-sweep run_id=%d: save_listings failed: %s", run_id, save_exc
                    )
                    counters.errors_count += 1
                    try:
                        db.rollback()
                    except Exception:
                        pass

            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
            scrape_runs.mark_done(db, run_id, counters.to_dict())
            logger.info(
                "nb-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) errors=%d",
                run_id,
                counters.lots_fetched,
                counters.lots_inserted,
                counters.lots_updated,
                counters.errors_count,
            )
            return counters

        except Exception as exc:
            logger.exception("nb-sweep run_id=%d: fatal error", run_id)
            scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
            raise


# ── Yandex city sweep ───────────────────────────────────────────


@dataclass
class YandexCitySweepCounters:
    """Aggregate counters для Yandex.Недвижимость city sweep run.

    Фазы: SERP (fetch_around_multi_room) → save_listings инкрементально → address-enrich.
    address_* — результат T10-обогащения (detail <title> → полный адрес с номером дома).
    combos_skipped — сколько (room×price) combo пропущено из-за таймаутов gate-API.
    """

    anchors_total: int = 0
    anchors_done: int = 0
    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: int = 0
    price_history_rows: int = 0
    combos_skipped: int = 0
    address_attempted: int = 0
    address_enriched: int = 0
    address_failed: 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 = 25000,
    pages_per_anchor: int = 2,
    request_delay_sec: float | None = None,
    enrich_address: bool = True,
    rooms_list: list[str] | None = None,
    price_ranges: list[tuple[int | None, int | None]] | None = None,
    segments: list[str] | None = None,
) -> YandexCitySweepCounters:
    """Yandex.Недвижимость city sweep: rooms × price combos от центра ЕКБ → save → address-enrich.

    Вместо 5 географических anchor'ов итерирует ВСЕ combinations (room × price_range)
    из одной центральной точки ЕКБ (56.8400, 60.6050). radius_m=25000 охватывает весь
    город; price/room segmentation обходит Yandex SERP ~575-card cap без geo-шрапнели.

    Фазы (единственный центральный якорь):
      1. SERP: YandexRealtyScraper.fetch_around_multi_room(lat, lon, radius_m,
               max_pages=pages_per_anchor, rooms_list=..., price_ranges=...) — combos mode.
      2. SAVE: save_listings(db, lots, run_id=run_id).
      3. ADDRESS-ENRICH (если enrich_address=True): для yandex-листингов без номера
         дома в address (address IS NOT NULL AND NOT (address ~ ',\\s*\\d+'))
         запрашиваем detail-страницу и извлекаем полный адрес из <title>.

    Параметр `anchors` сохранён для backward-compat тестов (anchors=[...]). В prod
    (anchors=None) всегда используется один EKB_CENTER anchor.

    Anti-bot / cancel семантика:
      - Cooperative cancel: is_cancelled(db, run_id) проверяется перед фазой SERP.
      - YandexRealtyScraper не propagate'ит block/rate exceptions (глотает → []).
        Ловим broad Exception; при N подряд — mark_banned + abort.
      - request_delay_sec: inter-request delay для address-enrich (поверх per-page
        sleep внутри scraper'а).

    Возвращает YandexCitySweepCounters.
    mark_done вызывается ВСЕГДА (кроме cancel / ban / fatal).
    """
    from sqlalchemy import text as _text

    from app.services.scrapers.yandex_realty import DEFAULT_PRICE_RANGES, ROOM_PATH
    from app.services.yandex_address_backfill import (
        _RE_HAS_HOUSE_NUMBER,
        _extract_address_from_title,
    )

    # EKB center — единственный anchor для citywide combos sweep.
    # anchors-параметр принимается для backward-compat тестов.
    if anchors is not None:
        _anchors: list[tuple[float, float, str]] = anchors
    else:
        _anchors = [(56.8400, 60.6050, "Центр (combos)")]

    _rooms_list = rooms_list or list(ROOM_PATH.keys())
    _price_ranges = price_ranges or DEFAULT_PRICE_RANGES
    # newFlat-сегменты: оба прохода (vtorichka + novostroyki) по одним combos.
    # Дефолт ["NO", "YES"] — оба; dedup по source_id span'ит оба прохода (один seen
    # внутри fetch_around_multi_room). Counters (lots_fetched/inserted/updated)
    # агрегируют оба прохода через len(anchor_lots).
    _segments = segments or ["NO", "YES"]

    counters = YandexCitySweepCounters(anchors_total=len(_anchors))
    inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
    enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0
    _resolved_delay = request_delay_sec if request_delay_sec is not None else 9.0
    consecutive_failures = 0
    yandex_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep

    # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None).
    # В combos-режиме один "anchor" выполняет num_segments × num_combos × max_pages
    # fetch'ей — каждый занимает примерно _resolved_delay + _YANDEX_COMBOS_PER_FETCH_S сек.
    # Для 2 segments × 30 combos × 3 pages × (9+12) + 300 ≈ 4080s (~68 мин).
    # Множитель len(_segments) масштабирует watchdog пропорционально числу проходов
    # (vtorichka + novostroyki ≈ ×2 wall-clock) — иначе второй проход гильотинится.
    # В explicit-anchor (тест/override) режиме оставляем ANCHOR_TIMEOUT_SEC (240s) —
    # там каждый anchor небольшой и watchdog работает как старый защитный барьер.
    _num_combos = len(_rooms_list) * len(_price_ranges)
    if anchors is None and _num_combos > 0:
        _sweep_timeout = max(
            ANCHOR_TIMEOUT_SEC,
            len(_segments)
            * _num_combos
            * pages_per_anchor
            * (_resolved_delay + _YANDEX_COMBOS_PER_FETCH_S)
            + _YANDEX_ADDRESS_ENRICH_BUDGET_S,
        )
    else:
        # Explicit anchors (backward-compat тесты или ручной override) — legacy timeout.
        _sweep_timeout = ANCHOR_TIMEOUT_SEC

    # Прокси для address-enrich curl_cffi сессии (зеркало yandex_address_backfill).
    _proxy_url = settings.scraper_proxy_url
    _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None

    try:
        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
            # ── Cooperative cancel ───────────────────────────────────────────
            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
            elif shutdown_requested():
                # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial).
                logger.info(
                    "yandex-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)",
                    run_id,
                    idx,
                    len(_anchors),
                    name,
                )
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                scrape_runs.mark_done(db, run_id, counters.to_dict())
                return counters

            logger.info(
                "yandex-sweep run_id=%d center-combos #%d/%d (%s, %.4f, %.4f) "
                "radius_m=%d rooms=%d price_ranges=%d",
                run_id,
                idx,
                len(_anchors),
                name,
                lat,
                lon,
                radius_m,
                len(_rooms_list),
                len(_price_ranges),
            )

            # ── Per-anchor phases под watchdog-таймаутом ────────────────────
            # Весь SERP + address-enrich одного anchor'а оборачиваем в wait_for.
            # TimeoutError → log + errors_count++ + continue (sweep не прерывается).

            anchor_lots: list[ScrapedLot] = []
            _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

            # _combo_lots_accumulator: per-anchor мутируемый контейнер для on_combo.
            # Захватывается по ссылке и в _yandex_anchor_phases и в _on_combo —
            # избегаем B023 (loop-variable capture) через явный bind в default-arg.
            _combo_lots_accumulator: list[ScrapedLot] = anchor_lots

            async def _yandex_anchor_phases(
                _a_lat: float = _lat,
                _a_lon: float = _lon,
                _a_name: str = _name,
                _al: list[ScrapedLot] = _combo_lots_accumulator,
            ) -> None:
                """Все фазы одного anchor'а (SERP + address-enrich).

                FIX: инкрементальный save per-combo — on_combo вызывается scraper'ом после
                каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat.
                anchor_lots накапливается (через _al) для address-enrich и price-history.
                """
                nonlocal consecutive_failures, yandex_rotations_done

                # ── Phase 1+2: SERP + инкрементальный save per-combo ──────
                def _on_combo(
                    combo_label: str,
                    new_lots: list[ScrapedLot],
                    _accumulator: list[ScrapedLot] = _al,
                ) -> None:
                    """Callback: вызывается scraper'ом после каждого combo с новыми лотами.

                    Сохраняет порцию новых лотов немедленно (анти-зомби: данные в БД
                    до конца anchor'а) и обновляет heartbeat для мониторинга прогресса.
                    """
                    _accumulator.extend(new_lots)
                    counters.lots_fetched += len(new_lots)
                    try:
                        ins, upd = save_listings(db, new_lots, run_id=run_id)
                        counters.lots_inserted += ins
                        counters.lots_updated += upd
                    except Exception as save_exc:
                        counters.errors_count += 1
                        logger.warning(
                            "yandex-sweep run_id=%d combo %s: save_listings failed: %s",
                            run_id,
                            combo_label,
                            save_exc,
                        )
                        try:
                            db.rollback()
                        except Exception:
                            pass
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                    logger.debug(
                        "yandex-sweep run_id=%d combo %s: saved %d lots "
                        "(ins=%d upd=%d total_fetched=%d)",
                        run_id,
                        combo_label,
                        len(new_lots),
                        counters.lots_inserted,
                        counters.lots_updated,
                        counters.lots_fetched,
                    )

                async with YandexRealtyScraper() as scraper:
                    await scraper.fetch_around_multi_room(
                        _a_lat,
                        _a_lon,
                        radius_m,
                        max_pages=pages_per_anchor,
                        rooms_list=_rooms_list,
                        price_ranges=_price_ranges,
                        segments=_segments,
                        on_combo=_on_combo,
                    )
                # _al накоплен on_combo; save уже вызван per-combo.

                # Price-history из gate price.previous/trend (graceful — провал
                # истории НЕ должен ронять сейв листингов).
                if _al:
                    try:
                        counters.price_history_rows += record_yandex_price_history(db, _al)
                    except Exception as ph_exc:
                        logger.warning(
                            "yandex-sweep run_id=%d: price-history failed: %s",
                            run_id,
                            ph_exc,
                        )
                        try:
                            db.rollback()
                        except Exception:
                            pass

                consecutive_failures = 0
                logger.info(
                    "yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d",
                    run_id,
                    _a_name,
                    counters.lots_fetched,
                    counters.lots_inserted,
                    counters.lots_updated,
                )

                # ── Phase 3: address-enrich ────────────────────────────────
                if not (enrich_address and _al):
                    return
                source_ids = [lot.source_id for lot in _al 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

                    # #1848: rotation tracking для Yandex address-enrich (stuck-proxy)
                    _yandex_enrich_consec_fail = 0
                    _yandex_enrich_abort = 3  # abort address-enrich после N подряд

                    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:
                        eidx = 0
                        while eidx < len(items):
                            item = items[eidx]
                            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 not in {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
                                    # 503/429 = прокси заблокирован → считаем как failure
                                    if resp.status_code in {429, 503}:
                                        _yandex_enrich_consec_fail += 1
                                    else:
                                        _yandex_enrich_consec_fail = 0
                                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
                                    _yandex_enrich_consec_fail = 0
                            except Exception as fetch_exc:
                                counters.address_failed += 1
                                _yandex_enrich_consec_fail += 1
                                logger.warning(
                                    "yandex-sweep run_id=%d: address-enrich "
                                    "fetch failed listing_id=%d "
                                    "(consecutive=%d): %s",
                                    run_id,
                                    lid,
                                    _yandex_enrich_consec_fail,
                                    fetch_exc,
                                )
                            # Ротация Yandex-прокси при stuck/ban (#1848)
                            if _yandex_enrich_consec_fail >= _yandex_enrich_abort:
                                if yandex_rotations_done < settings.yandex_proxy_max_rotations:
                                    rotated = await _rotate_proxy_ip(
                                        reason=(
                                            f"yandex enrich consecutive="
                                            f"{_yandex_enrich_consec_fail}"
                                        ),
                                        rotations_done=yandex_rotations_done,
                                        source="yandex",
                                    )
                                    yandex_rotations_done += 1
                                    if rotated:
                                        _yandex_enrich_consec_fail = 0
                                        logger.info(
                                            "yandex-sweep run_id=%d: enrich ROTATE — "
                                            "IP changed, reset consecutive "
                                            "(rotation #%d/%d)",
                                            run_id,
                                            yandex_rotations_done,
                                            settings.yandex_proxy_max_rotations,
                                        )
                                        # retry текущего item после ротации
                                        continue
                                logger.error(
                                    "yandex-sweep run_id=%d: address-enrich ABORT — "
                                    "%d consecutive failures (proxy likely stuck/banned)",
                                    run_id,
                                    _yandex_enrich_consec_fail,
                                )
                                break
                            if eidx < len(items) - 1:
                                await asyncio.sleep(enrich_delay)
                            eidx += 1

                    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=_sweep_timeout)
            except TimeoutError:
                logger.warning(
                    "yandex-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
                    "timed out after %.0fs — skipping",
                    run_id,
                    idx,
                    len(_anchors),
                    name,
                    lat,
                    lon,
                    _sweep_timeout,
                )
                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 {_sweep_timeout:.0f}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)
                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'ом даже при ошибке.
                if idx < len(_anchors):
                    await asyncio.sleep(inter_anchor_delay)
                continue

            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 lots=%d (ins=%d/upd=%d) "
            "address=%d/%d failed=%d errors=%d",
            run_id,
            counters.anchors_done,
            counters.anchors_total,
            counters.lots_fetched,
            counters.lots_inserted,
            counters.lots_updated,
            counters.address_enriched,
            counters.address_attempted,
            counters.address_failed,
            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


# ── Cian city sweep (#860) ──────────────────────────────────────────────────


@dataclass
class CianCitySweepCounters:
    """Aggregate counters для Cian city sweep run.

    Поля симметричны CitySweepCounters, но без imv_* (у Cian нет IMV-фазы).
    detail_* — обогащение через cian_detail (вкл. price-history).
    houses_* — обогащение newbuilding/ЖК через cian_newbuilding.
    """

    anchors_total: int = 0
    anchors_done: int = 0
    lots_fetched: int = 0
    lots_dropped_secondary: int = 0  # вторичка, отброшенная при newbuilding_only=True
    lots_inserted: int = 0
    lots_updated: int = 0
    detail_attempted: int = 0
    detail_enriched: int = 0
    detail_failed: int = 0
    houses_attempted: int = 0
    houses_enriched: int = 0
    houses_failed: 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: если N подряд anchor'ов упали с исключением
# (не просто вернули пустой результат) — прерываем sweep.
CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES = 3


async def run_cian_city_sweep(
    db: Session,
    *,
    run_id: int,
    anchors: list[tuple[float, float, str]] | None = None,
    radius_m: int = 1500,
    pages_per_anchor: int = 3,
    request_delay_sec: float = 5.0,
    detail_top_n: int = 10,
    enrich_houses: bool = True,
    newbuilding_only: bool = True,
) -> CianCitySweepCounters:
    """Cian newbuilding city sweep: SERP → detail(+price-history) → newbuilding/houses.

    Симметрично run_avito_city_sweep. Использует EKB_ANCHORS (общий список).

    newbuilding_only (default True): в SERP-фазе сохраняются только лоты с
    listing_segment == "novostroyki" — вторичку отбрасываем, ею авторитетно владеет
    run_cian_full_load (exhaustive региональный сбор). Зеркало secondary_only в
    fetch_all_secondary, которое фильтрует обратное (!= "novostroyki"). DETAIL-фаза
    (обогащение cian-листингов без detail_enriched_at) и HOUSES-фаза (newbuilding-enrich)
    не затронуты фильтром.

    Фазы per anchor:
      1. SERP: CianScraper.fetch_around_multi_room(lat, lon, radius_m, pages=pages_per_anchor)
      2. FILTER (newbuilding_only): отбросить вторичку (listing_segment != "novostroyki")
      3. SAVE: save_listings(db, lots, run_id=run_id)
      4. DETAIL: top-N свежих cian-листингов без detail_enriched_at →
             fetch_detail(source_url) + save_detail_enrichment(db, listing_id, enrichment)
             (save_detail_enrichment уже пишет price_changes → offer_price_history)
      5. HOUSES (если enrich_houses=True): для домов из этого anchor'а с cian_zhk_url →
             fetch_newbuilding(zhk_url) + save_newbuilding_enrichment(db, house_id, enr)

    Anti-bot:
      - CianScraper управляет своей curl_cffi-сессией (прокси уже внутри).
      - asyncio.sleep(request_delay_sec) между detail-запросами (jitter ±20%).
      - Defensive abort при CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES SERP-ошибок подряд.

    Cooperative cancel: scrape_runs.is_cancelled(db, run_id) перед каждым anchor.
    mark_done вызывается ВСЕГДА (finally outer).
    """
    from app.services.scrapers.cian import CianScraper
    from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment
    from app.services.scrapers.cian_newbuilding import (
        fetch_newbuilding,
        save_newbuilding_enrichment,
    )

    _anchors = anchors if anchors is not None else EKB_ANCHORS
    counters = CianCitySweepCounters(anchors_total=len(_anchors))
    consecutive_failures = 0
    cian_rotations_done = 0  # #1848: бюджет IP-ротаций на весь sweep

    try:
        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
            # Cooperative cancel перед каждым anchor
            if scrape_runs.is_cancelled(db, run_id):
                logger.info(
                    "cian-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
            elif shutdown_requested():
                # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial).
                logger.info(
                    "cian-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)",
                    run_id,
                    idx,
                    len(_anchors),
                    name,
                )
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                scrape_runs.mark_done(db, run_id, counters.to_dict())
                return counters

            logger.info(
                "cian-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
                run_id,
                idx,
                len(_anchors),
                name,
                lat,
                lon,
            )

            # ── Per-anchor phases под watchdog-таймаутом ────────────────────
            # Весь SERP + detail + houses одного anchor'а оборачиваем в wait_for.
            # TimeoutError → log + errors_count++ + continue (sweep не прерывается).

            anchor_lots: list[ScrapedLot] = []
            # 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, cian_rotations_done
                from sqlalchemy import text as _text

                # ── Phase 1+2: SERP + save ─────────────────────────────────
                async with CianScraper() as scraper:
                    anchor_lots = await scraper.fetch_around_multi_room(
                        _a_lat,
                        _a_lon,
                        radius_m,
                        pages=pages_per_anchor,
                    )
                counters.lots_fetched += len(anchor_lots)
                # ── Фильтр вторички (newbuilding_only) ─────────────────────
                # Зеркало secondary_only в fetch_all_secondary: тут оставляем
                # только новостройки — вторичкой авторитетно владеет run_cian_full_load.
                if newbuilding_only:
                    _before = len(anchor_lots)
                    anchor_lots = [
                        lot for lot in anchor_lots if lot.listing_segment == "novostroyki"
                    ]
                    counters.lots_dropped_secondary += _before - 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
                logger.info(
                    "cian-sweep run_id=%d anchor %s: SERP fetched=%d nb_kept=%d "
                    "dropped_secondary=%d ins=%d upd=%d",
                    run_id,
                    _a_name,
                    counters.lots_fetched,
                    len(anchor_lots),
                    counters.lots_dropped_secondary,
                    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()
                    )
                    # #1848: Cian bans/stuck-proxy tracking + rotation (зеркало avito)
                    _cian_detail_consec_failures = 0
                    _cian_detail_abort = 3  # abort detail-фазы после N подряд
                    didx = 0
                    while didx < len(priority_rows):
                        row = priority_rows[didx]
                        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,
                                )
                            _cian_detail_consec_failures = 0
                        except Exception as exc:
                            counters.detail_failed += 1
                            counters.errors_count += 1
                            _cian_detail_consec_failures += 1
                            logger.warning(
                                "cian-sweep run_id=%d: detail failed listing_id=%d "
                                "(consecutive=%d): %s",
                                run_id,
                                listing_id,
                                _cian_detail_consec_failures,
                                exc,
                            )
                            if _cian_detail_consec_failures >= _cian_detail_abort:
                                # Попытка ротации Cian-прокси перед abort'ом (#1848)
                                if cian_rotations_done < settings.cian_proxy_max_rotations:
                                    rotated = await _rotate_proxy_ip(
                                        reason=(
                                            f"cian detail consecutive="
                                            f"{_cian_detail_consec_failures}"
                                        ),
                                        rotations_done=cian_rotations_done,
                                        source="cian",
                                    )
                                    cian_rotations_done += 1
                                    if rotated:
                                        _cian_detail_consec_failures = 0
                                        logger.info(
                                            "cian-sweep run_id=%d: detail ROTATE — "
                                            "IP changed, reset consecutive "
                                            "(rotation #%d/%d)",
                                            run_id,
                                            cian_rotations_done,
                                            settings.cian_proxy_max_rotations,
                                        )
                                        # retry текущего detail (не инкрементируем didx)
                                        continue
                                logger.error(
                                    "cian-sweep run_id=%d: detail ABORT — "
                                    "%d consecutive failures (proxy likely banned/stuck)",
                                    run_id,
                                    _cian_detail_consec_failures,
                                )
                                break
                        if didx < len(priority_rows) - 1:
                            jitter = random.uniform(0.8, 1.2)
                            await asyncio.sleep(request_delay_sec * jitter)
                        didx += 1
                    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
                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 SERP 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"cian sweep aborted: {consecutive_failures} consecutive "
                        f"anchor SERP failures (last: {e})",
                        counters.to_dict(),
                    )
                    return counters
                counters.anchors_done = idx
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                continue

            counters.anchors_done = idx
            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

            # Пауза между anchor'ами (поверх per-request sleep внутри scraper'а)
            if idx < len(_anchors):
                await asyncio.sleep(request_delay_sec)

        scrape_runs.mark_done(db, run_id, counters.to_dict())
        logger.info(
            "cian-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) "
            "detail=%d/%d houses=%d/%d errors=%d",
            run_id,
            counters.anchors_done,
            counters.anchors_total,
            counters.lots_fetched,
            counters.lots_inserted,
            counters.lots_updated,
            counters.detail_enriched,
            counters.detail_attempted,
            counters.houses_enriched,
            counters.houses_attempted,
            counters.errors_count,
        )
        return counters

    except Exception as exc:
        logger.exception("cian-sweep run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise


# ── Cian exhaustive full load (region-wide, без anchor'ов) ─────────────────


@dataclass
class CianFullLoadCounters:
    """Aggregate counters для run_cian_full_load.

    Без anchor-полей — единственный региональный проход.
    detail_* — опциональное обогащение detail-страниц (если enrich_detail=True).
    """

    unique_fetched: int = 0
    saved_inserted: int = 0
    saved_updated: int = 0
    detail_attempted: int = 0
    detail_enriched: int = 0
    detail_failed: 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)}


async def run_cian_full_load(
    db: Session,
    *,
    run_id: int,
    price_cap_per_bucket: int = 1400,
    detail_top_n: int = 0,
    request_delay_sec: float = 4.0,
    enrich_detail: bool = False,
    concurrency: int = 5,
    resume_run_id: int | None = None,
) -> CianFullLoadCounters:
    """Exhaustive региональный сбор Cian ЕКБ вторички (БЕЗ anchor'ов).

    Cian SERP — region-wide, не geo-bbox. Anchor-loop избыточен (5 anchor'ов
    возвращают одну и ту же выдачу ×5). Один проход с партиционированием по
    КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление) охватывает весь ЕКБ.

    Параметры:
      price_cap_per_bucket: максимум офферов на price-бакет (< Cian SERP-cap ~1500).
      detail_top_n: сколько свежих cian-листингов без detail_enriched_at обогащать
                    (актуально только при enrich_detail=True).
      request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
      enrich_detail: включить detail-обогащение (по умолчанию отключено — тяжело).
      concurrency: параллельных page-фетчей в leaf-бакете (default=5).

    Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
    Краш/бан больше не приводит к потере всего накопленного.
    Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
    """
    from app.services.scrapers.cian import CianScraper

    counters = CianFullLoadCounters()

    # ── Checkpoint/resume: читаем done_buckets из прошлого run ───────────────
    skip_set: set[str] = set()
    if resume_run_id is not None:
        _prev_row = db.execute(
            text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"),
            {"rid": resume_run_id},
        ).fetchone()
        if _prev_row is not None and _prev_row.counters:
            _prev_counters: dict = (
                _prev_row.counters if isinstance(_prev_row.counters, dict) else {}
            )
            skip_set = set(_prev_counters.get("done_buckets", []))
        logger.info(
            "cian-full-load run_id=%d: resuming from run %d — %d buckets already done",
            run_id,
            resume_run_id,
            len(skip_set),
        )

    done: set[str] = set(skip_set)  # накапливаем завершённые бакеты этого прогона

    def _on_bucket(bucket_key: str, lots: list) -> None:  # type: ignore[type-arg]
        """Инкрементальный save после каждого leaf-бакета. Дописывает bucket_key в done."""
        nonlocal done
        if scrape_runs.is_cancelled(db, run_id):
            logger.info("cian-full-load run_id=%d: cancel detected in on_bucket", run_id)
            raise RuntimeError("cancelled")
        elif shutdown_requested():
            # SIGTERM-drain: тот же sentinel-механизм, что и user-cancel, но отдельный
            # маркер "shutdown" → handler финализирует mark_done(partial), не mark_failed.
            logger.info(
                "cian-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s",
                run_id,
                bucket_key,
            )
            raise RuntimeError("shutdown")
        if not lots:
            done.add(bucket_key)
            scrape_runs.update_heartbeat(
                db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
            )
            return
        inserted, updated = save_listings(
            db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today
        )
        # save_listings вызывает db.commit() внутри — данные в БД сразу
        counters.saved_inserted += inserted
        counters.saved_updated += updated
        counters.unique_fetched += len(lots)
        done.add(bucket_key)
        scrape_runs.update_heartbeat(
            db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
        )
        logger.info(
            "cian-full-load run_id=%d: bucket %s saved ins=%d upd=%d total_unique=%d",
            run_id,
            bucket_key,
            inserted,
            updated,
            counters.unique_fetched,
        )

    def _on_progress(unique_count: int) -> None:
        """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов)."""
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

    # #1949: фоновый heartbeat — обновляет heartbeat_at каждые 60s независимо от on_bucket.
    # Без него зависший asyncio.gather внутри bucket-loop замораживает heartbeat на часы
    # → reap_zombies помечает живой run zombie. Отменяется в finally при любом завершении.
    async def _background_heartbeat() -> None:
        """Периодический heartbeat каждые 60s, не завязанный на прогресс бакетов."""
        while True:
            await asyncio.sleep(60.0)
            try:
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
            except Exception:
                pass

    _hb_task: asyncio.Task[None] | None = None
    try:
        _hb_task = asyncio.create_task(_background_heartbeat())

        async with CianScraper() as scraper:
            scraper.request_delay_sec = request_delay_sec

            # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather.
            # BrowserFetcher.fetch имеет httpx-timeout 120s, но browser-сервис иногда
            # виснет так что httpx не получает ответ → gather морозит → heartbeat стоп.
            # Оборачиваем _browser.fetch в asyncio.wait_for(timeout=X) через monkey-patch:
            # TimeoutError → _fetch_page_html ловит как Exception → None → _one_page → []
            # → gather завершается нормально → on_bucket вызывается → heartbeat обновляется.
            # За флагом cian_full_load_per_fetch_timeout_s (0 = отключить).
            _fetch_timeout = settings.cian_full_load_per_fetch_timeout_s
            if _fetch_timeout > 0 and scraper._browser is not None:
                _orig_fetch = scraper._browser.fetch

                async def _timed_fetch(
                    url: str,
                    _f: Any = _orig_fetch,
                    _t: float = _fetch_timeout,
                    _rid: int = run_id,
                ) -> str:
                    try:
                        return await asyncio.wait_for(_f(url), timeout=_t)
                    except TimeoutError:
                        logger.warning(
                            "cian-full-load run_id=%d: browser-fetch timed out "
                            "after %.0fs — url=%s",
                            _rid,
                            _t,
                            url,
                        )
                        raise

                scraper._browser.fetch = _timed_fetch  # type: ignore[method-assign]
                logger.info(
                    "cian-full-load run_id=%d: per-fetch timeout %.0fs enabled",
                    run_id,
                    _fetch_timeout,
                )

            await scraper.fetch_all_secondary(
                price_cap_per_bucket=price_cap_per_bucket,
                concurrency=concurrency,
                secondary_only=True,
                on_bucket=_on_bucket,
                on_progress=_on_progress,
                skip_buckets=skip_set if skip_set else None,
            )

        logger.info(
            "cian-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",
            run_id,
            counters.unique_fetched,
            counters.saved_inserted,
            counters.saved_updated,
        )
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

        # ── Опциональный detail-энричмент ─────────────────────────────────────
        if enrich_detail and detail_top_n > 0:
            from sqlalchemy import text as _text

            from app.services.scrapers.cian_detail import (
                fetch_detail,
                save_detail_enrichment,
            )

            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 '4 hours'
                        ORDER BY scraped_at DESC NULLS LAST
                        LIMIT :lim
                    """),
                    {"lim": detail_top_n},
                )
                .mappings()
                .all()
            )
            for didx, row in enumerate(priority_rows):
                if scrape_runs.is_cancelled(db, run_id):
                    logger.info("cian-full-load run_id=%d: cancelled during detail enrich", run_id)
                    break
                elif shutdown_requested():
                    logger.info(
                        "cian-full-load run_id=%d: SIGTERM-drain — stopping detail at #%d/%d",
                        run_id,
                        didx + 1,
                        len(priority_rows),
                    )
                    break
                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
                except Exception as exc:
                    counters.detail_failed += 1
                    counters.errors_count += 1
                    logger.warning(
                        "cian-full-load run_id=%d: detail failed listing_id=%d: %s",
                        run_id,
                        listing_id,
                        exc,
                    )
                if didx < len(priority_rows) - 1:
                    await asyncio.sleep(request_delay_sec * random.uniform(0.8, 1.2))

            logger.info(
                "cian-full-load run_id=%d: detail done — enriched=%d/%d failed=%d",
                run_id,
                counters.detail_enriched,
                counters.detail_attempted,
                counters.detail_failed,
            )
            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

        scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
        logger.info(
            "cian-full-load run_id=%d done: unique=%d ins=%d upd=%d detail=%d/%d errors=%d",
            run_id,
            counters.unique_fetched,
            counters.saved_inserted,
            counters.saved_updated,
            counters.detail_enriched,
            counters.detail_attempted,
            counters.errors_count,
        )
        return counters

    except RuntimeError as exc:
        # on_bucket кидает RuntimeError("cancelled") при cooperative cancel,
        # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a).
        if str(exc) == "cancelled":
            logger.info(
                "cian-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d",
                run_id,
                counters.unique_fetched,
                counters.saved_inserted,
                counters.saved_updated,
            )
            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
            return counters
        if str(exc) == "shutdown":
            logger.info(
                "cian-full-load run_id=%d: SIGTERM-drain — partial results unique=%d ins=%d upd=%d",
                run_id,
                counters.unique_fetched,
                counters.saved_inserted,
                counters.saved_updated,
            )
            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
            return counters
        logger.exception("cian-full-load run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise

    except Exception as exc:
        logger.exception("cian-full-load run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise

    finally:
        # #1949: отменяем фоновый heartbeat при любом завершении (return / raise / cancel).
        if _hb_task is not None:
            _hb_task.cancel()
            try:
                await _hb_task
            except asyncio.CancelledError:
                pass


# ── Yandex exhaustive full load (region-wide, без anchor'ов) ──────────────────


@dataclass
class YandexFullLoadCounters:
    """Aggregate counters для run_yandex_full_load.

    Без detail_* полей — Yandex full load не делает detail-обогащение.
    """

    unique_fetched: int = 0
    saved_inserted: int = 0
    saved_updated: int = 0
    price_history_rows: 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)}


async def run_yandex_full_load(
    db: Session,
    *,
    run_id: int,
    price_cap_per_bucket: int = 500,
    request_delay_sec: float = 2.0,
    concurrency: int = 4,
    resume_run_id: int | None = None,
) -> YandexFullLoadCounters:
    """Exhaustive региональный сбор Yandex ЕКБ вторички (БЕЗ anchor'ов).

    Yandex SERP — path-based по городу, не geo-bbox. Anchor-loop избыточен
    (все anchor'ы дают одну и ту же городскую выдачу). Один проход с
    партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление)
    охватывает весь ЕКБ.

    Параметры:
      price_cap_per_bucket: максимум офферов на price-бакет (< Yandex SERP-cap ~575).
      request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
      concurrency: параллельных page-фетчей в leaf-бакете.
      resume_run_id: если задан — читает done_buckets из counters прошлого run и
        пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного)
        yandex_full_load прогона для resume. Без этого поля — full walk с нуля.

    Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
    Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
    """
    from app.services.scrapers.yandex_realty import YandexRealtyScraper

    counters = YandexFullLoadCounters()

    skip_set: set[str] = set()
    if resume_run_id is not None:
        _prev_row = db.execute(
            text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"),
            {"rid": resume_run_id},
        ).fetchone()
        if _prev_row is not None and _prev_row.counters:
            _prev_counters: dict = (
                _prev_row.counters if isinstance(_prev_row.counters, dict) else {}
            )
            skip_set = set(_prev_counters.get("done_buckets", []))
        logger.info(
            "yandex-full-load run_id=%d: resuming from run %d — %d buckets already done",
            run_id,
            resume_run_id,
            len(skip_set),
        )
    done: set[str] = set(skip_set)

    def _on_bucket(bucket_key: str, lots: list) -> None:  # type: ignore[type-arg]
        """Инкрементальный save после каждого leaf-бакета."""
        nonlocal done
        if scrape_runs.is_cancelled(db, run_id):
            logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id)
            raise RuntimeError("cancelled")
        elif shutdown_requested():
            logger.info(
                "yandex-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s",
                run_id,
                bucket_key,
            )
            raise RuntimeError("shutdown")
        if not lots:
            done.add(bucket_key)
            scrape_runs.update_heartbeat(
                db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
            )
            return
        inserted, updated = save_listings(
            db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today
        )
        # save_listings вызывает db.commit() внутри — данные в БД сразу
        counters.saved_inserted += inserted
        counters.saved_updated += updated
        counters.unique_fetched += len(lots)
        # Price-history из gate price.previous/trend (graceful — провал истории
        # НЕ должен ронять сейв листингов/бакета).
        try:
            counters.price_history_rows += record_yandex_price_history(db, lots)
        except Exception as ph_exc:
            logger.warning(
                "yandex-full-load run_id=%d: price-history failed bucket=%s: %s",
                run_id,
                bucket_key,
                ph_exc,
            )
            try:
                db.rollback()
            except Exception:
                pass
        done.add(bucket_key)
        scrape_runs.update_heartbeat(
            db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
        )
        logger.info(
            "yandex-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d",
            run_id,
            bucket_key,
            inserted,
            updated,
            counters.unique_fetched,
        )

    def _on_progress(unique_count: int) -> None:
        """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов)."""
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

    try:
        async with YandexRealtyScraper() as scraper:
            scraper.request_delay_sec = request_delay_sec

            await scraper.fetch_all_secondary(
                price_cap_per_bucket=price_cap_per_bucket,
                concurrency=concurrency,
                on_bucket=_on_bucket,
                on_progress=_on_progress,
                skip_buckets=skip_set if skip_set else None,
            )

        logger.info(
            "yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",
            run_id,
            counters.unique_fetched,
            counters.saved_inserted,
            counters.saved_updated,
        )
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
        scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
        logger.info(
            "yandex-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d",
            run_id,
            counters.unique_fetched,
            counters.saved_inserted,
            counters.saved_updated,
            counters.errors_count,
        )
        return counters

    except RuntimeError as exc:
        # on_bucket кидает RuntimeError("cancelled") при cooperative cancel,
        # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a).
        if str(exc) == "cancelled":
            logger.info(
                "yandex-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d",
                run_id,
                counters.unique_fetched,
                counters.saved_inserted,
                counters.saved_updated,
            )
            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
            return counters
        if str(exc) == "shutdown":
            logger.info(
                "yandex-full-load run_id=%d: SIGTERM-drain — partial results "
                "unique=%d ins=%d upd=%d",
                run_id,
                counters.unique_fetched,
                counters.saved_inserted,
                counters.saved_updated,
            )
            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
            return counters
        logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise

    except Exception as exc:
        logger.exception("yandex-full-load run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise


# ── Avito exhaustive full load (citywide, без anchor'ов) ──────────────────────


@dataclass
class AvitoFullLoadCounters:
    """Aggregate counters для run_avito_full_load.

    Зеркало YandexFullLoadCounters — citywide room×price bisection без anchor/detail.
    """

    unique_fetched: int = 0
    saved_inserted: int = 0
    saved_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)}


async def run_avito_full_load(
    db: Session,
    *,
    run_id: int,
    price_cap_per_bucket: int = 1400,
    request_delay_sec: float = 7.0,
    concurrency: int = 5,
    secondary_only: bool = True,
    resume_run_id: int | None = None,
    incremental_days: int | None = None,
) -> AvitoFullLoadCounters:
    """Exhaustive ИЛИ инкрементальный региональный сбор Avito ЕКБ вторички (БЕЗ anchor'ов).

    Avito anchor/citywide упирается в SERP-cap ~5000 результатов на запрос →
    вторичка катастрофически недобрана. Решение зеркалит cian/yandex full_load:
    один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное
    деление ценового диапазона через `pmin`/`pmax`) охватывает весь ЕКБ.

    Параметры:
      price_cap_per_bucket: максимум результатов на price-бакет (< Avito SERP-cap ~5000).
      request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
      concurrency: параллельных page-фетчей в leaf-бакете.
      secondary_only: отбрасывать новостройки (listing_segment=="novostroyki") до save.
      resume_run_id: если задан — читает done_buckets из counters прошлого run и
        пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного)
        avito_full_load прогона для resume. Без поля — full walk с нуля.
      incremental_days: если задан — shallow инкрементальный обход. Считается
        since = date.today() - timedelta(days=incremental_days) и передаётся в
        fetch_all_secondary(since=since): date early-stop останавливает пагинацию
        до глубоких страниц (избегает page-36 429-банов). None → since=None →
        exhaustive bisection (полный обход, текущее поведение).

    Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
    Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
    AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned'), как в
    run_avito_city_sweep — partial results уже сохранены через on_bucket.
    """
    counters = AvitoFullLoadCounters()

    # ── Инкрементальный режим: since-cutoff для shallow обхода ────────────────
    since: date | None = None
    if incremental_days is not None:
        since = date.today() - timedelta(days=incremental_days)
        logger.info(
            "avito-full-load run_id=%d: INCREMENTAL mode — since=%s (last %d days)",
            run_id,
            since.isoformat(),
            incremental_days,
        )

    # ── Checkpoint/resume: читаем done_buckets из прошлого run ───────────────
    skip_set: set[str] = set()
    if resume_run_id is not None:
        _prev_row = db.execute(
            text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"),
            {"rid": resume_run_id},
        ).fetchone()
        if _prev_row is not None and _prev_row.counters:
            _prev_counters: dict = (
                _prev_row.counters if isinstance(_prev_row.counters, dict) else {}
            )
            skip_set = set(_prev_counters.get("done_buckets", []))
        logger.info(
            "avito-full-load run_id=%d: resuming from run %d — %d buckets already done",
            run_id,
            resume_run_id,
            len(skip_set),
        )
    done: set[str] = set(skip_set)

    def _on_bucket(bucket_key: str, lots: list) -> None:  # type: ignore[type-arg]
        """Инкрементальный save после каждого leaf-бакета."""
        nonlocal done
        if scrape_runs.is_cancelled(db, run_id):
            logger.info("avito-full-load run_id=%d: cancel detected in on_bucket", run_id)
            raise RuntimeError("cancelled")
        elif shutdown_requested():
            logger.info(
                "avito-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s",
                run_id,
                bucket_key,
            )
            raise RuntimeError("shutdown")
        if not lots:
            done.add(bucket_key)
            scrape_runs.update_heartbeat(
                db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
            )
            return
        inserted, updated = save_listings(
            db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today
        )
        # save_listings вызывает db.commit() внутри — данные в БД сразу
        counters.saved_inserted += inserted
        counters.saved_updated += updated
        counters.unique_fetched += len(lots)
        done.add(bucket_key)
        scrape_runs.update_heartbeat(
            db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
        )
        logger.info(
            "avito-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d",
            run_id,
            bucket_key,
            inserted,
            updated,
            counters.unique_fetched,
        )

    def _on_progress(unique_count: int) -> None:
        """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов)."""
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

    try:
        async with AvitoScraper() as scraper:
            scraper.request_delay_sec = request_delay_sec

            await scraper.fetch_all_secondary(
                price_cap_per_bucket=price_cap_per_bucket,
                concurrency=concurrency,
                secondary_only=secondary_only,
                on_bucket=_on_bucket,
                on_progress=_on_progress,
                skip_buckets=skip_set if skip_set else None,
                since=since,
            )

        logger.info(
            "avito-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",
            run_id,
            counters.unique_fetched,
            counters.saved_inserted,
            counters.saved_updated,
        )
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
        scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
        logger.info(
            "avito-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d",
            run_id,
            counters.unique_fetched,
            counters.saved_inserted,
            counters.saved_updated,
            counters.errors_count,
        )
        return counters

    except (AvitoBlockedError, AvitoRateLimitedError) as exc:
        # IP-бан/rate-limit: partial results уже сохранены через on_bucket.
        logger.error("avito-full-load run_id=%d: blocked/rate-limited — %s", run_id, exc)
        counters.errors_count += 1
        scrape_runs.mark_banned(
            db,
            run_id,
            f"avito full load aborted: {exc}",
            {**counters.to_dict(), "done_buckets": sorted(done)},
        )
        return counters

    except RuntimeError as exc:
        # on_bucket кидает RuntimeError("cancelled") при cooperative cancel,
        # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a).
        if str(exc) == "cancelled":
            logger.info(
                "avito-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d",
                run_id,
                counters.unique_fetched,
                counters.saved_inserted,
                counters.saved_updated,
            )
            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
            return counters
        if str(exc) == "shutdown":
            logger.info(
                "avito-full-load run_id=%d: SIGTERM-drain — partial results "
                "unique=%d ins=%d upd=%d",
                run_id,
                counters.unique_fetched,
                counters.saved_inserted,
                counters.saved_updated,
            )
            scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
            return counters
        logger.exception("avito-full-load run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise

    except Exception as exc:
        logger.exception("avito-full-load run_id=%d: fatal error", run_id)
        scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
        raise


# ── DomClick citywide sweep (без anchor'ов) ────────────────────────────────


@dataclass
class DomClickCitySweepCounters:
    """Aggregate counters для DomClick citywide sweep run.

    DomClick BFF — citywide (city_id), не geo-bbox: anchor-loop отсутствует.
    Один проход fetch_city перебирает ROOM_BUCKETS × pages внутри scraper'а.
    Поля симметричны sibling-counters, но без detail_*/houses_* (SERP-only)
    и без anchors_* (нет anchor-фазы). pages_fetched = 6 buckets × pages worst-case.
    """

    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: int = 0
    pages_fetched: int = 0
    errors_count: int = 0
    blocked: int = 0  # 1 если QRATOR-блок был во время sweep
    geo_filtered: int = 0  # число офферов отфильтрованных geo-guard

    def to_dict(self) -> dict[str, int]:
        return {f.name: getattr(self, f.name) for f in fields(self)}


# Дефолтные параметры sweep'а (EKB city_id=4).
DOMCLICK_DEFAULT_CITY_ID: int = 4
DOMCLICK_DEFAULT_ROOMS: list[int] = [0, 1, 2, 3, 4]  # vestigial; scraper sweeps all buckets
# Число BFF-бакетов (st/1/2/3/4/5+) — фиксировано; используется для watchdog.
_DOMCLICK_NUM_BUCKETS: int = 6
# Оценка времени одного fetch'а (network + parse) для watchdog.
_DOMCLICK_PER_FETCH_S: float = 12.0
# Буфер сверху расчётного бюджета (cold browser start, save-фаза, price-splits).
_DOMCLICK_SWEEP_BUDGET_S: float = 300.0


async def run_domclick_city_sweep(
    db: Session,
    *,
    run_id: int,
    city_id: int = DOMCLICK_DEFAULT_CITY_ID,
    rooms: list[int] | None = None,
    pages: int = 100,
    request_delay_sec: float | None = None,
) -> DomClickCitySweepCounters:
    """DomClick citywide sweep через BFF JSON API.

    Структурно зеркалит run_cian_city_sweep / run_yandex_city_sweep, но CITYWIDE:
    DomClick не поддерживает geo-radius (fetch_around → NotImplementedError),
    поэтому anchor-loop отсутствует. DomClickScraper.fetch_city перебирает все
    ROOM_BUCKETS (st/1/2/3/4/5+) через BrowserFetcher → shared mobile proxy →
    обходит QRATOR.

    Фазы (единственный citywide проход):
      1. SERP: DomClickScraper().fetch_city(city_id, rooms, pages).
      2. SAVE: save_listings(db, lots, run_id=run_id) — house-match по lat/lon и
         address-fingerprint (BFF отдаёт координаты).

    Watchdog: весь fetch_city оборачивается в asyncio.wait_for. Таймаут считается
    из 6 buckets × pages × per-fetch + buffer (минимум ANCHOR_TIMEOUT_SEC).
    Реальный sweep ≈ 330 страниц (rooms=2 требует price-split).

    Cooperative cancel: is_cancelled(db, run_id) проверяется перед SERP-фазой.

    ЧЕСТНЫЙ СТАТУС (#1968): если scraper сообщил QRATOR-блок И lots == 0 →
    mark_failed с объяснением. Иначе mark_done (в т.ч. при lots==0 без блока:
    genuinely empty run).

    Возвращает DomClickCitySweepCounters.
    """
    from app.services.scrapers.domclick import ROOM_BUCKETS, DomClickScraper

    _resolved_delay = request_delay_sec if request_delay_sec is not None else 6.0
    counters = DomClickCitySweepCounters()

    # Watchdog: 6 buckets × pages × per_fetch + budget.
    # Реальный worst-case (rooms=2 price-split) ≈ 330 страниц × (delay+12s).
    # pages=100 → 600 fetch'ей × ~18s + 300 ≈ 11100s (~3ч) — намеренно generous.
    _num_fetches = _DOMCLICK_NUM_BUCKETS * max(1, pages)
    _sweep_timeout = max(
        ANCHOR_TIMEOUT_SEC,
        int(_num_fetches * (_resolved_delay + _DOMCLICK_PER_FETCH_S) + _DOMCLICK_SWEEP_BUDGET_S),
    )

    # Мутируемый контейнер для захвата scraper-ссылки из замыкания.
    _scraper_ref: list[DomClickScraper] = []

    try:
        # ── Cooperative cancel перед SERP-фазой ──────────────────────────────
        if scrape_runs.is_cancelled(db, run_id):
            logger.info("domclick-sweep run_id=%d: cancelled before SERP", run_id)
            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
            return counters
        elif shutdown_requested():
            # SIGTERM-drain до SERP-фазы: финализируем run как mark_done(partial),
            # минуя honest-status (это чистый drain, не QRATOR-блок).
            logger.info("domclick-sweep run_id=%d: SIGTERM-drain — stopping before SERP", run_id)
            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
            scrape_runs.mark_done(db, run_id, counters.to_dict())
            return counters

        logger.info(
            "domclick-sweep run_id=%d: BFF citywide sweep city_id=%d "
            "buckets=%d pages_cap=%d (watchdog %ds)",
            run_id,
            city_id,
            len(ROOM_BUCKETS),
            pages,
            _sweep_timeout,
        )

        lots: list[ScrapedLot] = []

        async def _domclick_phase() -> None:
            """Единственная citywide-фаза: fetch_city + save."""
            nonlocal lots
            async with DomClickScraper() as _scraper:
                _scraper_ref.append(_scraper)
                if request_delay_sec is not None:
                    _scraper.request_delay_sec = _resolved_delay
                lots = await _scraper.fetch_city(city_id=city_id, rooms=rooms, pages=pages)
            counters.lots_fetched += len(lots)
            if lots:
                inserted, updated = save_listings(db, lots, run_id=run_id)
                counters.lots_inserted += inserted
                counters.lots_updated += updated

        try:
            await asyncio.wait_for(_domclick_phase(), timeout=_sweep_timeout)
        except TimeoutError:
            logger.warning(
                "domclick-sweep run_id=%d: SERP phase timed out after %ds — partial results",
                run_id,
                _sweep_timeout,
            )
            counters.errors_count += 1
        except Exception:
            logger.exception("domclick-sweep run_id=%d: SERP phase failed", run_id)
            counters.errors_count += 1

        # Перенести счётчики scraper'а в counters (если scraper был создан).
        if _scraper_ref:
            _s = _scraper_ref[0]
            counters.blocked = 1 if _s.blocked else 0
            counters.geo_filtered = _s.geo_filtered
            # Не-block fetch-ошибки скрейпера учитываем в errors_count: они должны
            # влиять на honest-status (0 lots + fetch_errors → НЕ ложный mark_done).
            counters.errors_count += _s.fetch_errors

        # pages_fetched: worst-case число страниц (buckets × pages cap).
        counters.pages_fetched = _num_fetches
        scrape_runs.update_heartbeat(db, run_id, counters.to_dict())

        # ── ЧЕСТНЫЙ СТАТУС (#1968) ────────────────────────────────────────────
        # 0 lots + (block ИЛИ любая ошибка) → failed: run без результата по
        # внешней причине не должен маскироваться как 'done'. Нераспознанный
        # block-вариант (нет литерал-маркера) приходит как fetch_error, не blocked —
        # поэтому errors_count тоже триггерит failed. Genuinely empty 0-error run
        # → done. Partial run с lots>0 остаётся done даже если часть бакетов упала.
        if counters.lots_fetched == 0 and (counters.blocked or counters.errors_count > 0):
            logger.error(
                "domclick-sweep run_id=%d: 0 listings with blocked=%d errors=%d "
                "— marking failed",
                run_id,
                counters.blocked,
                counters.errors_count,
            )
            scrape_runs.mark_failed(
                db,
                run_id,
                "QRATOR block or fetch errors — 0 listings",
                counters.to_dict(),
            )
        else:
            scrape_runs.mark_done(db, run_id, counters.to_dict())

        logger.info(
            "domclick-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) "
            "pages=%d errors=%d blocked=%d geo_filtered=%d",
            run_id,
            counters.lots_fetched,
            counters.lots_inserted,
            counters.lots_updated,
            counters.pages_fetched,
            counters.errors_count,
            counters.blocked,
            counters.geo_filtered,
        )
        return counters

    except Exception as exc:
        logger.exception("domclick-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__ = [
    "ANCHOR_TIMEOUT_SEC",
    "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES",
    "EKB_ANCHORS",
    "CianCitySweepCounters",
    "CianFullLoadCounters",
    "CitySweepCounters",
    "DomClickCitySweepCounters",
    "NewbuildingSweepCounters",
    "PipelineCounters",
    "PipelineResult",
    "YandexCitySweepCounters",
    "YandexFullLoadCounters",
    "run_avito_city_sweep",
    "run_avito_newbuilding_sweep",
    "run_avito_pipeline",
    "run_cian_city_sweep",
    "run_cian_full_load",
    "run_domclick_city_sweep",
    "run_yandex_city_sweep",
    "run_yandex_full_load",
]