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

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

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

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

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

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

IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron.
"""

from __future__ import annotations

import asyncio
import logging
import random
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.n1 import N1Scraper
from app.services.scrapers.yandex_realty import YandexRealtyScraper

logger = logging.getLogger(__name__)

# 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: 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).
    """
    url = settings.avito_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


# ── 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,
) -> 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 у вызывающего).
    AvitoBlockedError / AvitoRateLimitedError propagate наружу.
    """
    counters = PipelineCounters()
    lots: list[ScrapedLot] = []
    detail_delay = request_delay_sec if request_delay_sec is not None else 7.0

    own_session = shared_session is None
    session = shared_session or AsyncSession(
        impersonate="chrome120",
        timeout=25,
        headers=_CHROME_HEADERS,
        proxies=_avito_proxies(),
    )

    try:
        # ── Step 1: search ──────────────────────────────────────
        scraper = AvitoScraper()
        scraper._cffi = session
        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 ──────
        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)
                    save_house_catalog_enrichment(db, enrichment)
                    counters.houses_enriched += 1
                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}")

                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
            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)
                    if save_detail_enrichment(db, enrichment_detail):
                        counters.detail_enriched += 1
                    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)
                    consecutive_blocks = 0

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

        logger.info(
            "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) "
            "houses=%d/%d detail=%d/%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(counters.errors),
        )
        return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)

    finally:
        if own_session:
            await session.close()


@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
    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,
) -> CitySweepCounters:
    """Full city sweep: iterate anchors × pages → save → enrich houses + detail.

    - Single shared AsyncSession на весь sweep (один TLS fingerprint)
    - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
    - Прочие errors per-anchor логируются, не валят весь sweep
    """
    _anchors = anchors if anchors is not None else EKB_ANCHORS
    counters = CitySweepCounters(anchors_total=len(_anchors))

    async with AsyncSession(
        impersonate="chrome120",
        timeout=25,
        headers=_CHROME_HEADERS,
        proxies=_avito_proxies(),
    ) as session:
        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,
                )
                try:
                    result = await run_avito_pipeline(
                        db,
                        lat=lat,
                        lon=lon,
                        radius_m=radius_m,
                        enrich_houses=enrich_houses,
                        enrich_detail_top_n=detail_top_n,
                        pages=pages_per_anchor,
                        request_delay_sec=request_delay_sec,
                        shared_session=session,
                    )
                    counters.lots_fetched += result.counters.lots_fetched
                    counters.lots_inserted += result.counters.lots_inserted
                    counters.lots_updated += result.counters.lots_updated
                    counters.unique_houses += result.counters.unique_houses
                    counters.houses_enriched += result.counters.houses_enriched
                    counters.houses_failed += result.counters.houses_failed
                    counters.detail_attempted += result.counters.detail_attempted
                    counters.detail_enriched += result.counters.detail_enriched
                    counters.detail_failed += result.counters.detail_failed
                    counters.errors_count += len(result.counters.errors)
                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())

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

    Проще avito CitySweepCounters: у Yandex SERP нет house-catalog / detail-enrich
    шага (только search → save), поэтому houses/detail-полей нет.
    """

    anchors_total: int = 0
    anchors_done: int = 0
    pages_fetched: int = 0
    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: int = 0
    errors_count: int = 0

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


# Defensive abort threshold: у YandexRealtyScraper нет block/rate-exception классов
# (fetch_around глотает ошибки и возвращает []). Чтобы не молотить Yandex впустую при
# системном блоке/бане IP, прерываем sweep после N подряд anchor'ов, упавших с
# исключением. NB: пустой результат (0 lots) — это НЕ ошибка, счётчик не растёт.
YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES = 3


async def run_yandex_city_sweep(
    db: Session,
    *,
    run_id: int,
    anchors: list[tuple[float, float, str]] | None = None,
    radius_m: int = 1500,
    pages_per_anchor: int = 2,
    request_delay_sec: float | None = None,
) -> YandexCitySweepCounters:
    """Yandex.Недвижимость city sweep: iterate anchors × pages → save.

    Проще avito sweep — у Yandex SERP нет house-catalog / detail-enrich шага:
      per anchor: YandexRealtyScraper().fetch_around(lat, lon, radius_m, page=p)
      → save_listings(db, lots) → accumulate counters.

    Anti-bot / cancel семантика (зеркало avito sweep, см. #561):
      - Cooperative cancel: is_cancelled(db, run_id) проверяется перед каждым anchor;
        при cancel — update_heartbeat + return (mark_cancelled делает caller-flow по статусу).
      - YandexRealtyScraper НЕ определяет block/rate exception классов (fetch_around
        ловит исключения внутри и возвращает []). Поэтому per-anchor ловим broad
        Exception, логируем и continue; защитно прерываем sweep (early abort + mark_banned),
        если N anchor'ов ПОДРЯД упали с исключением — это сигнал системного блока IP.
      - request_delay_sec: YandexRealtyScraper.fetch_around НЕ принимает delay-override
        (он сам спит self.request_delay_sec между страницами через sleep_between_requests).
        Поэтому request_delay_sec применяется здесь как доп. asyncio.sleep МЕЖДУ anchor'ами.

    Возвращает YandexCitySweepCounters (агрегат по всем anchor'ам).
    """
    _anchors = anchors if anchors is not None else EKB_ANCHORS
    counters = YandexCitySweepCounters(anchors_total=len(_anchors))
    inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
    consecutive_failures = 0

    try:
        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
            if scrape_runs.is_cancelled(db, run_id):
                logger.info(
                    "yandex-sweep run_id=%d: cancelled at anchor #%d/%d (%s)",
                    run_id,
                    idx,
                    len(_anchors),
                    name,
                )
                scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                return counters

            logger.info(
                "yandex-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
                run_id,
                idx,
                len(_anchors),
                name,
                lat,
                lon,
            )
            try:
                # Fresh scraper per anchor — uses its own httpx.AsyncClient via the
                # async-context-manager. fetch_around fetches ONE SERP page; loop pages.
                anchor_lots: list[ScrapedLot] = []
                async with YandexRealtyScraper() as scraper:
                    for page in range(pages_per_anchor):
                        page_lots = await scraper.fetch_around(lat, lon, radius_m, page=page)
                        counters.pages_fetched += 1
                        if not page_lots:
                            # Пустая страница → дальше результатов нет, прекращаем пагинацию.
                            break
                        anchor_lots.extend(page_lots)

                counters.lots_fetched += len(anchor_lots)
                if anchor_lots:
                    inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
                    counters.lots_inserted += inserted
                    counters.lots_updated += updated
                consecutive_failures = 0
            except Exception as e:
                # YandexRealtyScraper не propagate'ит block/rate exceptions, поэтому здесь
                # ловим broad Exception per-anchor: логируем и continue (изоляция anchor'а).
                logger.exception("yandex-sweep run_id=%d: anchor %s failed", run_id, name)
                counters.errors_count += 1
                consecutive_failures += 1
                if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES:
                    logger.error(
                        "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — "
                        "%d consecutive failures (IP likely blocked): %s",
                        run_id,
                        idx,
                        len(_anchors),
                        name,
                        consecutive_failures,
                        e,
                    )
                    counters.anchors_done = idx
                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
                    scrape_runs.mark_banned(
                        db,
                        run_id,
                        f"yandex sweep aborted: {consecutive_failures} consecutive "
                        f"anchor failures (last: {e})",
                        counters.to_dict(),
                    )
                    return counters

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

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

        scrape_runs.mark_done(db, run_id, counters.to_dict())
        logger.info(
            "yandex-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d "
            "(ins=%d/upd=%d) errors=%d",
            run_id,
            counters.anchors_done,
            counters.anchors_total,
            counters.pages_fetched,
            counters.lots_fetched,
            counters.lots_inserted,
            counters.lots_updated,
            counters.errors_count,
        )
        return counters

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


# ── N1.ru city sweep (#575) ─────────────────────────────────────────


@dataclass
class N1CitySweepCounters:
    """Aggregate counters для N1.ru city sweep run.

    Как Yandex: N1 SERP — простой search → save, без house-catalog / detail-enrich
    шага, поэтому houses/detail-полей нет.
    """

    anchors_total: int = 0
    anchors_done: int = 0
    pages_fetched: int = 0
    lots_fetched: int = 0
    lots_inserted: int = 0
    lots_updated: int = 0
    errors_count: int = 0

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


# Defensive abort: N1Scraper.fetch_around глотает ошибки и возвращает [] (нет
# block/rate exception-классов). Прерываем sweep после N подряд anchor'ов,
# упавших с исключением — сигнал системного блока IP. Пустой результат (0 lots)
# — НЕ ошибка, счётчик не растёт (зеркало yandex).
N1_SWEEP_MAX_CONSECUTIVE_FAILURES = 3


async def run_n1_city_sweep(
    db: Session,
    *,
    run_id: int,
    anchors: list[tuple[float, float, str]] | None = None,
    radius_m: int = 1500,
    pages_per_anchor: int = 2,
    request_delay_sec: float | None = None,
) -> N1CitySweepCounters:
    """N1.ru city sweep: iterate anchors × pages → save (#575).

    Зеркало run_yandex_city_sweep — N1 это региональный SERP-портал ЕКБ без
    house-catalog / detail-enrich шага:
      per anchor: N1Scraper().fetch_around(lat, lon, radius_m, page=p)
      → save_listings(db, lots) → accumulate counters.

    Anti-bot / cancel семантика (зеркало yandex/avito sweep):
      - Cooperative cancel: is_cancelled(db, run_id) перед каждым anchor.
      - N1Scraper не определяет block/rate exception-классов (fetch_around ловит
        ошибки внутри → []). Per-anchor ловим broad Exception, логируем, continue;
        защитно прерываем (early abort + mark_banned) при N подряд исключениях.
      - request_delay_sec применяется как доп. asyncio.sleep МЕЖДУ anchor'ами
        (N1Scraper сам спит self.request_delay_sec между страницами).
    """
    _anchors = anchors if anchors is not None else EKB_ANCHORS
    counters = N1CitySweepCounters(anchors_total=len(_anchors))
    inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0
    consecutive_failures = 0

    try:
        for idx, (lat, lon, name) in enumerate(_anchors, start=1):
            if scrape_runs.is_cancelled(db, run_id):
                logger.info(
                    "n1-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(
                "n1-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)",
                run_id,
                idx,
                len(_anchors),
                name,
                lat,
                lon,
            )
            try:
                anchor_lots: list[ScrapedLot] = []
                async with N1Scraper() as scraper:
                    for page in range(1, pages_per_anchor + 1):
                        page_lots = await scraper.fetch_around(lat, lon, radius_m, page=page)
                        counters.pages_fetched += 1
                        if not page_lots:
                            break
                        anchor_lots.extend(page_lots)

                counters.lots_fetched += len(anchor_lots)
                if anchor_lots:
                    inserted, updated = save_listings(db, anchor_lots, run_id=run_id)
                    counters.lots_inserted += inserted
                    counters.lots_updated += updated
                consecutive_failures = 0
            except Exception as e:
                logger.exception("n1-sweep run_id=%d: anchor %s failed", run_id, name)
                counters.errors_count += 1
                consecutive_failures += 1
                if consecutive_failures >= N1_SWEEP_MAX_CONSECUTIVE_FAILURES:
                    logger.error(
                        "n1-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"n1 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())

            if idx < len(_anchors):
                await asyncio.sleep(inter_anchor_delay)

        scrape_runs.mark_done(db, run_id, counters.to_dict())
        logger.info(
            "n1-sweep run_id=%d done: anchors=%d/%d pages=%d lots=%d (ins=%d/upd=%d) errors=%d",
            run_id,
            counters.anchors_done,
            counters.anchors_total,
            counters.pages_fetched,
            counters.lots_fetched,
            counters.lots_inserted,
            counters.lots_updated,
            counters.errors_count,
        )
        return counters

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


# ── Public re-exports ───────────────────────────────────────────
__all__ = [
    "EKB_ANCHORS",
    "CitySweepCounters",
    "PipelineCounters",
    "PipelineResult",
    "YandexCitySweepCounters",
    "run_avito_city_sweep",
    "run_avito_pipeline",
    "run_yandex_city_sweep",
]