"""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 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.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

logger = logging.getLogger(__name__)

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

# Константы для расчёта 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

# 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 propagate наружу.
    """
    counters = PipelineCounters()
    lots: list[ScrapedLot] = []
    detail_delay = request_delay_sec if request_delay_sec is not None else 7.0

    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()
            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:
            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
                    hid = house_counters.get("house_id")
                    if hid:
                        touched_house_ids.add(int(hid))
                except (AvitoBlockedError, AvitoRateLimitedError):
                    logger.error("pipeline:house BLOCKED at %s — propagating", house_path)
                    raise
                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:
            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()
            )

            consecutive_blocks = 0
            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)
                    consecutive_blocks = 0
                except (AvitoBlockedError, AvitoRateLimitedError) as e:
                    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),
                        consecutive_blocks,
                        e,
                    )
                    if consecutive_blocks >= 3:
                        logger.error(
                            "pipeline:detail ABORT — %d consecutive blocks, IP rate-limited",
                            consecutive_blocks,
                        )
                        counters.errors.append(
                            f"AVITO_RATE_LIMITED — sweep aborted after "
                            f"{idx + 1}/{len(priority_rows)} details"
                        )
                        raise
                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:
                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
                        hid = house_counters.get("house_id")
                        if hid:
                            touched_house_ids.add(int(hid))
                    except (AvitoBlockedError, AvitoRateLimitedError):
                        logger.error(
                            "pipeline:house(detail) BLOCKED at %s — propagating", house_path
                        )
                        raise
                    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"
    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())
        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

                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

                    # ── Phase 1+2: SERP + save ─────────────────────────────────
                    scraper = AvitoScraper()
                    if browser_mode:
                        scraper._browser = shared_bf
                    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()
                    if enrich_houses and unique_house_paths:
                        house_paths_list = list(unique_house_paths)
                        for h_idx, house_path in enumerate(house_paths_list):
                            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
                                hid = hc.get("house_id")
                                if hid:
                                    touched_house_ids.add(int(hid))
                            except (AvitoBlockedError, AvitoRateLimitedError):
                                logger.error(
                                    "city-sweep run_id=%d: house BLOCKED at %s — propagating",
                                    run_id,
                                    house_path,
                                )
                                raise
                            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
                            if h_idx < len(house_paths_list) - 1:
                                await asyncio.sleep(request_delay_sec)

                    # ── 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:
                                    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)
                            for nh_idx, house_path in enumerate(nh_list):
                                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
                                    hid = hc.get("house_id")
                                    if hid:
                                        touched_house_ids.add(int(hid))
                                except (AvitoBlockedError, AvitoRateLimitedError):
                                    logger.error(
                                        "city-sweep run_id=%d: house(detail) BLOCKED %s"
                                        " — propagating",
                                        run_id,
                                        house_path,
                                    )
                                    raise
                                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
                                if nh_idx < len(nh_list) - 1:
                                    await asyncio.sleep(request_delay_sec)

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

                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


# ── 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> → полный адрес с номером дома).
    """

    anchors_total: int = 0
    anchors_done: int = 0
    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: 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,
) -> 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

    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

    # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None).
    # В combos-режиме один "anchor" выполняет num_combos × max_pages fetch'ей —
    # каждый занимает примерно _resolved_delay + _YANDEX_COMBOS_PER_FETCH_S секунд.
    # Для 30 combos × 3 pages × (9+12) + 300 ≈ 2190s (~37 мин).
    # В 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,
            _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

            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

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

                # ── Phase 1+2: SERP + save ─────────────────────────────────
                async with YandexRealtyScraper() as scraper:
                    anchor_lots = await scraper.fetch_around_multi_room(
                        _a_lat,
                        _a_lon,
                        radius_m,
                        max_pages=pages_per_anchor,
                        rooms_list=_rooms_list,
                        price_ranges=_price_ranges,
                    )
                counters.lots_fetched += len(anchor_lots)
                if anchor_lots:
                    inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
                    counters.lots_inserted += inserted
                    counters.lots_updated += updated
                consecutive_failures = 0
                logger.info(
                    "yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d",
                    run_id,
                    _a_name,
                    len(anchor_lots),
                    counters.lots_inserted,
                    counters.lots_updated,
                )

                # ── Phase 3: address-enrich ────────────────────────────────
                if not (enrich_address and anchor_lots):
                    return
                source_ids = [lot.source_id for lot in anchor_lots if lot.source_id]
                if not source_ids:
                    return
                id_rows = (
                    db.execute(
                        _text("""
                            SELECT id, source_id, address, source_url
                            FROM listings
                            WHERE source = 'yandex'
                              AND source_id = ANY(CAST(:ids AS text[]))
                              AND address IS NOT NULL
                              AND NOT (address ~ ',\\s*\\d+')
                              AND source_url IS NOT NULL
                        """),
                        {"ids": source_ids},
                    )
                    .mappings()
                    .all()
                )
                items = [dict(r) for r in id_rows]
                counters.address_attempted += len(items)

                if items:
                    from curl_cffi.requests import AsyncSession as _AsyncSession

                    async with _AsyncSession(
                        impersonate="chrome120",
                        timeout=30.0,
                        proxies=_proxies,
                        headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"},
                    ) as enrich_session:
                        for eidx, item in enumerate(items):
                            lid: int = item["id"]
                            old_addr: str = item["address"]
                            url: str = item["source_url"]
                            try:
                                resp = await enrich_session.get(url, allow_redirects=True)
                                if resp.status_code != 200:
                                    logger.warning(
                                        "yandex-sweep run_id=%d address-enrich:"
                                        " HTTP %d listing_id=%d",
                                        run_id,
                                        resp.status_code,
                                        lid,
                                    )
                                    counters.address_failed += 1
                                else:
                                    new_addr = _extract_address_from_title(resp.text)
                                    if (
                                        new_addr is None
                                        or new_addr == old_addr
                                        or not _RE_HAS_HOUSE_NUMBER.search(new_addr)
                                    ):
                                        pass
                                    else:
                                        try:
                                            db.execute(
                                                _text("""
                                                    UPDATE listings
                                                    SET address = :addr,
                                                        geocode_tried_at = NULL
                                                    WHERE id = CAST(:id AS bigint)
                                                """),
                                                {"addr": new_addr, "id": lid},
                                            )
                                            db.commit()
                                            counters.address_enriched += 1
                                            logger.info(
                                                "yandex-sweep run_id=%d: "
                                                "address updated listing_id=%d "
                                                "%r -> %r",
                                                run_id,
                                                lid,
                                                old_addr,
                                                new_addr,
                                            )
                                        except Exception as upd_exc:
                                            counters.address_failed += 1
                                            counters.errors_count += 1
                                            logger.warning(
                                                "yandex-sweep run_id=%d: "
                                                "address DB update failed "
                                                "listing_id=%d: %s",
                                                run_id,
                                                lid,
                                                upd_exc,
                                            )
                                            try:
                                                db.rollback()
                                            except Exception:
                                                pass
                            except Exception as fetch_exc:
                                counters.address_failed += 1
                                logger.warning(
                                    "yandex-sweep run_id=%d: address-enrich "
                                    "fetch failed listing_id=%d: %s",
                                    run_id,
                                    lid,
                                    fetch_exc,
                                )
                            if eidx < len(items) - 1:
                                await asyncio.sleep(enrich_delay)

                    logger.info(
                        "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d",
                        run_id,
                        _a_name,
                        counters.address_enriched,
                        counters.address_attempted,
                        counters.address_failed,
                    )

            try:
                await asyncio.wait_for(_yandex_anchor_phases(), timeout=_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_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,
) -> CianCitySweepCounters:
    """Cian full city sweep: SERP → detail(+price-history) → newbuilding/houses.

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

    Фазы per anchor:
      1. SERP: CianScraper.fetch_around_multi_room(lat, lon, radius_m, pages=pages_per_anchor)
      2. SAVE: save_listings(db, lots, run_id=run_id)
      3. 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)
      4. 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

    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

            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
                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)
                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 ins=%d upd=%d",
                    run_id,
                    _a_name,
                    len(anchor_lots),
                    counters.lots_inserted,
                    counters.lots_updated,
                )

                # ── Phase 3: detail enrichment (top-N) ────────────────────
                if detail_top_n > 0:
                    priority_rows = (
                        db.execute(
                            _text("""
                                SELECT id, source_url
                                FROM listings
                                WHERE source = 'cian'
                                  AND source_url IS NOT NULL
                                  AND detail_enriched_at IS NULL
                                  AND scraped_at > NOW() - INTERVAL '2 hours'
                                ORDER BY scraped_at DESC NULLS LAST
                                LIMIT :lim
                            """),
                            {"lim": detail_top_n},
                        )
                        .mappings()
                        .all()
                    )
                    for didx, row in enumerate(priority_rows):
                        listing_id: int = row["id"]
                        source_url: str = row["source_url"]
                        counters.detail_attempted += 1
                        try:
                            enrichment = await fetch_detail(source_url)
                            if enrichment is not None:
                                save_detail_enrichment(db, listing_id, enrichment)
                                counters.detail_enriched += 1
                            else:
                                counters.detail_failed += 1
                                logger.warning(
                                    "cian-sweep run_id=%d: detail returned None for "
                                    "listing_id=%d url=%s",
                                    run_id,
                                    listing_id,
                                    source_url,
                                )
                        except Exception as exc:
                            counters.detail_failed += 1
                            counters.errors_count += 1
                            logger.warning(
                                "cian-sweep run_id=%d: detail failed listing_id=%d: %s",
                                run_id,
                                listing_id,
                                exc,
                            )
                        if didx < len(priority_rows) - 1:
                            jitter = random.uniform(0.8, 1.2)
                            await asyncio.sleep(request_delay_sec * jitter)
                    logger.info(
                        "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d",
                        run_id,
                        _a_name,
                        counters.detail_enriched,
                        counters.detail_attempted,
                        counters.detail_failed,
                    )

                # ── Phase 4: newbuilding/houses enrichment ─────────────────
                if not (enrich_houses and anchor_lots):
                    return
                cian_nb_ext_ids = {
                    lot.house_ext_id
                    for lot in anchor_lots
                    if lot.house_source == "cian_newbuilding" and lot.house_ext_id
                }
                if not cian_nb_ext_ids:
                    return
                nb_id_list = list(cian_nb_ext_ids)
                try:
                    house_rows = (
                        db.execute(
                            _text("""
                                SELECT h.id, h.cian_zhk_url
                                FROM houses h
                                JOIN house_sources hs ON hs.house_id = h.id
                                WHERE hs.ext_source = 'cian_newbuilding'
                                  AND hs.ext_id = ANY(CAST(:ids AS text[]))
                                  AND h.cian_zhk_url IS NOT NULL
                            """),
                            {"ids": nb_id_list},
                        )
                        .mappings()
                        .all()
                    )
                except Exception as exc:
                    logger.exception(
                        "cian-sweep run_id=%d anchor %s: houses DB query failed — "
                        "skipping houses phase (%d ids): %s",
                        run_id,
                        _a_name,
                        len(nb_id_list),
                        exc,
                    )
                    counters.houses_failed += len(nb_id_list)
                    try:
                        db.rollback()
                    except Exception:
                        pass
                    return
                for hidx, hrow in enumerate(house_rows):
                    house_id_val: int = hrow["id"]
                    zhk_url: str = hrow["cian_zhk_url"]
                    counters.houses_attempted += 1
                    try:
                        nb_enrichment = await fetch_newbuilding(zhk_url)
                        if nb_enrichment is not None:
                            save_newbuilding_enrichment(db, house_id_val, nb_enrichment)
                            counters.houses_enriched += 1
                        else:
                            counters.houses_failed += 1
                            logger.warning(
                                "cian-sweep run_id=%d: newbuilding returned None "
                                "house_id=%d url=%s",
                                run_id,
                                house_id_val,
                                zhk_url,
                            )
                    except Exception as exc:
                        counters.houses_failed += 1
                        counters.errors_count += 1
                        logger.warning(
                            "cian-sweep run_id=%d: houses failed house_id=%d: %s",
                            run_id,
                            house_id_val,
                            exc,
                        )
                        try:
                            db.rollback()
                        except Exception:
                            pass
                    if hidx < len(house_rows) - 1:
                        await asyncio.sleep(request_delay_sec)
                logger.info(
                    "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d",
                    run_id,
                    _a_name,
                    counters.houses_enriched,
                    counters.houses_attempted,
                    counters.houses_failed,
                )

            try:
                await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC)
            except TimeoutError:
                logger.warning(
                    "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) "
                    "timed out after %ds — skipping",
                    run_id,
                    idx,
                    len(_anchors),
                    name,
                    lat,
                    lon,
                    ANCHOR_TIMEOUT_SEC,
                )
                counters.errors_count += 1
                consecutive_failures += 1
                if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES:
                    logger.error(
                        "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
                        "%d consecutive failures (last: timeout): "
                        "IP likely blocked or proxy hung",
                        run_id,
                        idx,
                        len(_anchors),
                        name,
                        consecutive_failures,
                    )
                    counters.anchors_done = idx
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                    scrape_runs.mark_banned(
                        db,
                        run_id,
                        f"cian sweep aborted: {consecutive_failures} consecutive "
                        f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)",
                        counters.to_dict(),
                    )
                    return counters
                counters.anchors_done = idx
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                continue
            except Exception as e:
                logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name)
                counters.errors_count += 1
                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")
        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)
        # 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())

    try:
        async with CianScraper() 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=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
                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
        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
        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


# ── 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
    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")
        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)
        # 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(
            "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
        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
        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


# ── Public re-exports ───────────────────────────────────────────
__all__ = [
    "ANCHOR_TIMEOUT_SEC",
    "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES",
    "EKB_ANCHORS",
    "CianCitySweepCounters",
    "CianFullLoadCounters",
    "CitySweepCounters",
    "PipelineCounters",
    "PipelineResult",
    "YandexCitySweepCounters",
    "YandexFullLoadCounters",
    "run_avito_city_sweep",
    "run_avito_pipeline",
    "run_cian_city_sweep",
    "run_cian_full_load",
    "run_yandex_city_sweep",
    "run_yandex_full_load",
]