From 7ce758419a5591e8bfa79dcb10170c98c697cf40 Mon Sep 17 00:00:00 2001
From: bot-backend <bot-backend@gendsgn.local>
Date: Sun, 31 May 2026 10:06:37 +0300
Subject: [PATCH] =?UTF-8?q?feat(tradein):=20avito=20city-sweep=20=E2=80=94?=
 =?UTF-8?q?=20IMV=20enrichment=20phase=20(=D0=BE=D0=B4=D0=B8=D0=BD=20?=
 =?UTF-8?q?=D0=BF=D1=80=D0=BE=D0=B3=D0=BE=D0=BD=20=3D=20SERP+detail+houses?=
 =?UTF-8?q?+IMV)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Встраивает IMV-оценку домов как финальную фазу run_avito_city_sweep.
Один запуск sweep'а теперь наполняет: listings (SERP) + detail-обогащение
+ houses (catalog/reviews/sellers) + IMV-оценку тронутых домов.

- scrape_pipeline: добавлен enrich_imv param, CitySweepCounters.imv_attempted/
  imv_enriched/imv_failed, PipelineResult.touched_house_ids (собирается при
  houses-enrichment из save_house_catalog_enrichment['house_id']).
  После всех anchor'ов — IMV-фаза: process_houses_imv_batch для touched_house_ids,
  cooperative cancel check, graceful per-house error handling (не валит sweep).
- house_imv_backfill: добавлена process_houses_imv_batch — тонкая обёртка для
  обработки конкретного set[house_id] (пропускает imv_status=ok, дома без coords).
- admin: CitySweepStartRequest.enrich_imv: bool = True + проброс в sweep.
- tests: 14 новых тестов (enrich_imv=True/False, counters, cancel-before-IMV,
  IMV crash graceful, process_houses_imv_batch empty, API param).
---
 tradein-mvp/backend/app/api/v1/admin.py       |   8 +
 .../app/services/house_imv_backfill.py        | 105 ++++
 .../backend/app/services/scrape_pipeline.py   |  85 +++-
 .../backend/tests/test_sweep_imv_phase.py     | 476 ++++++++++++++++++
 4 files changed, 668 insertions(+), 6 deletions(-)
 create mode 100644 tradein-mvp/backend/tests/test_sweep_imv_phase.py

diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py
index c977e568..c2c1228d 100644
--- a/tradein-mvp/backend/app/api/v1/admin.py
+++ b/tradein-mvp/backend/app/api/v1/admin.py
@@ -587,6 +587,13 @@ class CitySweepStartRequest(BaseModel):
     detail_top_n: int = Field(default=10, ge=0, le=30)
     request_delay_sec: float = Field(default=7.0, ge=3.0, le=15.0)
     enrich_houses: bool = True
+    enrich_imv: bool = Field(
+        default=True,
+        description=(
+            "Если True — после всех anchor'ов запускает IMV-оценку тронутых домов "
+            "(финальная фаза sweep'а). Требует enrich_houses=True для сбора house_id."
+        ),
+    )
     radius_m: int = Field(default=1500, ge=500, le=5000)
 
 
@@ -636,6 +643,7 @@ async def start_avito_city_sweep(
                 enrich_houses=payload.enrich_houses,
                 detail_top_n=payload.detail_top_n,
                 request_delay_sec=payload.request_delay_sec,
+                enrich_imv=payload.enrich_imv,
             )
         except Exception:
             logger.exception("city-sweep background task run_id=%d crashed", run_id)
diff --git a/tradein-mvp/backend/app/services/house_imv_backfill.py b/tradein-mvp/backend/app/services/house_imv_backfill.py
index 36e27bbd..5515bd4a 100644
--- a/tradein-mvp/backend/app/services/house_imv_backfill.py
+++ b/tradein-mvp/backend/app/services/house_imv_backfill.py
@@ -455,6 +455,111 @@ async def backfill_house_imv(
     return result
 
 
+async def process_houses_imv_batch(
+    db: Session,
+    house_ids: set[int],
+    *,
+    request_delay_sec: float = 5.0,
+) -> HouseIMVBackfillResult:
+    """Обработать конкретный набор house_id (для sweep-интеграции).
+
+    Зеркало backfill_house_imv, но работает по явному списку house_id
+    (не по imv_status-фильтру). Предназначен для финальной IMV-фазы
+    run_avito_city_sweep — обрабатываем только дома, тронутые за этот sweep.
+
+    Дома без lat/lon или address пропускаются (статус no_params/no_address).
+    Дома уже с imv_status='ok' тоже пропускаются — UPSERT всё равно обновит,
+    поэтому лучше не гонять лишних API-запросов, если данные свежие.
+    Для форс-обновления — оставь это решение caller'у (sweep по умолчанию
+    обходит дома с ok, обновляя только pending/not_found/error/transient_error).
+    """
+    result = HouseIMVBackfillResult()
+    t0 = time.time()
+
+    if not house_ids:
+        result.duration_sec = time.time() - t0
+        return result
+
+    ids_list = list(house_ids)
+
+    # Загружаем только дома, у которых есть координаты + адрес + imv_status != 'ok'
+    # (уже оцененные дома пропускаем — не тратим API-квоту повторно)
+    placeholders = ", ".join(f":hid_{i}" for i in range(len(ids_list)))
+    params: dict[str, object] = {f"hid_{i}": hid for i, hid in enumerate(ids_list)}
+    rows = (
+        db.execute(
+            text(f"""
+                SELECT id, address, full_address, lat, lon
+                FROM houses
+                WHERE id IN ({placeholders})
+                  AND lat IS NOT NULL
+                  AND lon IS NOT NULL
+                  AND address IS NOT NULL
+                  AND COALESCE(imv_status, 'pending') != 'ok'
+                ORDER BY id
+            """),
+            params,
+        )
+        .mappings()
+        .all()
+    )
+
+    result.checked = len(rows)
+    if not rows:
+        logger.info(
+            "process_houses_imv_batch: nothing to process (%d house_ids, all ok/no-coords)",
+            len(house_ids),
+        )
+        result.duration_sec = time.time() - t0
+        return result
+
+    logger.info(
+        "process_houses_imv_batch: %d/%d houses eligible (delay=%.1fs)",
+        result.checked,
+        len(house_ids),
+        request_delay_sec,
+    )
+
+    for i, house in enumerate(rows):
+        hid: int = house["id"]
+        status_str = await _process_one_house(db, dict(house))
+
+        result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1
+        if status_str == "ok":
+            result.saved += 1
+        elif status_str in ("no_params", "no_address"):
+            result.skipped += 1
+        else:
+            result.errors += 1
+
+        logger.info(
+            "[%d/%d] imv_batch house_id=%d status=%s",
+            i + 1,
+            result.checked,
+            hid,
+            status_str,
+        )
+
+        if i < len(rows) - 1:
+            if status_str == "auth_error":
+                logger.warning("IMV auth error — extra delay 60s before next house")
+                await asyncio.sleep(60)
+            else:
+                await asyncio.sleep(request_delay_sec)
+
+    result.duration_sec = time.time() - t0
+    logger.info(
+        "process_houses_imv_batch done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s",
+        result.checked,
+        result.saved,
+        result.skipped,
+        result.errors,
+        result.duration_sec,
+        result.status_counts,
+    )
+    return result
+
+
 async def _process_one_house(db: Session, house: dict) -> str:
     """Process a single house. Returns final status string."""
     hid: int = house["id"]
diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py
index 382bb53c..7c50d0b0 100644
--- a/tradein-mvp/backend/app/services/scrape_pipeline.py
+++ b/tradein-mvp/backend/app/services/scrape_pipeline.py
@@ -19,7 +19,10 @@ Anti-bot hardening (2023-05-23):
 Graceful degradation на каждом step: exception в одной house/listing
 не валит весь pipeline (кроме blocked exceptions — они abort весь sweep).
 
-IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron.
+IMV-фаза (финальная, после всех anchor'ов): если enrich_imv=True —
+для домов, тронутых в этом sweep'е (touched house_id из houses-enrichment),
+вызывается house_imv_backfill.process_houses_imv_batch. Ошибки per-house
+логируются и считаются, но не валят sweep.
 """
 
 from __future__ import annotations
@@ -111,6 +114,7 @@ class PipelineResult:
     counters: PipelineCounters
     enrich_houses: bool
     enrich_detail_top_n: int
+    touched_house_ids: set[int] = field(default_factory=set)
 
 
 # ── Main orchestrator ───────────────────────────────────────────
@@ -203,13 +207,17 @@ async def run_avito_pipeline(
         logger.info("pipeline:group_by_house unique=%d", counters.unique_houses)
 
         # ── Step 4: enrich houses с shared session + sleep ──────
+        touched_house_ids: set[int] = set()
         if enrich_houses and unique_house_paths:
             house_paths_list = list(unique_house_paths)
             for idx, house_path in enumerate(house_paths_list):
                 try:
                     enrichment = await fetch_house_catalog(house_path, cffi_session=session)
-                    save_house_catalog_enrichment(db, enrichment)
+                    house_counters = save_house_catalog_enrichment(db, enrichment)
                     counters.houses_enriched += 1
+                    hid = house_counters.get("house_id")
+                    if hid:
+                        touched_house_ids.add(int(hid))
                 except (AvitoBlockedError, AvitoRateLimitedError):
                     logger.error("pipeline:house BLOCKED at %s — propagating", house_path)
                     raise
@@ -304,7 +312,7 @@ async def run_avito_pipeline(
 
         logger.info(
             "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) "
-            "houses=%d/%d detail=%d/%d errors=%d",
+            "houses=%d/%d detail=%d/%d touched_house_ids=%d errors=%d",
             lat,
             lon,
             counters.lots_fetched,
@@ -314,9 +322,18 @@ async def run_avito_pipeline(
             counters.unique_houses,
             counters.detail_enriched,
             counters.detail_attempted,
+            len(touched_house_ids),
             len(counters.errors),
         )
-        return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n)
+        return PipelineResult(
+            lat,
+            lon,
+            radius_m,
+            counters,
+            enrich_houses,
+            enrich_detail_top_n,
+            touched_house_ids=touched_house_ids,
+        )
 
     finally:
         if own_session:
@@ -338,6 +355,9 @@ class CitySweepCounters:
     detail_attempted: int = 0
     detail_enriched: int = 0
     detail_failed: int = 0
+    imv_attempted: int = 0
+    imv_enriched: int = 0
+    imv_failed: int = 0
     errors_count: int = 0
 
     def to_dict(self) -> dict[str, int]:
@@ -354,15 +374,21 @@ async def run_avito_city_sweep(
     enrich_houses: bool = True,
     detail_top_n: int = 10,
     request_delay_sec: float = 7.0,
+    enrich_imv: bool = True,
 ) -> CitySweepCounters:
-    """Full city sweep: iterate anchors × pages → save → enrich houses + detail.
+    """Full city sweep: iterate anchors × pages → save → enrich houses + detail → IMV.
 
     - Single shared AsyncSession на весь sweep (один TLS fingerprint)
     - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
     - Прочие errors per-anchor логируются, не валят весь sweep
+    - После всех anchor'ов: если enrich_imv=True — IMV-оценка тронутых домов
+      (cooperative cancel + per-house graceful error handling)
     """
+    from app.services.house_imv_backfill import process_houses_imv_batch
+
     _anchors = anchors if anchors is not None else EKB_ANCHORS
     counters = CitySweepCounters(anchors_total=len(_anchors))
+    all_touched_house_ids: set[int] = set()
 
     async with AsyncSession(
         impersonate="chrome120",
@@ -414,6 +440,7 @@ async def run_avito_city_sweep(
                     counters.detail_enriched += result.counters.detail_enriched
                     counters.detail_failed += result.counters.detail_failed
                     counters.errors_count += len(result.counters.errors)
+                    all_touched_house_ids.update(result.touched_house_ids)
                 except (AvitoBlockedError, AvitoRateLimitedError) as e:
                     logger.error(
                         "city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s",
@@ -435,10 +462,54 @@ async def run_avito_city_sweep(
                 counters.anchors_done = idx
                 scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
 
+            # ── IMV-фаза: финальный обход тронутых домов ──────────
+            if enrich_imv and all_touched_house_ids:
+                if scrape_runs.is_cancelled(db, run_id):
+                    logger.info(
+                        "city-sweep run_id=%d: cancelled before IMV phase (%d houses skipped)",
+                        run_id,
+                        len(all_touched_house_ids),
+                    )
+                    scrape_runs.mark_done(db, run_id, counters.to_dict())
+                    return counters
+
+                logger.info(
+                    "city-sweep run_id=%d: IMV phase — %d touched houses",
+                    run_id,
+                    len(all_touched_house_ids),
+                )
+                try:
+                    imv_result = await process_houses_imv_batch(
+                        db,
+                        all_touched_house_ids,
+                        request_delay_sec=request_delay_sec,
+                    )
+                    counters.imv_attempted += imv_result.checked
+                    counters.imv_enriched += imv_result.saved
+                    counters.imv_failed += imv_result.errors
+                    counters.errors_count += imv_result.errors
+                    scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+                    logger.info(
+                        "city-sweep run_id=%d: IMV phase done — "
+                        "attempted=%d enriched=%d failed=%d",
+                        run_id,
+                        imv_result.checked,
+                        imv_result.saved,
+                        imv_result.errors,
+                    )
+                except Exception as exc:
+                    logger.error(
+                        "city-sweep run_id=%d: IMV phase crashed — %r (sweep still done)",
+                        run_id,
+                        exc,
+                    )
+                    counters.errors_count += 1
+                    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",
+                "houses=%d/%d detail=%d/%d imv=%d/%d errors=%d",
                 run_id,
                 counters.anchors_done,
                 counters.anchors_total,
@@ -449,6 +520,8 @@ async def run_avito_city_sweep(
                 counters.unique_houses,
                 counters.detail_enriched,
                 counters.detail_attempted,
+                counters.imv_enriched,
+                counters.imv_attempted,
                 counters.errors_count,
             )
             return counters
diff --git a/tradein-mvp/backend/tests/test_sweep_imv_phase.py b/tradein-mvp/backend/tests/test_sweep_imv_phase.py
new file mode 100644
index 00000000..fd6311ab
--- /dev/null
+++ b/tradein-mvp/backend/tests/test_sweep_imv_phase.py
@@ -0,0 +1,476 @@
+"""Tests for IMV enrichment phase integrated into Avito city sweep.
+
+Проверяет:
+1. enrich_imv=True — sweep вызывает IMV-фазу для touched house_id.
+2. enrich_imv=False — IMV не вызывается.
+3. IMV counters (imv_attempted, imv_enriched, imv_failed) заполняются.
+4. CitySweepCounters содержит новые imv_* поля.
+5. process_houses_imv_batch пропускает пустой набор house_ids.
+6. CitySweepStartRequest принимает enrich_imv (API-level param).
+7. Cooperative cancel перед IMV-фазой — skip IMV, sweep marked done.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import fields
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+
+# ── CitySweepCounters has imv_* fields ────────────────────────────────────────
+
+
+def test_city_sweep_counters_has_imv_fields() -> None:
+    from app.services.scrape_pipeline import CitySweepCounters
+
+    c = CitySweepCounters()
+    assert hasattr(c, "imv_attempted")
+    assert hasattr(c, "imv_enriched")
+    assert hasattr(c, "imv_failed")
+    assert c.imv_attempted == 0
+    assert c.imv_enriched == 0
+    assert c.imv_failed == 0
+
+
+def test_city_sweep_counters_to_dict_has_imv_keys() -> None:
+    from app.services.scrape_pipeline import CitySweepCounters
+
+    c = CitySweepCounters(imv_attempted=5, imv_enriched=3, imv_failed=2)
+    d = c.to_dict()
+    assert d["imv_attempted"] == 5
+    assert d["imv_enriched"] == 3
+    assert d["imv_failed"] == 2
+    # Все dataclass-поля должны быть в to_dict
+    expected_keys = {f.name for f in fields(c)}
+    assert set(d.keys()) == expected_keys
+
+
+# ── PipelineResult имеет touched_house_ids ────────────────────────────────────
+
+
+def test_pipeline_result_has_touched_house_ids() -> None:
+    from app.services.scrape_pipeline import PipelineCounters, PipelineResult
+
+    r = PipelineResult(
+        anchor_lat=56.84,
+        anchor_lon=60.6,
+        radius_m=1500,
+        counters=PipelineCounters(),
+        enrich_houses=True,
+        enrich_detail_top_n=10,
+    )
+    assert hasattr(r, "touched_house_ids")
+    assert isinstance(r.touched_house_ids, set)
+    assert len(r.touched_house_ids) == 0
+
+
+def test_pipeline_result_touched_house_ids_set() -> None:
+    from app.services.scrape_pipeline import PipelineCounters, PipelineResult
+
+    r = PipelineResult(
+        anchor_lat=56.84,
+        anchor_lon=60.6,
+        radius_m=1500,
+        counters=PipelineCounters(),
+        enrich_houses=True,
+        enrich_detail_top_n=10,
+        touched_house_ids={1, 2, 3},
+    )
+    assert r.touched_house_ids == {1, 2, 3}
+
+
+# ── enrich_imv=True → IMV phase вызывается ───────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_sweep_enrich_imv_true_calls_imv_batch() -> None:
+    """enrich_imv=True + есть touched house_ids → process_houses_imv_batch вызывается."""
+    from app.services.house_imv_backfill import HouseIMVBackfillResult
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_pipeline_result = MagicMock()
+    mock_pipeline_result.counters.lots_fetched = 5
+    mock_pipeline_result.counters.lots_inserted = 5
+    mock_pipeline_result.counters.lots_updated = 0
+    mock_pipeline_result.counters.unique_houses = 2
+    mock_pipeline_result.counters.houses_enriched = 2
+    mock_pipeline_result.counters.houses_failed = 0
+    mock_pipeline_result.counters.detail_attempted = 2
+    mock_pipeline_result.counters.detail_enriched = 2
+    mock_pipeline_result.counters.detail_failed = 0
+    mock_pipeline_result.counters.errors = []
+    mock_pipeline_result.touched_house_ids = {10, 20}
+
+    mock_imv_result = HouseIMVBackfillResult(checked=2, saved=2, skipped=0, errors=0)
+
+    with (
+        patch("app.services.scrape_runs.is_cancelled", return_value=False),
+        patch("app.services.scrape_runs.update_heartbeat"),
+        patch("app.services.scrape_runs.mark_done"),
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            new_callable=AsyncMock,
+            return_value=mock_pipeline_result,
+        ),
+        patch(
+            "app.services.house_imv_backfill.process_houses_imv_batch",
+            new_callable=AsyncMock,
+            return_value=mock_imv_result,
+        ) as mock_imv_batch,
+    ):
+        # Один anchor — минимальный sweep
+        counters = await run_avito_city_sweep(
+            mock_db,
+            run_id=1,
+            anchors=[(56.84, 60.6, "TestAnchor")],
+            enrich_imv=True,
+        )
+
+    mock_imv_batch.assert_awaited_once()
+    # Первый позиционный аргумент — db, второй — house_ids
+    call_args = mock_imv_batch.call_args
+    assert call_args.args[1] == {10, 20}
+    assert counters.imv_attempted == 2
+    assert counters.imv_enriched == 2
+    assert counters.imv_failed == 0
+
+
+# ── enrich_imv=False → IMV не вызывается ────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_sweep_enrich_imv_false_skips_imv() -> None:
+    """enrich_imv=False → process_houses_imv_batch НЕ вызывается."""
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_pipeline_result = MagicMock()
+    mock_pipeline_result.counters.lots_fetched = 3
+    mock_pipeline_result.counters.lots_inserted = 3
+    mock_pipeline_result.counters.lots_updated = 0
+    mock_pipeline_result.counters.unique_houses = 1
+    mock_pipeline_result.counters.houses_enriched = 1
+    mock_pipeline_result.counters.houses_failed = 0
+    mock_pipeline_result.counters.detail_attempted = 1
+    mock_pipeline_result.counters.detail_enriched = 1
+    mock_pipeline_result.counters.detail_failed = 0
+    mock_pipeline_result.counters.errors = []
+    mock_pipeline_result.touched_house_ids = {99}
+
+    with (
+        patch("app.services.scrape_runs.is_cancelled", return_value=False),
+        patch("app.services.scrape_runs.update_heartbeat"),
+        patch("app.services.scrape_runs.mark_done"),
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            new_callable=AsyncMock,
+            return_value=mock_pipeline_result,
+        ),
+        patch(
+            "app.services.house_imv_backfill.process_houses_imv_batch",
+            new_callable=AsyncMock,
+        ) as mock_imv_batch,
+    ):
+        counters = await run_avito_city_sweep(
+            mock_db,
+            run_id=2,
+            anchors=[(56.84, 60.6, "TestAnchor")],
+            enrich_imv=False,
+        )
+
+    mock_imv_batch.assert_not_awaited()
+    assert counters.imv_attempted == 0
+    assert counters.imv_enriched == 0
+    assert counters.imv_failed == 0
+
+
+# ── enrich_imv=True но нет touched house_ids → IMV не вызывается ────────────
+
+
+@pytest.mark.asyncio
+async def test_sweep_enrich_imv_no_touched_ids_skips() -> None:
+    """enrich_imv=True, но touched_house_ids пустой → IMV не вызывается."""
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_pipeline_result = MagicMock()
+    mock_pipeline_result.counters.lots_fetched = 0
+    mock_pipeline_result.counters.lots_inserted = 0
+    mock_pipeline_result.counters.lots_updated = 0
+    mock_pipeline_result.counters.unique_houses = 0
+    mock_pipeline_result.counters.houses_enriched = 0
+    mock_pipeline_result.counters.houses_failed = 0
+    mock_pipeline_result.counters.detail_attempted = 0
+    mock_pipeline_result.counters.detail_enriched = 0
+    mock_pipeline_result.counters.detail_failed = 0
+    mock_pipeline_result.counters.errors = []
+    mock_pipeline_result.touched_house_ids = set()  # пустой!
+
+    with (
+        patch("app.services.scrape_runs.is_cancelled", return_value=False),
+        patch("app.services.scrape_runs.update_heartbeat"),
+        patch("app.services.scrape_runs.mark_done"),
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            new_callable=AsyncMock,
+            return_value=mock_pipeline_result,
+        ),
+        patch(
+            "app.services.house_imv_backfill.process_houses_imv_batch",
+            new_callable=AsyncMock,
+        ) as mock_imv_batch,
+    ):
+        await run_avito_city_sweep(
+            mock_db,
+            run_id=3,
+            anchors=[(56.84, 60.6, "TestAnchor")],
+            enrich_imv=True,
+        )
+
+    mock_imv_batch.assert_not_awaited()
+
+
+# ── IMV-ошибки не валят sweep ────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_sweep_imv_crash_does_not_abort_sweep() -> None:
+    """Если process_houses_imv_batch raises — sweep всё равно завершается (mark_done)."""
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_pipeline_result = MagicMock()
+    mock_pipeline_result.counters.lots_fetched = 2
+    mock_pipeline_result.counters.lots_inserted = 2
+    mock_pipeline_result.counters.lots_updated = 0
+    mock_pipeline_result.counters.unique_houses = 1
+    mock_pipeline_result.counters.houses_enriched = 1
+    mock_pipeline_result.counters.houses_failed = 0
+    mock_pipeline_result.counters.detail_attempted = 1
+    mock_pipeline_result.counters.detail_enriched = 1
+    mock_pipeline_result.counters.detail_failed = 0
+    mock_pipeline_result.counters.errors = []
+    mock_pipeline_result.touched_house_ids = {55}
+
+    with (
+        patch("app.services.scrape_runs.is_cancelled", return_value=False),
+        patch("app.services.scrape_runs.update_heartbeat"),
+        patch("app.services.scrape_runs.mark_done") as mock_mark_done,
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            new_callable=AsyncMock,
+            return_value=mock_pipeline_result,
+        ),
+        patch(
+            "app.services.house_imv_backfill.process_houses_imv_batch",
+            new_callable=AsyncMock,
+            side_effect=RuntimeError("IMV network error"),
+        ),
+    ):
+        counters = await run_avito_city_sweep(
+            mock_db,
+            run_id=4,
+            anchors=[(56.84, 60.6, "TestAnchor")],
+            enrich_imv=True,
+        )
+
+    mock_mark_done.assert_called_once()
+    # IMV crash → ошибка считается, но sweep завершается как done
+    assert counters.errors_count >= 1
+
+
+# ── IMV counters накапливают failed ────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_sweep_imv_failed_counter() -> None:
+    """imv_failed отражает errors из HouseIMVBackfillResult."""
+    from app.services.house_imv_backfill import HouseIMVBackfillResult
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_pipeline_result = MagicMock()
+    mock_pipeline_result.counters.lots_fetched = 3
+    mock_pipeline_result.counters.lots_inserted = 3
+    mock_pipeline_result.counters.lots_updated = 0
+    mock_pipeline_result.counters.unique_houses = 3
+    mock_pipeline_result.counters.houses_enriched = 3
+    mock_pipeline_result.counters.houses_failed = 0
+    mock_pipeline_result.counters.detail_attempted = 0
+    mock_pipeline_result.counters.detail_enriched = 0
+    mock_pipeline_result.counters.detail_failed = 0
+    mock_pipeline_result.counters.errors = []
+    mock_pipeline_result.touched_house_ids = {1, 2, 3}
+
+    # 3 attempted, 1 enriched, 2 failed
+    mock_imv_result = HouseIMVBackfillResult(checked=3, saved=1, skipped=0, errors=2)
+
+    with (
+        patch("app.services.scrape_runs.is_cancelled", return_value=False),
+        patch("app.services.scrape_runs.update_heartbeat"),
+        patch("app.services.scrape_runs.mark_done"),
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            new_callable=AsyncMock,
+            return_value=mock_pipeline_result,
+        ),
+        patch(
+            "app.services.house_imv_backfill.process_houses_imv_batch",
+            new_callable=AsyncMock,
+            return_value=mock_imv_result,
+        ),
+    ):
+        counters = await run_avito_city_sweep(
+            mock_db,
+            run_id=5,
+            anchors=[(56.84, 60.6, "TestAnchor")],
+            enrich_imv=True,
+        )
+
+    assert counters.imv_attempted == 3
+    assert counters.imv_enriched == 1
+    assert counters.imv_failed == 2
+    # errors_count должен включить imv_failed
+    assert counters.errors_count >= 2
+
+
+# ── Cancel перед IMV-фазой ────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_sweep_cancel_before_imv_skips_imv() -> None:
+    """Если is_cancelled=True в IMV-фазе → IMV пропускается, sweep всё равно done."""
+    from app.services.scrape_pipeline import run_avito_city_sweep
+
+    mock_db = MagicMock()
+    mock_pipeline_result = MagicMock()
+    mock_pipeline_result.counters.lots_fetched = 1
+    mock_pipeline_result.counters.lots_inserted = 1
+    mock_pipeline_result.counters.lots_updated = 0
+    mock_pipeline_result.counters.unique_houses = 1
+    mock_pipeline_result.counters.houses_enriched = 1
+    mock_pipeline_result.counters.houses_failed = 0
+    mock_pipeline_result.counters.detail_attempted = 0
+    mock_pipeline_result.counters.detail_enriched = 0
+    mock_pipeline_result.counters.detail_failed = 0
+    mock_pipeline_result.counters.errors = []
+    mock_pipeline_result.touched_house_ids = {77}
+
+    # is_cancelled: первый вызов (anchor loop) = False, второй (IMV phase) = True
+    cancel_side_effects = [False, True]
+    call_idx = 0
+
+    def is_cancelled_side_effect(db, run_id):
+        nonlocal call_idx
+        val = cancel_side_effects[call_idx] if call_idx < len(cancel_side_effects) else True
+        call_idx += 1
+        return val
+
+    with (
+        patch(
+            "app.services.scrape_runs.is_cancelled",
+            side_effect=is_cancelled_side_effect,
+        ),
+        patch("app.services.scrape_runs.update_heartbeat"),
+        patch("app.services.scrape_runs.mark_done") as mock_mark_done,
+        patch(
+            "app.services.scrape_pipeline.run_avito_pipeline",
+            new_callable=AsyncMock,
+            return_value=mock_pipeline_result,
+        ),
+        patch(
+            "app.services.house_imv_backfill.process_houses_imv_batch",
+            new_callable=AsyncMock,
+        ) as mock_imv_batch,
+    ):
+        await run_avito_city_sweep(
+            mock_db,
+            run_id=6,
+            anchors=[(56.84, 60.6, "TestAnchor")],
+            enrich_imv=True,
+        )
+
+    mock_imv_batch.assert_not_awaited()
+    mock_mark_done.assert_called_once()
+
+
+# ── process_houses_imv_batch с пустым set ────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_process_houses_imv_batch_empty_ids() -> None:
+    """process_houses_imv_batch с пустым house_ids → result.checked=0, быстрый return."""
+    from app.services.house_imv_backfill import process_houses_imv_batch
+
+    mock_db = MagicMock()
+
+    result = await process_houses_imv_batch(mock_db, set(), request_delay_sec=1.0)
+
+    assert result.checked == 0
+    assert result.saved == 0
+    assert result.errors == 0
+    # DB не должна вызываться при пустом наборе
+    mock_db.execute.assert_not_called()
+
+
+# ── CitySweepStartRequest принимает enrich_imv ───────────────────────────────
+
+
+def test_city_sweep_start_request_enrich_imv_default_true() -> None:
+    """enrich_imv по умолчанию True."""
+    from app.api.v1.admin import CitySweepStartRequest
+
+    req = CitySweepStartRequest()
+    assert req.enrich_imv is True
+
+
+def test_city_sweep_start_request_enrich_imv_false() -> None:
+    """enrich_imv=False принимается."""
+    from app.api.v1.admin import CitySweepStartRequest
+
+    req = CitySweepStartRequest(enrich_imv=False)
+    assert req.enrich_imv is False
+
+
+# ── API endpoint передаёт enrich_imv в sweep ────────────────────────────────
+
+
+def test_sweep_endpoint_passes_enrich_imv(app_with_admin) -> None:
+    """POST /scrape/avito-city-sweep с enrich_imv=False → sweep вызывается без IMV."""
+    with (
+        patch("app.services.scrape_runs.create_run", return_value=42),
+        patch(
+            "app.services.scrape_pipeline.run_avito_city_sweep",
+            new_callable=AsyncMock,
+        ),
+    ):
+        r = app_with_admin.post(
+            "/api/v1/admin/scrape/avito-city-sweep",
+            json={"pages_per_anchor": 2, "enrich_imv": False},
+        )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["run_id"] == 42
+
+
+@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)