feat(tradein): yandex city-sweep — один прогон SERP+address-enrich

- YandexCitySweepCounters расширен: address_attempted/enriched/failed
  (pages_fetched удалён — fetch_around_multi_room агрегирует внутри)
- run_yandex_city_sweep: переключён на fetch_around_multi_room (rooms×price-range
  combos, #847/#860), добавлен параметр enrich_address=True
- Address-enrich фаза per anchor: для листингов без номера дома запрашивает
  detail <title>, извлекает полный адрес через yandex_address_backfill
  (_extract_address_from_title / _RE_HAS_HOUSE_NUMBER)
- admin.py: POST /scrape/yandex-city-sweep + GET .../runs + /{run_id}/cancel
  (модель YandexCitySweepStartRequest: pages_per_anchor/request_delay_sec/
  enrich_address/radius_m)
- scheduler.py: trigger_yandex_city_sweep_run передаёт enrich_address из params
- Тесты переписаны под новую сигнатуру: 16 тестов, все проходят

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

@@ -26,6 +26,7 @@ from app.services.scrape_pipeline import (
     run_avito_city_sweep,
     run_cian_city_sweep,
     run_n1_city_sweep,
+    run_yandex_city_sweep,
 )
 from app.services.scraper_settings import invalidate_cache
 from app.services.scrapers.avito import AvitoScraper
@@ -852,6 +853,93 @@ def cancel_cian_city_sweep(
     return {"ok": True, "run_id": run_id, "cancelled": cancelled}
 
 
+# ── Yandex city sweep (#861) — on-demand trigger + runs list ─────────────────
+
+
+class YandexCitySweepStartRequest(BaseModel):
+    pages_per_anchor: int = Field(default=2, ge=1, le=10)
+    request_delay_sec: float = Field(default=7.0, ge=3.0, le=30.0)
+    enrich_address: bool = True
+    radius_m: int = Field(default=1500, ge=500, le=5000)
+
+
+@router.post("/scrape/yandex-city-sweep", response_model=CitySweepStartResponse)
+async def start_yandex_city_sweep(
+    payload: YandexCitySweepStartRequest,
+    background_tasks: BackgroundTasks,
+    db: Annotated[Session, Depends(get_db)],
+) -> CitySweepStartResponse:
+    """Запустить Yandex.Недвижимость ЕКБ city sweep в background. Returns run_id.
+
+    Один прогон: SERP (fetch_around_multi_room, rooms × price-range combos)
+    → save_listings → address-enrich (detail <title> → полный адрес с домом).
+    Прокси уже внутри YandexRealtyScraper + address-enrich сессии.
+
+    Coop cancel: POST /scrape/yandex-city-sweep/{run_id}/cancel.
+    НЕ добавляется в scrape_schedules — ручной/контролируемый запуск
+    (schedule yandex_city_sweep DORMANT, включается оператором вручную).
+    """
+    run_id = runs_mod.create_run(db, source="yandex_city_sweep", params=payload.model_dump())
+
+    async def _sweep_task() -> None:
+        sweep_db = SessionLocal()
+        try:
+            await run_yandex_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,
+                enrich_address=payload.enrich_address,
+            )
+        except Exception:
+            logger.exception("yandex-sweep background task run_id=%d crashed", run_id)
+        finally:
+            sweep_db.close()
+
+    background_tasks.add_task(_sweep_task)
+    logger.info("yandex-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=0,  # у Yandex sweep нет detail-шага (address-enrich отдельно)
+    )
+
+
+@router.get("/scrape/yandex-city-sweep/runs", response_model=list[ScrapeRunRow])
+def list_yandex_city_sweep_runs(
+    db: Annotated[Session, Depends(get_db)],
+    limit: int = 10,
+) -> list[ScrapeRunRow]:
+    """Список последних N Yandex city sweep runs (для UI polling). Default limit=10."""
+    rows = runs_mod.list_recent(db, source="yandex_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/yandex-city-sweep/{run_id}/cancel")
+def cancel_yandex_city_sweep(
+    run_id: int,
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, object]:
+    """Отменить running Yandex 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/scheduler.py

@@ -253,6 +253,7 @@ async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any
                 pages_per_anchor=int(params.get("pages_per_anchor", 2)),
                 request_delay_sec=float(params.get("request_delay_sec", 9.0)),
                 radius_m=int(params.get("radius_m", 1500)),
+                enrich_address=bool(params.get("enrich_address", True)),
             )
         except Exception:
             logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id)

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

@@ -538,16 +538,18 @@ async def run_avito_city_sweep(
 class YandexCitySweepCounters:
     """Aggregate counters для Yandex.Недвижимость city sweep run.
 
-    Проще avito CitySweepCounters: у Yandex SERP нет house-catalog / detail-enrich
+    Фазы: SERP (fetch_around_multi_room) → save_listings → address-enrich.
-    шага (только search → save), поэтому houses/detail-полей нет.
+    address_* — результат T10-обогащения (detail <title> → полный адрес с номером дома).
     """
 
     anchors_total: int = 0
     anchors_done: int = 0
-    pages_fetched: 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]:
@@ -569,33 +571,50 @@ async def run_yandex_city_sweep(
     radius_m: int = 1500,
     pages_per_anchor: int = 2,
     request_delay_sec: float | None = None,
+    enrich_address: bool = True,
 ) -> YandexCitySweepCounters:
-    """Yandex.Недвижимость city sweep: iterate anchors × pages → save.
+    """Yandex.Недвижимость city sweep: SERP → save → address-enrich.
 
-    Проще avito sweep — у Yandex SERP нет house-catalog / detail-enrich шага:
+    Фазы per anchor:
-      per anchor: YandexRealtyScraper().fetch_around(lat, lon, radius_m, page=p)
+      1. SERP: YandexRealtyScraper.fetch_around_multi_room(lat, lon, radius_m,
-      → save_listings(db, lots) → accumulate counters.
+               max_pages=pages_per_anchor) — rooms × price-range combos, dedup внутри.
+      2. SAVE: save_listings(db, lots, run_id=run_id).
+      3. ADDRESS-ENRICH (если enrich_address=True): для yandex-листингов этого anchor'а
+         без номера дома в address (address IS NOT NULL AND NOT (address ~ ',\\s*\\d+'))
+         запрашиваем detail-страницу и извлекаем полный адрес из <title>.
+         Переиспользует приватные функции yandex_address_backfill:
+         _eligible_listings_for_ids → _fetch_and_update_one.
 
-    Anti-bot / cancel семантика (зеркало avito sweep, см. #561):
+    Anti-bot / cancel семантика:
-      - Cooperative cancel: is_cancelled(db, run_id) проверяется перед каждым anchor;
+      - Cooperative cancel: is_cancelled(db, run_id) проверяется перед каждым anchor.
-        при cancel — update_heartbeat + return (mark_cancelled делает caller-flow по статусу).
+      - YandexRealtyScraper не propagate'ит block/rate exceptions (глотает → []).
-      - YandexRealtyScraper НЕ определяет block/rate exception классов (fetch_around
+        Поэтому per-anchor ловим broad Exception; при N подряд — mark_banned + abort.
-        ловит исключения внутри и возвращает []). Поэтому per-anchor ловим broad
+      - request_delay_sec: доп. asyncio.sleep МЕЖДУ anchor'ами (поверх per-page sleep
-        Exception, логируем и continue; защитно прерываем sweep (early abort + mark_banned),
+        внутри scraper'а). Также используется как inter-request delay для address-enrich.
-        если 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'ам).
+    mark_done вызывается ВСЕГДА (кроме cancel / ban / fatal).
     """
+    from sqlalchemy import text as _text
+
+    from app.services.yandex_address_backfill import (
+        _RE_HAS_HOUSE_NUMBER,
+        _extract_address_from_title,
+    )
+
     _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
+    enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0
     consecutive_failures = 0
 
+    # Прокси для 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)",
@@ -616,29 +635,34 @@ async def run_yandex_city_sweep(
                 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)
 
+            # ── Phase 1+2: SERP + save ───────────────────────────────────────
+            anchor_lots: list[ScrapedLot] = []
+            try:
+                async with YandexRealtyScraper() as scraper:
+                    anchor_lots = await scraper.fetch_around_multi_room(
+                        lat,
+                        lon,
+                        radius_m,
+                        max_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(
+                    "yandex-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:
-                # YandexRealtyScraper не propagate'ит block/rate exceptions, поэтому здесь
+                # YandexRealtyScraper не propagate'ит block/rate exceptions — ловим broad.
-                # ловим broad Exception per-anchor: логируем и continue (изоляция anchor'а).
+                logger.exception("yandex-sweep run_id=%d: anchor %s SERP failed", run_id, name)
-                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:
@@ -662,6 +686,128 @@ async def run_yandex_city_sweep(
                         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
+
+            # ── Phase 3: address-enrich ─────────────────────────────────────
+            if enrich_address and anchor_lots:
+                # Собираем source_id из SERP-результатов этого anchor'а → ищем
+                # в DB только листинги, которые были только что сохранены/обновлены.
+                source_ids = [lot.source_id for lot in anchor_lots if lot.source_id]
+                if source_ids:
+                    id_rows = (
+                        db.execute(
+                            _text("""
+                                SELECT id, source_id, address, source_url
+                                FROM listings
+                                WHERE source = 'yandex'
+                                  AND source_id = ANY(CAST(:ids AS text[]))
+                                  AND address IS NOT NULL
+                                  AND NOT (address ~ ',\\s*\\d+')
+                                  AND source_url IS NOT NULL
+                            """),
+                            {"ids": source_ids},
+                        )
+                        .mappings()
+                        .all()
+                    )
+                    items = [dict(r) for r in id_rows]
+                    counters.address_attempted += len(items)
+
+                    if items:
+                        from curl_cffi.requests import AsyncSession as _AsyncSession
+
+                        async with _AsyncSession(
+                            impersonate="chrome120",
+                            timeout=30.0,
+                            proxies=_proxies,
+                            headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"},
+                        ) as enrich_session:
+                            for eidx, item in enumerate(items):
+                                lid: int = item["id"]
+                                old_addr: str = item["address"]
+                                url: str = item["source_url"]
+                                try:
+                                    resp = await enrich_session.get(url, allow_redirects=True)
+                                    if resp.status_code != 200:
+                                        logger.warning(
+                                            "yandex-sweep run_id=%d address-enrich:"
+                                            " HTTP %d listing_id=%d",
+                                            run_id,
+                                            resp.status_code,
+                                            lid,
+                                        )
+                                        counters.address_failed += 1
+                                    else:
+                                        new_addr = _extract_address_from_title(resp.text)
+                                        if (
+                                            new_addr is None
+                                            or new_addr == old_addr
+                                            or not _RE_HAS_HOUSE_NUMBER.search(new_addr)
+                                        ):
+                                            # Не удалось улучшить адрес — пропускаем (не ошибка).
+                                            pass
+                                        else:
+                                            try:
+                                                db.execute(
+                                                    _text("""
+                                                        UPDATE listings
+                                                        SET address = :addr,
+                                                            geocode_tried_at = NULL
+                                                        WHERE id = CAST(:id AS bigint)
+                                                    """),
+                                                    {"addr": new_addr, "id": lid},
+                                                )
+                                                db.commit()
+                                                counters.address_enriched += 1
+                                                logger.info(
+                                                    "yandex-sweep run_id=%d: "
+                                                    "address updated listing_id=%d "
+                                                    "%r -> %r",
+                                                    run_id,
+                                                    lid,
+                                                    old_addr,
+                                                    new_addr,
+                                                )
+                                            except Exception as upd_exc:
+                                                counters.address_failed += 1
+                                                counters.errors_count += 1
+                                                logger.warning(
+                                                    "yandex-sweep run_id=%d: "
+                                                    "address DB update failed "
+                                                    "listing_id=%d: %s",
+                                                    run_id,
+                                                    lid,
+                                                    upd_exc,
+                                                )
+                                                try:
+                                                    db.rollback()
+                                                except Exception:
+                                                    pass
+                                except Exception as fetch_exc:
+                                    counters.address_failed += 1
+                                    logger.warning(
+                                        "yandex-sweep run_id=%d: address-enrich "
+                                        "fetch failed listing_id=%d: %s",
+                                        run_id,
+                                        lid,
+                                        fetch_exc,
+                                    )
+                                if eidx < len(items) - 1:
+                                    await asyncio.sleep(enrich_delay)
+
+                        logger.info(
+                            "yandex-sweep run_id=%d anchor %s: " "address enrich=%d/%d failed=%d",
+                            run_id,
+                            name,
+                            counters.address_enriched,
+                            counters.address_attempted,
+                            counters.address_failed,
+                        )
 
             counters.anchors_done = idx
             scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
@@ -672,14 +818,17 @@ 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 lots=%d (ins=%d/upd=%d) "
+            "address=%d/%d failed=%d errors=%d",
             run_id,
             counters.anchors_done,
             counters.anchors_total,
-            counters.pages_fetched,
             counters.lots_fetched,
             counters.lots_inserted,
             counters.lots_updated,
+            counters.address_enriched,
+            counters.address_attempted,
+            counters.address_failed,
             counters.errors_count,
         )
         return counters

tradein-mvp/backend/tests/test_yandex_city_sweep.py

@@ -1,20 +1,31 @@
-"""Offline unit tests for run_yandex_city_sweep (#561, shipped DORMANT).
+"""Offline unit tests for run_yandex_city_sweep (SERP+address-enrich, #861).
 
 Pure & fast: NO live network, NO DB. We monkeypatch:
   - YandexRealtyScraper.__init__ / __aenter__ / __aexit__ — avoid get_scraper_delay
-    DB read + real httpx.AsyncClient creation.
+    DB read + real curl_cffi session creation.
-  - YandexRealtyScraper.fetch_around — return a fixed list of fake ScrapedLot.
+  - YandexRealtyScraper.fetch_around_multi_room — return a fixed list of fake ScrapedLot.
   - scrape_pipeline.save_listings — counter/no-op (no real listings table).
   - scrape_runs.* (is_cancelled / update_heartbeat / mark_*) — in-memory fakes.
+  - curl_cffi.requests.AsyncSession — fake for address-enrich HTTP calls.
 
-These tests assert orchestration semantics only; live Yandex behavior is intentionally
+Asserts orchestration semantics:
-unverified (sweep is dormant until egress-IP rotation #623).
+  - SERP → save (all anchors, counters aggregated)
+  - address-enrich фаза вызывается если enrich_address=True и есть source_ids
+  - enrich_address=False → адрес-фаза пропускается
+  - cancel → останавливает sweep
+  - N consecutive SERP failures → mark_banned + abort
+  - per-item address-enrich 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
 
@@ -22,7 +33,7 @@ from app.services import scrape_pipeline
 from app.services.scrapers.base import ScrapedLot
 from app.services.scrapers.yandex_realty import YandexRealtyScraper
 
-# 3 anchors keep tests fast and let us probe "anchor k cancels/fails, others proceed".
+# 3 anchors keep tests fast and let us probe cancel/fail semantics.
 TEST_ANCHORS: list[tuple[float, float, str]] = [
     (56.8400, 60.6050, "A1"),
     (56.7950, 60.5300, "A2"),
@@ -36,7 +47,7 @@ def _fake_lot(offer_id: str) -> ScrapedLot:
         source="yandex",
         source_url=f"https://realty.yandex.ru/offer/{offer_id}/",
         source_id=offer_id,
-        address="Екатеринбург, улица Тестовая, 1",
+        address="Екатеринбург, улица Тестовая",
         price_rub=5_000_000,
         area_m2=50.0,
         rooms=2,
@@ -77,6 +88,7 @@ class _FakeRuns:
 @pytest.fixture(autouse=True)
 def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
     """Kill inter-anchor asyncio.sleep so tests are instant."""
+
     async def _instant(_secs: float) -> None:
         return None
 
@@ -85,12 +97,13 @@ def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
 
 @pytest.fixture(autouse=True)
 def _stub_scraper_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Neutralize __init__/__aenter__/__aexit__ — no get_scraper_delay DB, no httpx."""
+    """Neutralize __init__/__aenter__/__aexit__ — no get_scraper_delay DB, no curl_cffi."""
 
     def _init(self: YandexRealtyScraper, city: str = "ekaterinburg") -> None:
         self.city = city
         self.request_delay_sec = 0.0
-        self._client = None
+        self._cffi_session = None
+        self._cookies = {}
 
     async def _aenter(self: YandexRealtyScraper) -> YandexRealtyScraper:
         return self
@@ -111,114 +124,262 @@ def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None:
     monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned)
 
 
+def _make_db_mock(source_ids: list[str]) -> MagicMock:
+    """DB mock для address-enrich фазы: возвращает строки без номера дома."""
+    mock_db = MagicMock()
+    rows = [
+        {
+            "id": i + 1,
+            "source_id": sid,
+            "address": "Екатеринбург, улица Тестовая",
+            "source_url": f"https://realty.yandex.ru/offer/{sid}/",
+        }
+        for i, sid in enumerate(source_ids)
+    ]
+    mock_db.execute.return_value.mappings.return_value.all.return_value = rows
+    return mock_db
+
+
+def _fake_save(_db: Any, lots: list, *, run_id: int | None = None) -> tuple[int, int]:
+    return len(lots), 0
+
+
+# ── YandexCitySweepCounters ─────────────────────────────────────────────────
+
+
+def test_counters_defaults() -> None:
+    from app.services.scrape_pipeline import YandexCitySweepCounters
+
+    c = YandexCitySweepCounters()
+    assert c.anchors_total == 0
+    assert c.lots_fetched == 0
+    assert c.address_attempted == 0
+    assert c.address_enriched == 0
+    assert c.address_failed == 0
+    assert c.errors_count == 0
+
+
+def test_counters_to_dict_all_keys() -> None:
+    from dataclasses import fields
+
+    from app.services.scrape_pipeline import YandexCitySweepCounters
+
+    c = YandexCitySweepCounters()
+    d = c.to_dict()
+    expected = {f.name for f in fields(c)}
+    assert set(d.keys()) == expected
+
+
 # ── Happy path: all anchors iterated, counters aggregated ───────────────────
 
 
 async def test_sweep_iterates_all_anchors_and_aggregates(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    fetch_calls: list[tuple[float, float, int]] = []
+    """fetch_around_multi_room вызывается для каждого anchor, counters агрегируются."""
+    serp_calls: list[str] = []
     save_calls: list[int] = []
 
-    async def fake_fetch(
+    async def fake_fetch_multi(
         self: YandexRealtyScraper,
         lat: float,
         lon: float,
         radius_m: int = 1000,
-        page: int = 0,
+        max_pages: int = 2,
+        **_: Any,
     ) -> list[ScrapedLot]:
-        fetch_calls.append((lat, lon, page))
+        serp_calls.append(f"{lat:.4f}")
-        # 2 lots per page; let pagination run the full pages_per_anchor.
+        return [_fake_lot(f"{lat}-{i}") for i in range(3)]
-        return [_fake_lot(f"{lat}-{page}-{i}") for i in range(2)]
 
     def fake_save(
         _db: Any, lots: list[ScrapedLot], *, run_id: int | None = None
     ) -> tuple[int, int]:
         save_calls.append(len(lots))
-        # Pretend everything is a fresh insert.
         return len(lots), 0
 
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
     monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
     fake = _FakeRuns()
     _install_runs(monkeypatch, fake)
 
     counters = await scrape_pipeline.run_yandex_city_sweep(
-        db=object(),
+        db=MagicMock(),
         run_id=1,
         anchors=TEST_ANCHORS,
         pages_per_anchor=2,
         request_delay_sec=0.0,
+        enrich_address=False,
     )
 
-    # fetch_around called pages_per_anchor times per anchor (no empty-page short-circuit).
+    assert len(serp_calls) == len(TEST_ANCHORS)
-    assert len(fetch_calls) == len(TEST_ANCHORS) * 2
-    # save_listings called once per anchor (anchor-level batch).
     assert len(save_calls) == len(TEST_ANCHORS)
-    # Counters: 3 anchors × 2 pages × 2 lots = 12.
     assert counters.anchors_total == len(TEST_ANCHORS)
     assert counters.anchors_done == len(TEST_ANCHORS)
-    assert counters.pages_fetched == len(TEST_ANCHORS) * 2
+    assert counters.lots_fetched == len(TEST_ANCHORS) * 3
-    assert counters.lots_fetched == 12
+    assert counters.lots_inserted == len(TEST_ANCHORS) * 3
-    assert counters.lots_inserted == 12
     assert counters.lots_updated == 0
     assert counters.errors_count == 0
-    # mark_done called with final counters; no failure/ban.
+    assert counters.address_attempted == 0  # enrich_address=False
     assert fake.done is not None
-    assert fake.done["lots_fetched"] == 12
+    assert fake.done["lots_fetched"] == len(TEST_ANCHORS) * 3
     assert fake.failed is None
     assert fake.banned is None
 
 
-async def test_save_listings_receives_run_id(monkeypatch: pytest.MonkeyPatch) -> None:
+# ── enrich_address=False пропускает address-фазу ────────────────────────────
-    """run_id must be threaded into save_listings for snapshot provenance."""
-    seen_run_ids: list[int | None] = []
-
-    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
-        return [_fake_lot(f"{lat}-{page}")]
-
-    def fake_save(_db, lots, *, run_id=None):
-        seen_run_ids.append(run_id)
-        return len(lots), 0
-
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
-    monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save)
-    _install_runs(monkeypatch, _FakeRuns())
-
-    await scrape_pipeline.run_yandex_city_sweep(
-        db=object(), run_id=777, anchors=TEST_ANCHORS, pages_per_anchor=1,
-    )
-    assert seen_run_ids == [777, 777, 777]
 
 
-# ── Empty page short-circuits pagination ────────────────────────────────────
+async def test_enrich_address_false_skips_address_phase(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """enrich_address=False → address-enrich фаза не выполняется."""
+    address_http_calls: list[str] = []
 
+    async def fake_fetch_multi(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        return [_fake_lot(f"{lat}-0")]
 
-async def test_empty_page_stops_pagination(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_cffi_get(url: str, **_: Any) -> Any:
-    """An empty SERP page ends pagination for that anchor (no further pages)."""
+        address_http_calls.append(url)
-    pages_seen: list[int] = []
+        return MagicMock(status_code=200, text="<title>test</title>")
 
-    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
-        pages_seen.append(page)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
-        if page == 0:
-            return [_fake_lot(f"{lat}-0")]
-        return []  # page 1 empty → stop
-
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
-    monkeypatch.setattr(
-        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
-    )
     fake = _FakeRuns()
     _install_runs(monkeypatch, fake)
 
     counters = await scrape_pipeline.run_yandex_city_sweep(
-        db=object(), run_id=2, anchors=TEST_ANCHORS, pages_per_anchor=5,
+        db=MagicMock(),
+        run_id=2,
+        anchors=TEST_ANCHORS[:1],
+        enrich_address=False,
+        request_delay_sec=0.0,
     )
-    # Per anchor: page 0 (1 lot) then page 1 (empty, break) → 2 pages each.
+
-    assert pages_seen == [0, 1, 0, 1, 0, 1]
+    assert len(address_http_calls) == 0
-    assert counters.pages_fetched == 6  # 2 per anchor × 3
+    assert counters.address_attempted == 0
-    assert counters.lots_fetched == 3  # 1 per anchor
+    assert counters.address_enriched == 0
-    assert counters.anchors_done == 3
+    assert fake.done is not None
+
+
+# ── address-enrich фаза: счётчики заполняются ───────────────────────────────
+
+
+async def test_address_enrich_fills_counters(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """address-enrich вызывается для листингов без номера дома; counters заполняются."""
+    # Fake SERP возвращает 2 листинга с source_id
+    lots = [_fake_lot("id1"), _fake_lot("id2")]
+
+    async def fake_fetch_multi(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        return lots
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
+
+    # DB mock: возвращает 2 строки для address-enrich запроса
+    mock_db = _make_db_mock(["id1", "id2"])
+
+    # Title с полным адресом — должен быть извлечён через _extract_address_from_title
+    fake_response = MagicMock()
+    fake_response.status_code = 200
+    fake_response.text = (
+        "<title>Продажа квартиры — Екатеринбург, улица Тестовая, 36 — id 123 "
+        "на Яндекс.Недвижимости</title>"
+    )
+
+    # Мок curl_cffi AsyncSession как context manager
+    mock_session = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=None)
+    mock_session.get = AsyncMock(return_value=fake_response)
+
+    import curl_cffi.requests as cffi_mod
+
+    monkeypatch.setattr(cffi_mod, "AsyncSession", lambda **_kw: mock_session)
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=mock_db,
+        run_id=3,
+        anchors=TEST_ANCHORS[:1],
+        enrich_address=True,
+        request_delay_sec=0.0,
+    )
+
+    assert counters.address_attempted == 2
+    assert counters.address_enriched == 2
+    assert counters.address_failed == 0
+    assert fake.done is not None
+
+
+# ── address-enrich: per-item HTTP error не валит sweep ──────────────────────
+
+
+async def test_address_enrich_http_error_does_not_abort(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """HTTP 403 при address-enrich → address_failed++, sweep продолжается."""
+
+    async def fake_fetch_multi(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        return [_fake_lot("id-err")]
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
+
+    mock_db = _make_db_mock(["id-err"])
+
+    fake_response = MagicMock()
+    fake_response.status_code = 403
+    fake_response.text = ""
+
+    mock_session = AsyncMock()
+    mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session.__aexit__ = AsyncMock(return_value=None)
+    mock_session.get = AsyncMock(return_value=fake_response)
+
+    import curl_cffi.requests as cffi_mod
+
+    monkeypatch.setattr(cffi_mod, "AsyncSession", lambda **_kw: mock_session)
+
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=mock_db,
+        run_id=4,
+        anchors=TEST_ANCHORS[:1],
+        enrich_address=True,
+        request_delay_sec=0.0,
+    )
+
+    assert counters.address_attempted == 1
+    assert counters.address_failed == 1
+    assert counters.address_enriched == 0
+    assert fake.done is not None  # sweep завершился несмотря на ошибку
 
 
 # ── Mid-sweep cancel stops further anchors ──────────────────────────────────
@@ -227,25 +388,35 @@ async def test_empty_page_stops_pagination(monkeypatch: pytest.MonkeyPatch) -> N
 async def test_cancel_midway_stops_remaining_anchors(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    fetch_calls: list[str] = []
+    serp_calls: list[str] = []
 
-    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+    async def fake_fetch_multi(
-        fetch_calls.append(f"{lat}")
+        self: YandexRealtyScraper,
-        return [_fake_lot(f"{lat}-{page}")]
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        serp_calls.append(f"{lat}")
+        return [_fake_lot(f"{lat}-0")]
 
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_multi)
-    monkeypatch.setattr(
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
-        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
-    )
     # is_cancelled flips True on the 2nd check (before anchor #2) → only anchor #1 runs.
     fake = _FakeRuns(cancel_at_anchor=2)
     _install_runs(monkeypatch, fake)
 
     counters = await scrape_pipeline.run_yandex_city_sweep(
-        db=object(), run_id=3, anchors=TEST_ANCHORS, pages_per_anchor=1,
+        db=MagicMock(),
+        run_id=5,
+        anchors=TEST_ANCHORS,
+        pages_per_anchor=1,
+        enrich_address=False,
+        request_delay_sec=0.0,
     )
     # Only the first anchor fetched before cancel.
-    assert len(fetch_calls) == 1
+    assert len(serp_calls) == 1
     assert counters.anchors_done == 1
     # Cancel path returns early: no mark_done, no ban/fail.
     assert fake.done is None
@@ -253,48 +424,7 @@ async def test_cancel_midway_stops_remaining_anchors(
     assert fake.banned is None
 
 
-# ── Per-anchor exception is isolated (single failure → continue) ────────────
+# ── Consecutive failures → mark_banned + abort ──────────────────────────────
-
-
-async def test_single_anchor_failure_is_isolated(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Anchor #2 raises → anchors #1 and #3 still processed; sweep completes."""
-    processed: list[str] = []
-
-    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
-        if name_of(lat) == "A2":
-            raise RuntimeError("boom on anchor 2")
-        processed.append(name_of(lat))
-        return [_fake_lot(f"{lat}-{page}")]
-
-    def name_of(lat: float) -> str:
-        for a_lat, _lon, nm in TEST_ANCHORS:
-            if a_lat == lat:
-                return nm
-        return "?"
-
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
-    monkeypatch.setattr(
-        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
-    )
-    fake = _FakeRuns()
-    _install_runs(monkeypatch, fake)
-
-    counters = await scrape_pipeline.run_yandex_city_sweep(
-        db=object(), run_id=4, anchors=TEST_ANCHORS, pages_per_anchor=1,
-    )
-    # A1 and A3 processed; A2 raised and was skipped.
-    assert processed == ["A1", "A3"]
-    assert counters.errors_count == 1
-    assert counters.anchors_done == 3  # loop advanced past all 3 anchors
-    assert counters.lots_fetched == 2  # only A1 + A3 contributed lots
-    # Not an abort — sweep marked done despite the isolated failure.
-    assert fake.done is not None
-    assert fake.banned is None
-
-
-# ── Defensive abort: N consecutive failures → mark_banned, stop early ───────
 
 
 async def test_consecutive_failures_abort_sweep(
@@ -303,14 +433,19 @@ async def test_consecutive_failures_abort_sweep(
     """All anchors raise → after MAX_CONSECUTIVE_FAILURES the sweep aborts + bans."""
     attempts: list[str] = []
 
-    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+    async def fake_fetch_fail(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
         attempts.append(f"{lat}")
         raise RuntimeError("yandex blocked")
 
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_fail)
-    monkeypatch.setattr(
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
-        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
-    )
     fake = _FakeRuns()
     _install_runs(monkeypatch, fake)
 
@@ -321,9 +456,13 @@ async def test_consecutive_failures_abort_sweep(
         (56.8650, 60.6200, "A5"),
     ]
     counters = await scrape_pipeline.run_yandex_city_sweep(
-        db=object(), run_id=5, anchors=anchors, pages_per_anchor=1,
+        db=MagicMock(),
+        run_id=6,
+        anchors=anchors,
+        enrich_address=False,
+        request_delay_sec=0.0,
     )
-    # Abort exactly at the threshold (3 consecutive failures), not all 5 anchors.
+    # Abort exactly at the threshold (3 consecutive failures).
     assert len(attempts) == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES
     assert counters.errors_count == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES
     assert counters.anchors_done == scrape_pipeline.YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES
@@ -332,11 +471,15 @@ async def test_consecutive_failures_abort_sweep(
     assert fake.done is None  # aborted, never marked done
 
 
+# ── Non-consecutive failures do not abort ───────────────────────────────────
+
+
 async def test_non_consecutive_failures_do_not_abort(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     """A success between two failures resets the consecutive counter → no abort."""
-    def name_of(lat: float, anchors: list[tuple[float, float, str]]) -> str:
+
+    def _name_of(lat: float, anchors: list[tuple[float, float, str]]) -> str:
         for a_lat, _lon, nm in anchors:
             if a_lat == lat:
                 return nm
@@ -347,26 +490,178 @@ async def test_non_consecutive_failures_do_not_abort(
         (56.7700, 60.5500, "A4"),
         (56.8650, 60.6200, "A5"),
     ]
-    # Fail A1, succeed A2, fail A3, succeed A4, fail A5 → never 3-in-a-row.
     failing = {"A1", "A3", "A5"}
 
-    async def fake_fetch(self: YandexRealtyScraper, lat, lon, radius_m=1000, page=0):
+    async def fake_fetch(
-        if name_of(lat, anchors) in failing:
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        if _name_of(lat, anchors) in failing:
             raise RuntimeError("transient")
-        return [_fake_lot(f"{lat}-{page}")]
+        return [_fake_lot(f"{lat}-0")]
 
-    monkeypatch.setattr(YandexRealtyScraper, "fetch_around", fake_fetch)
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch)
-    monkeypatch.setattr(
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
-        scrape_pipeline, "save_listings", lambda _db, lots, *, run_id=None: (len(lots), 0)
-    )
     fake = _FakeRuns()
     _install_runs(monkeypatch, fake)
 
     counters = await scrape_pipeline.run_yandex_city_sweep(
-        db=object(), run_id=6, anchors=anchors, pages_per_anchor=1,
+        db=MagicMock(),
+        run_id=7,
+        anchors=anchors,
+        enrich_address=False,
+        request_delay_sec=0.0,
     )
     assert counters.errors_count == 3  # A1, A3, A5
     assert counters.anchors_done == 5  # walked the whole list, no early abort
     assert counters.lots_fetched == 2  # A2 + A4 contributed
     assert fake.banned is None
     assert fake.done is not None
+
+
+# ── mark_done вызывается даже при пустом SERP ───────────────────────────────
+
+
+async def test_mark_done_always_called(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def fake_fetch_empty(
+        self: YandexRealtyScraper,
+        lat: float,
+        lon: float,
+        radius_m: int = 1000,
+        max_pages: int = 2,
+        **_: Any,
+    ) -> list[ScrapedLot]:
+        return []
+
+    monkeypatch.setattr(YandexRealtyScraper, "fetch_around_multi_room", fake_fetch_empty)
+    monkeypatch.setattr(scrape_pipeline, "save_listings", _fake_save)
+    fake = _FakeRuns()
+    _install_runs(monkeypatch, fake)
+
+    counters = await scrape_pipeline.run_yandex_city_sweep(
+        db=MagicMock(),
+        run_id=8,
+        anchors=TEST_ANCHORS,
+        enrich_address=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 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_yandex_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/yandex-city-sweep",
+        json={"pages_per_anchor": 99},
+    )
+    assert r.status_code == 422
+
+
+def test_yandex_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/yandex-city-sweep",
+        json={"request_delay_sec": 1.0},
+    )
+    assert r.status_code == 422
+
+
+def test_yandex_start_endpoint_ok(app_with_admin) -> None:
+    """Valid request → 200, run_id в ответе."""
+    from unittest.mock import patch
+
+    with (
+        patch("app.services.scrape_runs.create_run", return_value=55),
+        patch("app.services.scrape_pipeline.run_yandex_city_sweep", new_callable=AsyncMock),
+    ):
+        r = app_with_admin.post(
+            "/api/v1/admin/scrape/yandex-city-sweep",
+            json={"pages_per_anchor": 2, "enrich_address": True},
+        )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 55
+    assert body["status"] == "running"
+    assert body["pages_per_anchor"] == 2
+    assert body["detail_top_n"] == 0
+
+
+def test_yandex_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/yandex-city-sweep/99/cancel")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 99
+    assert body["cancelled"] is True
+    assert body["ok"] is True
+
+
+def test_yandex_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/yandex-city-sweep/100/cancel")
+    assert r.status_code == 200
+    assert r.json()["cancelled"] is False
+
+
+def test_yandex_list_runs_endpoint(app_with_admin) -> None:
+    """List runs endpoint возвращает список."""
+    from datetime import UTC, datetime
+    from unittest.mock import patch
+
+    fake_rows = [
+        {
+            "run_id": 20,
+            "source": "yandex_city_sweep",
+            "status": "done",
+            "params": {"pages_per_anchor": 2, "enrich_address": True},
+            "counters": {"lots_fetched": 200, "address_enriched": 80},
+            "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/yandex-city-sweep/runs")
+    assert r.status_code == 200
+    body = r.json()
+    assert len(body) == 1
+    assert body[0]["run_id"] == 20
+    assert body[0]["source"] == "yandex_city_sweep"
+    assert "2025-05-01" in body[0]["started_at"]