feat(scheduler): wire house_imv_backfill as scheduled source (#854)

The house_imv_backfill service (services/house_imv_backfill.backfill_house_imv)
populates house_imv_evaluations on demand but was never on a schedule, so the
table stayed ~empty on prod. Register it as a standalone scrape_schedules source:

- trigger_house_imv_backfill_run mirrors trigger_avito_detail_backfill_run /
  trigger_yandex_address_backfill_run (_claim_run -> asyncio.create_task -> fresh
  SessionLocal -> add_done_callback). Since the service has no tasks/*-wrapper
  owning the run lifecycle, the trigger finalises scrape_runs itself
  (mark_done/mark_failed) like trigger_refresh_search_matview_run, and passes a
  heartbeat callback so reap_zombies does not kill a long live IMV run (#1363).
- dispatch branch in scheduler_loop routes source=="house_imv_backfill".
- migration 132 seeds an idempotent scrape_schedules row (window 15:00-17:00 UTC,
  after the Avito sweep/full-load + geocode/cadastral enrichment blocks).

Refs #854

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

@@ -99,6 +99,17 @@ Sources:
                               Pure internal DB op — one FDW read + local UPDATE, no HTTP/anti-bot;
                               window 09:00-10:00 UTC — после geocode_missing_listings 06-09 UTC
                               чтобы listings имели свежие lat/lon/geom)
+  - house_imv_backfill      → backfill_house_imv
+                              (services/house_imv_backfill.py, #854; nightly bulk Avito IMV
+                              на уровне домов — итерирует houses imv_status='pending' с lat/lon
+                              + linked listings, считает медианные lot-params, зовёт
+                              evaluate_via_imv, наполняет house_imv_evaluations /
+                              house_placement_history / house_suggestions, переводит
+                              houses.imv_status pending→ok/not_found/error. Resumable +
+                              idempotent (UPSERT по house_id, фильтр imv_status). Avito IMV
+                              использует собственную curl_cffi-сессию; window 15:00-17:00 UTC —
+                              после avito_full_load 13-15 UTC и утреннего Avito-блока, дома
+                              уже geocoded/enriched дневными sweep'ами)
 """
 
 from __future__ import annotations
@@ -1570,6 +1581,79 @@ async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, A
     return run_id
 
 
+async def trigger_house_imv_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch backfill_house_imv в asyncio.create_task (#854).
+
+    Bulk Avito IMV-оценка на уровне домов: итерирует houses с imv_status=:only_status
+    (default 'pending'), у которых есть lat/lon + хотя бы один listing с rooms+area,
+    считает медианные lot-params, зовёт evaluate_via_imv и наполняет
+    house_imv_evaluations + house_placement_history + house_suggestions (3×UPSERT),
+    переводя houses.imv_status pending→ok/not_found/error/transient_error.
+
+    Зеркало trigger_avito_detail_backfill_run / trigger_yandex_address_backfill_run:
+    _claim_run → asyncio.create_task → fresh SessionLocal → add_done_callback (RUF006).
+    Отличие: у house_imv_backfill нет tasks/*-обёртки, владеющей lifecycle, поэтому
+    финализацию scrape_runs (mark_done/mark_failed) делаем здесь, как в
+    trigger_refresh_search_matview_run. heartbeat-колбэк дёргает update_heartbeat,
+    чтобы reap_zombies не пометил живой долгий IMV-run 'zombie' (#1363).
+
+    Resumable + idempotent: backfill_house_imv фильтрует imv_status и UPSERT'ит по
+    house_id, поэтому повторный fire дренит следующий batch без дублей.
+
+    Returns run_id или None (skip — already running).
+    """
+    run_id = _claim_run(db, schedule_row)
+    if run_id is None:
+        return None
+
+    params = schedule_row.get("default_params") or {}
+    batch_size = int(params.get("batch_size", 50))
+    request_delay_sec = float(params.get("request_delay_sec", 5.0))
+    only_status = str(params.get("only_status", "pending"))
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.services.house_imv_backfill import backfill_house_imv
+
+            # heartbeat дёргается каждые N домов внутри backfill — обновляем
+            # scrape_runs.heartbeat_at, иначе reap_zombies снимет живой run (#1363).
+            def _heartbeat() -> None:
+                runs_mod.update_heartbeat(run_db, run_id, {})
+
+            result = await backfill_house_imv(
+                run_db,
+                batch_size=batch_size,
+                request_delay_sec=request_delay_sec,
+                only_status=only_status,
+                heartbeat=_heartbeat,
+            )
+            runs_mod.mark_done(
+                run_db,
+                run_id,
+                {
+                    "checked": result.checked,
+                    "saved": result.saved,
+                    "skipped": result.skipped,
+                    "errors": result.errors,
+                    "duration_sec": int(result.duration_sec),
+                },
+            )
+        except Exception as exc:
+            logger.exception("scheduler: backfill_house_imv crashed run_id=%d", run_id)
+            try:
+                runs_mod.mark_failed(run_db, run_id, str(exc)[:1000], {})
+            except Exception:
+                logger.exception("scheduler: mark_failed crashed run_id=%d", run_id)
+        finally:
+            run_db.close()
+
+    task = asyncio.create_task(_run())
+    task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
+    logger.info("scheduler: triggered house_imv_backfill run_id=%d", run_id)
+    return run_id
+
+
 def get_due_schedules(db: Session) -> list[dict[str, Any]]:
     """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
     rows = (
@@ -1651,6 +1735,8 @@ async def scheduler_loop() -> None:
                         await trigger_yandex_detail_backfill_run(db, sch)
                     elif source == "cadastral_geo_match":
                         await trigger_cadastral_geo_match_run(db, sch)
+                    elif source == "house_imv_backfill":
+                        await trigger_house_imv_backfill_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

tradein-mvp/backend/data/sql/132_scrape_schedules_seed_house_imv.sql

@@ -0,0 +1,65 @@
+-- 132_scrape_schedules_seed_house_imv.sql
+-- Scheduler seed for the house-level Avito IMV bulk backfill (#854).
+--
+-- WHAT (source='house_imv_backfill'):
+--   trigger_house_imv_backfill_run (scheduler.py) → backfill_house_imv
+--   (services/house_imv_backfill.py). Nightly bulk IMV-оценка на уровне домов:
+--     1. SELECT houses WHERE imv_status=:only_status AND lat/lon NOT NULL
+--        AND address NOT NULL ORDER BY last_imv_attempt_at NULLS FIRST, id LIMIT batch_size.
+--     2. Для каждого дома: median lot-params из linked listings (rooms/area/floor +
+--        mode(house_type)), region-bbox address prefix, evaluate_via_imv (Avito IMV API).
+--     3. 3×UPSERT: house_imv_evaluations + house_placement_history + house_suggestions,
+--        затем UPDATE houses.imv_status pending→ok/not_found/error/transient_error.
+--   ВАЖНО: пишет в house_imv_evaluations (per-house, миграция 064) — НЕ в
+--   avito_imv_evaluations (per-cache-key, миграция 018, наполняется on-demand из estimator).
+--
+--   Resumable + idempotent: imv_status-фильтр + UPSERT по house_id, поэтому каждый fire
+--   дренит следующий batch без дублей. Avito IMV использует собственную curl_cffi-сессию
+--   (proxy через env HTTP_PROXY/HTTPS_PROXY) — request_delay_sec анти-бот пауза между домами.
+--
+-- Window 15:00-17:00 UTC (18:00-20:00 MSK):
+--   - После avito_full_load (13:00-15:00 UTC) и утреннего Avito-блока (city_sweep 02-05,
+--     detail_backfill 09-12) — не делим Avito-инфраструктуру одновременно.
+--   - После geocode_missing_listings (06-09 UTC) и cadastral_geo_match (09-10 UTC) — дома
+--     к этому моменту имеют свежие lat/lon, обязательные для IMV-геокода.
+--   - next_run_at = завтра 15:00 UTC (NOT NULL — не палим run на деплое; следуем
+--     avito/cian seed-конвенции, NULL заставил бы get_due_schedules взять на первом тике).
+--
+-- default_params:
+--   batch_size        -- домов за один прогон (50 ~ 4–7 мин при 5s/дом, дренит backlog за ночи).
+--   request_delay_sec -- анти-бот пауза между IMV-вызовами (5s; НЕ снижать < 3s — Avito IMV
+--                        реагирует на частые запросы с datacenter-IP).
+--   only_status       -- imv_status домов к обработке ('pending' по умолчанию;
+--                        'transient_error' для ретрая упавших).
+--
+-- enabled=true: безопасно — backfill resumable, batch-bounded, без DELETE/деструктива.
+--
+-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)),
+--   064_house_imv_phase_c.sql (house_imv_evaluations + house_placement_history + houses.imv_status).
+-- Idempotent: ON CONFLICT (source) DO NOTHING.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'house_imv_backfill',
+    true,
+    15,
+    17,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 15)) AT TIME ZONE 'UTC',
+    '{"batch_size": 50, "request_delay_sec": 5, "only_status": "pending"}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (replaces cron-script setup). Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), yandex_address_backfill (#855, EKB pilot), sber_index_pull (#887, monthly), rosreestr_quarter_poll (#888, monthly), cian_city_sweep (dormant, #973), yandex_newbuilding_sweep (dormant, #974), geocode_missing_listings (#1: listings geom backfill, all sources), avito_detail_backfill (#1551: nightly detail-enrichment backfill for legacy avito listings), house_imv_backfill (#854: nightly bulk Avito IMV at house level → house_imv_evaluations).';
+
+COMMIT;

tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py

@@ -0,0 +1,236 @@
+"""Tests for wiring house_imv_backfill as an in-app scheduler source (#854).
+
+The bulk IMV backfill service (services/house_imv_backfill.backfill_house_imv) is
+already built + covered elsewhere; this gate covers ONLY the scheduler wiring added
+by #854:
+  - trigger_house_imv_backfill_run shape (_claim_run → asyncio.create_task → fresh
+    SessionLocal → mark_done/mark_failed lifecycle → add_done_callback / RUF006),
+  - dispatch branch in scheduler_loop routes source=="house_imv_backfill",
+  - migration 132 seeds an idempotent scrape_schedules row with default_params.
+
+Convention mirrors tests/tasks/test_cadastral_geo_match.py: hermetic static analysis
+(inspect.getsource + migration file read) plus a fake-async behavioural test driving
+the trigger's lifecycle without a live Postgres.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import scheduler
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_132 = _SQL_DIR / "132_scrape_schedules_seed_house_imv.sql"
+
+_TRIGGER_SRC = inspect.getsource(scheduler.trigger_house_imv_backfill_run)
+
+
+# ── Scheduler wiring ──────────────────────────────────────────────────────────
+
+
+def test_scheduler_has_trigger_and_dispatch() -> None:
+    assert hasattr(scheduler, "trigger_house_imv_backfill_run")
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source == "house_imv_backfill"' in loop_src
+    assert "trigger_house_imv_backfill_run(db, sch)" in loop_src
+
+
+def test_trigger_mirrors_backfill_sibling_pattern() -> None:
+    """_claim_run gate → fresh SessionLocal → backfill_house_imv → add_done_callback."""
+    assert "_claim_run(db, schedule_row)" in _TRIGGER_SRC
+    assert "if run_id is None:" in _TRIGGER_SRC
+    # fresh session inside the spawned task (not the scheduler-tick session).
+    assert "run_db = SessionLocal()" in _TRIGGER_SRC
+    assert "backfill_house_imv" in _TRIGGER_SRC
+    # async service → bare await (NOT run_in_executor, unlike the sync DB-only triggers).
+    assert "run_in_executor" not in _TRIGGER_SRC
+    # RUF006: keep a reference + done-callback so the task is not GC'd mid-flight.
+    assert "task = asyncio.create_task(_run())" in _TRIGGER_SRC
+    assert "task.add_done_callback(" in _TRIGGER_SRC
+    assert "run_db.close()" in _TRIGGER_SRC
+
+
+def test_trigger_owns_run_lifecycle() -> None:
+    """No tasks/*-wrapper for this service → the trigger finalises scrape_runs itself."""
+    assert "runs_mod.mark_done(" in _TRIGGER_SRC
+    assert "runs_mod.mark_failed(" in _TRIGGER_SRC
+    assert "logger.exception" in _TRIGGER_SRC
+    # heartbeat callback keeps the long IMV run alive vs reap_zombies (#1363).
+    assert "heartbeat=" in _TRIGGER_SRC
+    assert "update_heartbeat" in _TRIGGER_SRC
+
+
+def test_trigger_reads_default_params() -> None:
+    """default_params drive batch_size / request_delay_sec / only_status with safe defaults."""
+    assert 'params.get("batch_size"' in _TRIGGER_SRC
+    assert 'params.get("request_delay_sec"' in _TRIGGER_SRC
+    assert 'params.get("only_status"' in _TRIGGER_SRC
+
+
+# ── Migration content ─────────────────────────────────────────────────────────
+
+
+def test_migration_132_seeds_schedule_idempotent() -> None:
+    sql = _MIGRATION_132.read_text(encoding="utf-8")
+    assert "INSERT INTO scrape_schedules" in sql
+    assert "'house_imv_backfill'" in sql
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+    assert "BEGIN;" in sql and "COMMIT;" in sql
+
+
+def test_migration_132_next_run_not_null_tomorrow() -> None:
+    """next_run_at NOT NULL = tomorrow (avoid deploy-time fire), per seed convention."""
+    sql = _MIGRATION_132.read_text(encoding="utf-8")
+    assert "CURRENT_DATE + INTERVAL '1 day'" in sql
+    assert "make_interval(hours => 15)" in sql
+
+
+def test_migration_132_default_params() -> None:
+    sql = _MIGRATION_132.read_text(encoding="utf-8")
+    assert '"batch_size": 50' in sql
+    assert '"request_delay_sec": 5' in sql
+    assert '"only_status": "pending"' in sql
+
+
+def test_migration_132_no_psycopg_v3_colon_colon_cast() -> None:
+    """psycopg v3: never :param::type — must use CAST(:param AS type)."""
+    sql = _MIGRATION_132.read_text(encoding="utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+def test_migration_132_matches_seed_column_set() -> None:
+    """Exact column set of existing scrape_schedules seeds (source..default_params)."""
+    sql = _MIGRATION_132.read_text(encoding="utf-8")
+    for col in (
+        "source",
+        "enabled",
+        "window_start_hour",
+        "window_end_hour",
+        "next_run_at",
+        "default_params",
+    ):
+        assert col in sql
+
+
+# ── Fake-async behavioural: trigger lifecycle without Postgres ────────────────
+
+
+class _FakeDB:
+    """Minimal Session stand-in: only needs to be a distinct object + close()."""
+
+    def __init__(self) -> None:
+        self.closed = 0
+
+    def close(self) -> None:
+        self.closed += 1
+
+
+def test_trigger_skips_when_claim_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
+    """If _claim_run finds a running run (None) → trigger returns None, spawns nothing."""
+    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: None)
+
+    async def _go() -> int | None:
+        return await scheduler.trigger_house_imv_backfill_run(
+            _FakeDB(),  # type: ignore[arg-type]
+            {"source": "house_imv_backfill", "default_params": {}},
+        )
+
+    assert asyncio.run(_go()) is None
+
+
+def test_trigger_claims_runs_backfill_and_marks_done(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Happy path: claim run → backfill_house_imv awaited with params → mark_done(counters)."""
+    captured: dict[str, Any] = {}
+    marked: dict[str, Any] = {}
+
+    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 77)
+    monkeypatch.setattr(scheduler, "SessionLocal", _FakeDB)
+    monkeypatch.setattr(scheduler.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(
+        scheduler.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(scheduler.runs_mod, "mark_failed", lambda *a, **k: None)
+
+    class _Result:
+        checked, saved, skipped, errors, duration_sec = 10, 6, 2, 2, 42.0
+
+    async def _fake_backfill(_db: Any, **kwargs: Any) -> _Result:
+        captured.update(kwargs)
+        return _Result()
+
+    import app.services.house_imv_backfill as _svc
+
+    monkeypatch.setattr(_svc, "backfill_house_imv", _fake_backfill)
+
+    async def _go() -> int | None:
+        run_id = await scheduler.trigger_house_imv_backfill_run(
+            _FakeDB(),  # type: ignore[arg-type]
+            {
+                "source": "house_imv_backfill",
+                "default_params": {
+                    "batch_size": 25,
+                    "request_delay_sec": 3,
+                    "only_status": "transient_error",
+                },
+            },
+        )
+        # let the spawned _run() task finish.
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+        return run_id
+
+    run_id = asyncio.run(_go())
+    assert run_id == 77
+    assert captured["batch_size"] == 25
+    assert captured["request_delay_sec"] == 3.0
+    assert captured["only_status"] == "transient_error"
+    assert callable(captured["heartbeat"])
+    assert marked["run_id"] == 77
+    assert marked["counters"]["checked"] == 10
+    assert marked["counters"]["saved"] == 6
+
+
+def test_trigger_marks_failed_on_backfill_crash(monkeypatch: pytest.MonkeyPatch) -> None:
+    """If backfill_house_imv raises → trigger marks the run failed (does not bubble)."""
+    failed: dict[str, Any] = {}
+
+    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 88)
+    monkeypatch.setattr(scheduler, "SessionLocal", _FakeDB)
+    monkeypatch.setattr(scheduler.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(scheduler.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(
+        scheduler.runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
+    )
+
+    async def _boom(_db: Any, **kwargs: Any) -> None:
+        raise RuntimeError("imv 403")
+
+    import app.services.house_imv_backfill as _svc
+
+    monkeypatch.setattr(_svc, "backfill_house_imv", _boom)
+
+    async def _go() -> None:
+        await scheduler.trigger_house_imv_backfill_run(
+            _FakeDB(),  # type: ignore[arg-type]
+            {"source": "house_imv_backfill", "default_params": {}},
+        )
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+    asyncio.run(_go())
+    assert failed["run_id"] == 88
+    assert "imv 403" in failed["err"]