feat(tradein): cian city-sweep — один прогон SERP+detail(+price-history)+houses

run_cian_city_sweep итерирует EKB_ANCHORS, per anchor:
  SERP (fetch_around_multi_room) → save_listings → detail top-N (incl. price-history)
  → newbuilding/houses (if enrich_houses=True).
Admin endpoints: POST /scrape/cian-city-sweep, GET /runs, POST /{id}/cancel.
Без scrape_schedules seed (ручной запуск, cron-bulk = 429-бан на Cian).

tradein-mvp/backend/app/api/v1/admin.py

@@ -22,7 +22,11 @@ from app.schemas.trade_in import ScheduleConfig, ScheduleConfigUpdate
 from app.services import cian_session as cian_session_svc
 from app.services import scrape_runs as runs_mod
 from app.services.geocoder import geocode
-from app.services.scrape_pipeline import run_avito_city_sweep, run_n1_city_sweep
+from app.services.scrape_pipeline import (
+    run_avito_city_sweep,
+    run_cian_city_sweep,
+    run_n1_city_sweep,
+)
 from app.services.scraper_settings import invalidate_cache
 from app.services.scrapers.avito import AvitoScraper
 from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment
@@ -759,6 +763,95 @@ def list_n1_city_sweep_runs(
     ]
 
 
+# ── Cian city sweep (#860) — on-demand trigger + runs list ───────────────────
+
+
+class CianCitySweepStartRequest(BaseModel):
+    pages_per_anchor: int = Field(default=3, ge=1, le=10)
+    detail_top_n: int = Field(default=10, ge=0, le=30)
+    request_delay_sec: float = Field(default=5.0, ge=3.0, le=15.0)
+    enrich_houses: bool = True
+    radius_m: int = Field(default=1500, ge=500, le=5000)
+
+
+@router.post("/scrape/cian-city-sweep", response_model=CitySweepStartResponse)
+async def start_cian_city_sweep(
+    payload: CianCitySweepStartRequest,
+    background_tasks: BackgroundTasks,
+    db: Annotated[Session, Depends(get_db)],
+) -> CitySweepStartResponse:
+    """Запустить Cian ЕКБ city sweep в background (#860). Returns run_id.
+
+    Один прогон: SERP (fetch_around_multi_room) → detail-обогащение (incl.
+    price-history) → newbuilding/houses (если enrich_houses=True).
+    Прокси уже внутри CianScraper.
+
+    Coop cancel: POST /scrape/cian-city-sweep/{run_id}/cancel.
+    НЕ добавляется в scrape_schedules — ручной/контролируемый запуск
+    (bulk cron → 429-бан на Cian).
+    """
+    run_id = runs_mod.create_run(db, source="cian_city_sweep", params=payload.model_dump())
+
+    async def _sweep_task() -> None:
+        sweep_db = SessionLocal()
+        try:
+            await run_cian_city_sweep(
+                sweep_db,
+                run_id=run_id,
+                radius_m=payload.radius_m,
+                pages_per_anchor=payload.pages_per_anchor,
+                request_delay_sec=payload.request_delay_sec,
+                detail_top_n=payload.detail_top_n,
+                enrich_houses=payload.enrich_houses,
+            )
+        except Exception:
+            logger.exception("cian-sweep background task run_id=%d crashed", run_id)
+        finally:
+            sweep_db.close()
+
+    background_tasks.add_task(_sweep_task)
+    logger.info("cian-sweep queued run_id=%d params=%s", run_id, payload.model_dump())
+    return CitySweepStartResponse(
+        run_id=run_id,
+        status="running",
+        pages_per_anchor=payload.pages_per_anchor,
+        detail_top_n=payload.detail_top_n,
+    )
+
+
+@router.get("/scrape/cian-city-sweep/runs", response_model=list[ScrapeRunRow])
+def list_cian_city_sweep_runs(
+    db: Annotated[Session, Depends(get_db)],
+    limit: int = 10,
+) -> list[ScrapeRunRow]:
+    """Список последних N Cian city sweep runs (для UI polling). Default limit=10."""
+    rows = runs_mod.list_recent(db, source="cian_city_sweep", limit=limit)
+    return [
+        ScrapeRunRow(
+            run_id=r["run_id"],
+            source=r["source"],
+            status=r["status"],
+            params=r.get("params"),
+            counters=r.get("counters"),
+            error=r.get("error"),
+            started_at=r["started_at"].isoformat() if r.get("started_at") else None,
+            finished_at=r["finished_at"].isoformat() if r.get("finished_at") else None,
+            heartbeat_at=r["heartbeat_at"].isoformat() if r.get("heartbeat_at") else None,
+        )
+        for r in rows
+    ]
+
+
+@router.post("/scrape/cian-city-sweep/{run_id}/cancel")
+def cancel_cian_city_sweep(
+    run_id: int,
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, object]:
+    """Отменить running Cian sweep. Cooperative: проверяется каждый anchor."""
+    cancelled = runs_mod.mark_cancelled(db, run_id)
+    return {"ok": True, "run_id": run_id, "cancelled": cancelled}
+
+
 # ── Scraper settings: global + per-source delay management ───────────────────
 
 

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

@@ -490,8 +490,7 @@ async def run_avito_city_sweep(
                     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",
+                        "city-sweep run_id=%d: IMV phase done — attempted=%d enriched=%d failed=%d",
                         run_id,
                         imv_result.checked,
                         imv_result.saved,
@@ -673,8 +672,7 @@ async def run_yandex_city_sweep(
 
         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",
+            "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,
@@ -841,14 +839,333 @@ async def run_n1_city_sweep(
         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,
+            )
+
+            # ── Phase 1+2: SERP + save ──────────────────────────────────────
+            anchor_lots: list[ScrapedLot] = []
+            try:
+                async with CianScraper() as scraper:
+                    anchor_lots = await scraper.fetch_around_multi_room(
+                        lat,
+                        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,
+                    name,
+                    len(anchor_lots),
+                    counters.lots_inserted,
+                    counters.lots_updated,
+                )
+            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
+
+            # ── Phase 3: detail enrichment (top-N) ─────────────────────────
+            if detail_top_n > 0:
+                from sqlalchemy import text as _text
+
+                priority_rows = (
+                    db.execute(
+                        _text("""
+                            SELECT id, source_url
+                            FROM listings
+                            WHERE source = 'cian'
+                              AND source_url IS NOT NULL
+                              AND detail_enriched_at IS NULL
+                              AND scraped_at > NOW() - INTERVAL '2 hours'
+                            ORDER BY scraped_at DESC NULLS LAST
+                            LIMIT :lim
+                        """),
+                        {"lim": detail_top_n},
+                    )
+                    .mappings()
+                    .all()
+                )
+                for didx, row in enumerate(priority_rows):
+                    listing_id: int = row["id"]
+                    source_url: str = row["source_url"]
+                    counters.detail_attempted += 1
+                    try:
+                        enrichment = await fetch_detail(source_url)
+                        if enrichment is not None:
+                            save_detail_enrichment(db, listing_id, enrichment)
+                            counters.detail_enriched += 1
+                        else:
+                            counters.detail_failed += 1
+                            logger.warning(
+                                "cian-sweep run_id=%d: detail returned None for "
+                                "listing_id=%d url=%s",
+                                run_id,
+                                listing_id,
+                                source_url,
+                            )
+                    except Exception as exc:
+                        counters.detail_failed += 1
+                        counters.errors_count += 1
+                        logger.warning(
+                            "cian-sweep run_id=%d: detail failed listing_id=%d: %s",
+                            run_id,
+                            listing_id,
+                            exc,
+                        )
+                    if didx < len(priority_rows) - 1:
+                        jitter = random.uniform(0.8, 1.2)
+                        await asyncio.sleep(request_delay_sec * jitter)
+                logger.info(
+                    "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d",
+                    run_id,
+                    name,
+                    counters.detail_enriched,
+                    counters.detail_attempted,
+                    counters.detail_failed,
+                )
+
+            # ── Phase 4: newbuilding/houses enrichment ─────────────────────
+            if enrich_houses and anchor_lots:
+                from sqlalchemy import text as _text2
+
+                # Собираем cian_internal_house_ids из SERP-объявлений этого anchor'а.
+                # lot.raw_payload["newbuilding_id"] → SERP предоставляет house_ext_id.
+                # Используем houses.cian_zhk_url для тех домов, у которых она заполнена.
+                # Fallback: дома у которых cian_internal_house_id совпадает с house_ext_id.
+                cian_nb_ext_ids = {
+                    lot.house_ext_id
+                    for lot in anchor_lots
+                    if lot.house_source == "cian_newbuilding" and lot.house_ext_id
+                }
+                if cian_nb_ext_ids:
+                    nb_id_list = list(cian_nb_ext_ids)
+                    house_rows = (
+                        db.execute(
+                            _text2("""
+                                SELECT h.id, h.cian_zhk_url
+                                FROM houses h
+                                WHERE h.ext_source = 'cian_newbuilding'
+                                  AND h.ext_id = ANY(CAST(:ids AS text[]))
+                                  AND h.cian_zhk_url IS NOT NULL
+                            """),
+                            {"ids": nb_id_list},
+                        )
+                        .mappings()
+                        .all()
+                    )
+                    for hidx, hrow in enumerate(house_rows):
+                        house_id_val: int = hrow["id"]
+                        zhk_url: str = hrow["cian_zhk_url"]
+                        counters.houses_attempted += 1
+                        try:
+                            nb_enrichment = await fetch_newbuilding(zhk_url)
+                            if nb_enrichment is not None:
+                                save_newbuilding_enrichment(db, house_id_val, nb_enrichment)
+                                counters.houses_enriched += 1
+                            else:
+                                counters.houses_failed += 1
+                                logger.warning(
+                                    "cian-sweep run_id=%d: newbuilding returned None "
+                                    "house_id=%d url=%s",
+                                    run_id,
+                                    house_id_val,
+                                    zhk_url,
+                                )
+                        except Exception as exc:
+                            counters.houses_failed += 1
+                            counters.errors_count += 1
+                            logger.warning(
+                                "cian-sweep run_id=%d: houses failed house_id=%d: %s",
+                                run_id,
+                                house_id_val,
+                                exc,
+                            )
+                            try:
+                                db.rollback()
+                            except Exception:
+                                pass
+                        if hidx < len(house_rows) - 1:
+                            await asyncio.sleep(request_delay_sec)
+                    logger.info(
+                        "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d",
+                        run_id,
+                        name,
+                        counters.houses_enriched,
+                        counters.houses_attempted,
+                        counters.houses_failed,
+                    )
+
+            counters.anchors_done = idx
+            scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+            # Пауза между 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
+
+
 # ── Public re-exports ───────────────────────────────────────────
 __all__ = [
+    "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES",
     "EKB_ANCHORS",
+    "CianCitySweepCounters",
     "CitySweepCounters",
     "PipelineCounters",
     "PipelineResult",
     "YandexCitySweepCounters",
     "run_avito_city_sweep",
     "run_avito_pipeline",
+    "run_cian_city_sweep",
     "run_yandex_city_sweep",
 ]

tradein-mvp/backend/tests/test_cian_city_sweep.py

@@ -0,0 +1,574 @@
+"""Offline unit tests for run_cian_city_sweep (#860).
+
+Pure & fast: NO live network, NO DB. Mirrors test_n1_city_sweep / test_city_sweep.
+Asserts orchestration semantics only.
+
+Coverage:
+  - Все фазы (SERP → save → detail → houses) вызываются в нужном порядке
+  - CianCitySweepCounters корректно заполняются
+  - Cooperative cancel прерывает sweep перед anchor'ом
+  - enrich_houses=False пропускает houses-фазу
+  - Consecutive SERP failures → mark_banned + abort
+  - Per-item detail/house error не валит весь sweep
+  - mark_done вызывается при нормальном завершении
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from app.services import scrape_pipeline
+from app.services.scrapers.base import ScrapedLot
+from app.services.scrapers.cian import CianScraper
+
+# Три тестовых anchor'а — не все 5, чтобы тесты были быстрее
+TEST_ANCHORS: list[tuple[float, float, str]] = [
+    (56.8400, 60.6050, "A1"),
+    (56.7950, 60.5300, "A2"),
+    (56.8970, 60.6100, "A3"),
+]
+
+
+def _fake_lot(offer_id: str, *, nb: bool = False) -> ScrapedLot:
+    """Создаём ScrapedLot с опциональным newbuilding-ссылкой."""
+    return ScrapedLot(
+        source="cian",
+        source_url=f"https://ekb.cian.ru/sale/flat/{offer_id}/",
+        source_id=offer_id,
+        address="Екатеринбург, улица Тестовая, 1",
+        price_rub=5_000_000,
+        area_m2=50.0,
+        rooms=2,
+        house_source="cian_newbuilding" if nb else "cian",
+        house_ext_id="99999" if nb else None,
+    )
+
+
+class _FakeRuns:
+    """Stub для scrape_runs: отслеживает вызовы, поддерживает cooperative cancel."""
+
+    def __init__(self, *, cancel_at_anchor: int | None = None) -> None:
+        self.cancel_at_anchor = cancel_at_anchor
+        self._is_cancelled_calls = 0
+        self.heartbeats: list[dict[str, int]] = []
+        self.done: dict[str, int] | None = None
+        self.failed: tuple[str, dict[str, int]] | None = None
+        self.banned: tuple[str, dict[str, int]] | None = None
+
+    def is_cancelled(self, _db: Any, _run_id: int) -> bool:
+        self._is_cancelled_calls += 1
+        if self.cancel_at_anchor is None:
+            return False
+        return self._is_cancelled_calls >= self.cancel_at_anchor
+
+    def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.heartbeats.append(dict(counters))
+
+    def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None:
+        self.done = dict(counters)
+
+    def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.failed = (error, dict(counters))
+
+    def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None:
+        self.banned = (error, dict(counters))
+
+
+@pytest.fixture(autouse=True)
+def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
+    """asyncio.sleep → instant (no test delays)."""
+
+    async def _instant(_secs: float) -> None:
+        return None
+
+    monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant)
+
+
+@pytest.fixture(autouse=True)
+def _stub_cian_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Нейтрализуем CianScraper.__init__/__aenter__/__aexit__ — нет DB вызовов."""
+
+    def _init(self: CianScraper) -> None:
+        self.request_delay_sec = 0.0
+        self._cffi = None
+
+    async def _aenter(self: CianScraper) -> CianScraper:
+        return self
+
+    async def _aexit(self: CianScraper, *_args: Any) -> None:
+        return None
+
+    monkeypatch.setattr(CianScraper, "__init__", _init)
+    monkeypatch.setattr(CianScraper, "__aenter__", _aenter)
+    monkeypatch.setattr(CianScraper, "__aexit__", _aexit)
+
+
+def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None:
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed)
+    monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned)
+
+
+# ── CianCitySweepCounters ───────────────────────────────────────────────────
+
+
+def test_cian_sweep_counters_defaults() -> None:
+    from app.services.scrape_pipeline import CianCitySweepCounters
+
+    c = CianCitySweepCounters()
+    assert c.anchors_total == 0
+    assert c.lots_fetched == 0
+    assert c.detail_attempted == 0
+    assert c.houses_attempted == 0
+    assert c.errors_count == 0
+
+
+def test_cian_sweep_counters_to_dict_all_keys() -> None:
+    from dataclasses import fields
+
+    from app.services.scrape_pipeline import CianCitySweepCounters
+
+    c = CianCitySweepCounters()
+    d = c.to_dict()
+    expected = {f.name for f in fields(c)}
+    assert set(d.keys()) == expected
+
+
+# ── Full sweep — фазы и агрегация ──────────────────────────────────────────
+
+
+async def test_sweep_iterates_all_anchors_serp_and_save(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """SERP вызывается для каждого anchor, save_listings получает накопленные lots."""
+    serp_calls: list[str] = []
+    save_calls: list[int] = []
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        serp_calls.append(f"{lat:.4f}")
+        return [_fake_lot(f"{lat}-{i}") for i in range(3)]
+
+    def fake_save(_db: Any, lots: list, *, run_id: int | None = None):
+        save_calls.append(len(lots))
+        return len(lots), 0
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
+
+    # Stub detail + houses fetchers (возвращают None → graceful skip)
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        AsyncMock(return_value=None),
+    )
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=MagicMock(),
+        run_id=1,
+        anchors=TEST_ANCHORS,
+        pages_per_anchor=2,
+        detail_top_n=0,  # skip detail phase
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert len(serp_calls) == len(TEST_ANCHORS)
+    assert len(save_calls) == len(TEST_ANCHORS)
+    assert counters.anchors_done == len(TEST_ANCHORS)
+    assert counters.lots_fetched == len(TEST_ANCHORS) * 3
+    assert counters.lots_inserted == len(TEST_ANCHORS) * 3
+    assert counters.errors_count == 0
+    assert fake.done is not None
+    assert fake.done["lots_fetched"] == len(TEST_ANCHORS) * 3
+    assert fake.failed is None
+    assert fake.banned is None
+
+
+async def test_detail_phase_runs_and_fills_counters(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """detail-фаза вызывается и счётчики заполняются."""
+    from app.services.scrapers.cian_detail import DetailEnrichment
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        return [_fake_lot(f"{lat}-{i}") for i in range(2)]
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+
+    # DB mock returning 2 rows for detail query
+    mock_db = MagicMock()
+    mock_db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 101, "source_url": "https://ekb.cian.ru/sale/flat/101/"},
+        {"id": 102, "source_url": "https://ekb.cian.ru/sale/flat/102/"},
+    ]
+
+    fake_enrichment = DetailEnrichment(cian_id=101)
+    mock_fetch_detail = AsyncMock(return_value=fake_enrichment)
+    mock_save_detail = MagicMock()
+
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        mock_fetch_detail,
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.save_detail_enrichment",
+        mock_save_detail,
+    )
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=mock_db,
+        run_id=2,
+        anchors=TEST_ANCHORS[:1],  # один anchor
+        detail_top_n=2,
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert counters.detail_attempted == 2
+    assert counters.detail_enriched == 2
+    assert counters.detail_failed == 0
+    assert mock_fetch_detail.call_count == 2
+    assert mock_save_detail.call_count == 2
+    assert fake.done is not None
+
+
+async def test_enrich_houses_false_skips_houses_phase(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """enrich_houses=False → houses-фаза не выполняется."""
+    nb_fetch_calls: list[str] = []
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        return [_fake_lot("1001", nb=True)]
+
+    async def fake_fetch_nb(url: str, **_: Any) -> None:
+        nb_fetch_calls.append(url)
+        return None
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_newbuilding.fetch_newbuilding",
+        fake_fetch_nb,
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        AsyncMock(return_value=None),
+    )
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=MagicMock(),
+        run_id=3,
+        anchors=TEST_ANCHORS[:1],
+        detail_top_n=0,
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert len(nb_fetch_calls) == 0
+    assert counters.houses_attempted == 0
+    assert counters.houses_enriched == 0
+    assert fake.done is not None
+
+
+async def test_cancel_midway_stops_remaining(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Cooperative cancel перед anchor #2 → anchor #1 выполняется, #2+ нет."""
+    serp_calls: list[str] = []
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        serp_calls.append(f"{lat:.4f}")
+        return [_fake_lot(f"{lat}-1")]
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        AsyncMock(return_value=None),
+    )
+
+    fake = _FakeRuns(cancel_at_anchor=2)  # cancel при is_cancelled_calls >= 2 → перед anchor #2
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=MagicMock(),
+        run_id=4,
+        anchors=TEST_ANCHORS,
+        detail_top_n=0,
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert len(serp_calls) == 1, "только anchor #1 должен выполниться"
+    assert counters.anchors_done == 1
+    assert fake.done is None
+    assert fake.banned is None
+
+
+async def test_consecutive_serp_failures_abort_with_banned(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """N подряд SERP-ошибок → mark_banned + abort sweep."""
+    from app.services.scrape_pipeline import CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
+
+    attempt_count = 0
+
+    async def fake_fetch_fail(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        nonlocal attempt_count
+        attempt_count += 1
+        raise RuntimeError("cian blocked")
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_fail)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    # Передаём больше anchor'ов чем CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
+    anchors = [*TEST_ANCHORS, (56.77, 60.55, "A4"), (56.865, 60.62, "A5")]
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=MagicMock(),
+        run_id=5,
+        anchors=anchors,
+        detail_top_n=0,
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert attempt_count == CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
+    assert counters.errors_count == CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES
+    assert fake.banned is not None, "mark_banned должен быть вызван"
+    assert "consecutive" in fake.banned[0]
+    assert fake.done is None
+
+
+async def test_per_item_detail_error_does_not_abort_sweep(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Ошибка при detail-обогащении одного листинга не валит sweep."""
+    call_count = 0
+
+    async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        return [_fake_lot(f"lot-{idx}") for idx in range(2)]
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+
+    mock_db = MagicMock()
+    # Возвращаем 2 listing-а для detail
+    mock_db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 201, "source_url": "https://ekb.cian.ru/sale/flat/201/"},
+        {"id": 202, "source_url": "https://ekb.cian.ru/sale/flat/202/"},
+    ]
+
+    async def fetch_detail_sometimes_fails(url: str, **_: Any):
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            raise RuntimeError("transient network error")
+        from app.services.scrapers.cian_detail import DetailEnrichment
+
+        return DetailEnrichment(cian_id=202)
+
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        fetch_detail_sometimes_fails,
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.save_detail_enrichment",
+        MagicMock(),
+    )
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=mock_db,
+        run_id=6,
+        anchors=TEST_ANCHORS[:1],
+        detail_top_n=2,
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert counters.detail_attempted == 2
+    assert counters.detail_failed == 1
+    assert counters.detail_enriched == 1
+    assert counters.errors_count >= 1
+    assert fake.done is not None, "sweep должен завершиться mark_done несмотря на ошибку"
+    assert fake.banned is None
+
+
+async def test_mark_done_always_called(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """mark_done вызывается даже при пустом SERP."""
+
+    async def fake_fetch_empty(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8):
+        return []
+
+    monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_empty)
+    monkeypatch.setattr(
+        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
+    )
+    monkeypatch.setattr(
+        "app.services.scrapers.cian_detail.fetch_detail",
+        AsyncMock(return_value=None),
+    )
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_cian_city_sweep(
+        db=MagicMock(),
+        run_id=7,
+        anchors=TEST_ANCHORS,
+        detail_top_n=0,
+        enrich_houses=False,
+        request_delay_sec=0.0,
+    )
+
+    assert fake.done is not None
+    assert counters.lots_fetched == 0
+    assert counters.anchors_done == len(TEST_ANCHORS)
+
+
+# ── Admin endpoint tests (offline) ──────────────────────────────────────────
+
+
+@pytest.fixture
+def app_with_admin():
+    from unittest.mock import MagicMock
+
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+
+    from app.api.v1 import admin as admin_module
+    from app.core.db import get_db
+
+    app = FastAPI()
+    app.include_router(admin_module.router, prefix="/api/v1/admin")
+
+    def fake_db():
+        yield MagicMock()
+
+    app.dependency_overrides[get_db] = fake_db
+    return TestClient(app)
+
+
+def test_cian_start_endpoint_validates_pages_per_anchor(app_with_admin) -> None:
+    """pages_per_anchor=99 превышает max=10 → 422."""
+    r = app_with_admin.post(
+        "/api/v1/admin/scrape/cian-city-sweep",
+        json={"pages_per_anchor": 99},
+    )
+    assert r.status_code == 422
+
+
+def test_cian_start_endpoint_validates_delay_too_low(app_with_admin) -> None:
+    """request_delay_sec=1.0 меньше min=3.0 → 422."""
+    r = app_with_admin.post(
+        "/api/v1/admin/scrape/cian-city-sweep",
+        json={"request_delay_sec": 1.0},
+    )
+    assert r.status_code == 422
+
+
+def test_cian_start_endpoint_ok(app_with_admin) -> None:
+    """Valid request → 200, run_id в ответе."""
+    from unittest.mock import AsyncMock, patch
+
+    with (
+        patch("app.services.scrape_runs.create_run", return_value=42),
+        patch("app.services.scrape_pipeline.run_cian_city_sweep", new_callable=AsyncMock),
+    ):
+        r = app_with_admin.post(
+            "/api/v1/admin/scrape/cian-city-sweep",
+            json={"pages_per_anchor": 2, "detail_top_n": 5, "enrich_houses": False},
+        )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 42
+    assert body["status"] == "running"
+    assert body["pages_per_anchor"] == 2
+    assert body["detail_top_n"] == 5
+
+
+def test_cian_cancel_endpoint(app_with_admin) -> None:
+    """Cancel endpoint возвращает run_id + cancelled=True."""
+    from unittest.mock import patch
+
+    with patch("app.services.scrape_runs.mark_cancelled", return_value=True):
+        r = app_with_admin.post("/api/v1/admin/scrape/cian-city-sweep/77/cancel")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 77
+    assert body["cancelled"] is True
+    assert body["ok"] is True
+
+
+def test_cian_cancel_endpoint_not_running(app_with_admin) -> None:
+    """Cancel на не-running sweep → cancelled=False."""
+    from unittest.mock import patch
+
+    with patch("app.services.scrape_runs.mark_cancelled", return_value=False):
+        r = app_with_admin.post("/api/v1/admin/scrape/cian-city-sweep/88/cancel")
+    assert r.status_code == 200
+    assert r.json()["cancelled"] is False
+
+
+def test_cian_list_runs_endpoint(app_with_admin) -> None:
+    """List runs endpoint возвращает список."""
+    from datetime import UTC, datetime
+    from unittest.mock import patch
+
+    fake_rows = [
+        {
+            "run_id": 10,
+            "source": "cian_city_sweep",
+            "status": "done",
+            "params": {"pages_per_anchor": 3},
+            "counters": {"lots_fetched": 150},
+            "error": None,
+            "started_at": datetime(2025, 5, 1, tzinfo=UTC),
+            "finished_at": datetime(2025, 5, 1, 1, tzinfo=UTC),
+            "heartbeat_at": datetime(2025, 5, 1, 1, tzinfo=UTC),
+        }
+    ]
+    with patch("app.services.scrape_runs.list_recent", return_value=fake_rows):
+        r = app_with_admin.get("/api/v1/admin/scrape/cian-city-sweep/runs")
+    assert r.status_code == 200
+    body = r.json()
+    assert len(body) == 1
+    assert body[0]["run_id"] == 10
+    assert body[0]["source"] == "cian_city_sweep"
+    assert "2025-05-01" in body[0]["started_at"]