From 3f1925052b4e72c4efb061ebb947430bdbf20db1 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 26 Jun 2026 07:45:08 +0000 Subject: [PATCH] feat(scheduler): wire house_imv_backfill as scheduled source (#854) (#1910) wire house_imv_backfill as scheduled source (#854): trigger fn + seed migration 132. Refs #854 --- tradein-mvp/backend/app/services/scheduler.py | 86 +++++++ .../132_scrape_schedules_seed_house_imv.sql | 65 +++++ .../test_house_imv_backfill_scheduler.py | 236 ++++++++++++++++++ 3 files changed, 387 insertions(+) create mode 100644 tradein-mvp/backend/data/sql/132_scrape_schedules_seed_house_imv.sql create mode 100644 tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 480d4caf..17f3c7ae 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/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: diff --git a/tradein-mvp/backend/data/sql/132_scrape_schedules_seed_house_imv.sql b/tradein-mvp/backend/data/sql/132_scrape_schedules_seed_house_imv.sql new file mode 100644 index 00000000..6617024d --- /dev/null +++ b/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; diff --git a/tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py b/tradein-mvp/backend/tests/test_house_imv_backfill_scheduler.py new file mode 100644 index 00000000..4606ac41 --- /dev/null +++ b/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"]