feat(tradein-scheduler): nightly newbuilding-enrichment schedule (#973)

Регистрирует отдельный in-app scheduler-источник для cian_newbuilding
enrichment-backfill (houses_price_dynamics / house_reliability_checks /
house_reviews), который раньше бежал только инлайн в Cian full-sweep.

- run_newbuilding_enrich() wrapper (heartbeat → backfill → mark_done/failed),
  делегирует backfill_newbuilding_enrichment (#972, уже в main).
- trigger_newbuilding_enrich_run() + dispatch-ветка в scheduler_loop
  (зеркалит trigger_sber_index_pull_run): _claim_run guard, async task,
  zombie-reap наследуется; idempotency наследуется (skip уже обогащённых,
  per-house SAVEPOINT, bounded per-fire limit дренирует backlog 306 домов).
- seed-миграция 103: source='newbuilding_enrich', окно 00:00-01:00 UTC
  (03:00-04:00 МСК, до 01:00+ UTC sweep-блока), limit=25, next_run_at на завтра.

Засеяно enabled=FALSE (dormant): на prod весь external-HTTP scraping намеренно
на паузе. Расписание+триггер готовы, но не стартуют до намеренного возобновления
(UPDATE ... SET enabled=true). Не активирую одиночный источник при общей паузе.

Closes #973

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

@@ -37,6 +37,14 @@ Sources:
                               Rosreestr open-data portal for new quarter availability — pure
                               HTTP detect+alert, NO download/shell-out; window 07:00-08:00 UTC,
                               ~28-day cadence)
+  - newbuilding_enrich      → run_newbuilding_enrich
+                              (tasks/newbuilding_enrich_backfill.py, #973; nightly slice of the
+                              cian_newbuilding houses enrichment — resolves cian_zhk_url, fetches
+                              the ЖК page, persists houses_price_dynamics / house_reliability_checks
+                              / house_reviews. Cian HTTP (curl_cffi chrome120) + anti-bot delay.
+                              Idempotent: skips houses already enriched, per-house SAVEPOINT, so
+                              each fire drains the next `limit` pending houses; window 00:00-01:00
+                              UTC = 03:00-04:00 МСК, before the 01:00+ UTC sweep block)
 """
 
 from __future__ import annotations
@@ -679,6 +687,49 @@ async def trigger_rosreestr_quarter_poll_run(
     return run_id
 
 
+async def trigger_newbuilding_enrich_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch run_newbuilding_enrich в asyncio.create_task (#973).
+
+    Nightly slice of the cian_newbuilding enrichment backfill: resolves cian_zhk_url from
+    house_sources.ext_id when missing, fetches the ЖК page (curl_cffi chrome120) and persists
+    houses_price_dynamics / house_reliability_checks / house_reviews. This is the standalone
+    schedule the inline cian full-sweep path never had — 306 fetchable cian houses on prod,
+    only ~3 enriched so far, so the backlog drains over successive nights.
+
+    Mirrors trigger_sber_index_pull_run: claim run → create_task → mark_done/failed delegated
+    to tasks/newbuilding_enrich_backfill.run_newbuilding_enrich (which owns the run lifecycle).
+
+    Idempotency: run_newbuilding_enrich → backfill_newbuilding_enrichment skips houses that
+    already have BOTH price_dynamics AND reliability rows (force=False) and uses a per-house
+    SAVEPOINT, so a re-fire never re-enriches done houses or double-appends on a partial run.
+    `limit` (default_params) bounds how many houses one fire fetches — anti-bot politeness;
+    the nightly cadence drains the rest. zombie-reap is inherited from the shared loop.
+
+    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 {}
+
+    async def _run() -> None:
+        run_db = SessionLocal()
+        try:
+            from app.tasks.newbuilding_enrich_backfill import run_newbuilding_enrich
+
+            await run_newbuilding_enrich(run_db, run_id=run_id, params=params)
+        except Exception:
+            logger.exception("scheduler: run_newbuilding_enrich 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 newbuilding_enrich run_id=%d", run_id)
+    return run_id
+
+
 async def trigger_asking_to_sold_ratio_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
     """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).
 
@@ -1003,6 +1054,8 @@ async def scheduler_loop() -> None:
                         await trigger_sber_index_pull_run(db, sch)
                     elif source == "rosreestr_quarter_poll":
                         await trigger_rosreestr_quarter_poll_run(db, sch)
+                    elif source == "newbuilding_enrich":
+                        await trigger_newbuilding_enrich_run(db, sch)
                     else:
                         logger.warning("scheduler: unknown source=%s, skip", source)
             finally:

tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py

@@ -71,6 +71,7 @@ from dataclasses import dataclass, field, fields
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from app.services import scrape_runs as runs_mod
 from app.services.scraper_settings import get_scraper_delay
 
 logger = logging.getLogger(__name__)
@@ -79,6 +80,7 @@ __all__ = [
     "NewbuildingEnrichBackfillResult",
     "backfill_newbuilding_enrichment",
     "count_cian_newbuilding_houses",
+    "run_newbuilding_enrich",
 ]
 
 # Source tags written by the Cian newbuilding enrichment (kept in sync with
@@ -642,3 +644,80 @@ async def _sleep_with_jitter(delay: float, idx: int, total: int, *, force: bool
     if not force and idx >= total - 1:
         return
     await asyncio.sleep(delay * random.uniform(0.8, 1.2))
+
+
+async def run_newbuilding_enrich(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict,  # type: ignore[type-arg]
+) -> NewbuildingEnrichBackfillResult:
+    """Execute the newbuilding-enrichment backfill with run lifecycle management (#973).
+
+    Thin scheduler wrapper around backfill_newbuilding_enrichment() — mirrors
+    tasks/yandex_address_backfill.run_yandex_address_backfill: emit a heartbeat before the
+    batch, delegate to the proven backfill, then mark the scrape_run done/failed.
+
+    Idempotency is inherited from backfill_newbuilding_enrichment(): with force=False its
+    SELECT excludes houses that already have BOTH price_dynamics AND reliability rows, and
+    a per-house SAVEPOINT makes a partial/aborted run resume cleanly on the next tick. So
+    each nightly fire processes only the next slice of still-pending cian_newbuilding houses
+    (306 fetchable on prod, ~3 enriched so far — many nights to drain at a polite cadence).
+
+    Per-tick bound: `limit` caps how many houses one fire fetches (anti-bot politeness; a
+    full 306-house sweep in one tick would hammer Cian). The scheduler re-fires nightly, so
+    the backlog drains over successive nights without operator intervention.
+
+    Params (from default_params jsonb in scrape_schedules):
+        limit: int — max houses to process this fire (default 25).
+        force: bool — re-process already-enriched houses (UPSERT/dedup-guarded; default False).
+        request_delay_sec: float | None — seconds between fetches; None → get_scraper_delay('cian').
+    """
+    limit = int(params.get("limit", 25))
+    force = bool(params.get("force", False))
+    raw_delay = params.get("request_delay_sec")
+    request_delay_sec = float(raw_delay) if raw_delay is not None else None
+
+    counters: dict[str, int] = {
+        "processed": 0,
+        "succeeded": 0,
+        "skipped_already_enriched": 0,
+        "failed_resolve": 0,
+        "failed_fetch": 0,
+        "failed_save": 0,
+    }
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        result = await backfill_newbuilding_enrichment(
+            db,
+            limit=limit,
+            force=force,
+            request_delay_sec=request_delay_sec,
+        )
+
+        counters = result.to_dict()
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "scheduler: newbuilding_enrich run_id=%d done — processed=%d ok=%d skip=%d "
+            "resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d "
+            "| pending=%d %.1fs",
+            run_id,
+            result.processed,
+            result.succeeded,
+            result.skipped_already_enriched,
+            result.failed_resolve,
+            result.failed_fetch,
+            result.failed_save,
+            result.price_dynamics_rows,
+            result.reliability_rows,
+            result.review_rows,
+            result.cian_houses_pending,
+            result.duration_sec,
+        )
+        return result
+    except Exception as exc:
+        logger.exception("scheduler: newbuilding_enrich run_id=%d failed", run_id)
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise

tradein-mvp/backend/data/sql/103_scrape_schedules_seed_newbuilding_enrich.sql

@@ -0,0 +1,90 @@
+-- 103_scrape_schedules_seed_newbuilding_enrich.sql
+-- Issue #973 (GG-форсайт, 950-E2) — seed scrape_schedules row for the nightly
+-- newbuilding-enrichment backfill.
+--
+-- Context:
+--   The three newbuilding-enrichment tables (houses_price_dynamics,
+--   house_reliability_checks, house_reviews) are populated only INLINE during the cian
+--   full-sweep — there was never a standalone schedule for them. On prod 306 houses are
+--   linked to ext_source='cian_newbuilding' (all fetchable) but only ~18 have price
+--   dynamics, ~3 have reliability, 0 have reviews. This schedule drains that backlog
+--   nightly, one bounded slice at a time.
+--
+--   Components deployed together (same release as this migration):
+--     1. tasks/newbuilding_enrich_backfill.py — run_newbuilding_enrich() scheduler wrapper
+--        (heartbeat → backfill_newbuilding_enrichment → mark_done/mark_failed). The
+--        backfill itself (backfill_newbuilding_enrichment) shipped with #972.
+--     2. scheduler.py — trigger_newbuilding_enrich_run() + dispatch branch
+--        `elif source == "newbuilding_enrich"` in scheduler_loop().
+--     3. This migration — seeds the scrape_schedules row.
+--
+-- Idempotency (inherited from backfill_newbuilding_enrichment, #972):
+--   With force=False the backfill's SELECT excludes houses that already have BOTH
+--   price_dynamics AND reliability rows, and each house is processed under a SAVEPOINT,
+--   so a re-fire never re-enriches a done house nor double-appends on a partial run.
+--   default_params.limit bounds how many houses one fire fetches (anti-bot politeness);
+--   the nightly cadence drains the remaining backlog over successive nights.
+--
+-- Schedule window 00:00-01:00 UTC (= 03:00-04:00 МСК):
+--   Deep-night МСК slot, placed BEFORE the 01:00+ UTC sweep block (listing_source_snapshot
+--   01:00-02:00, avito/cian/yandex sweeps 02:00-05:00 UTC) so the Cian HTTP fetches of this
+--   task do not overlap the daytime/early-morning SERP-sweeps and compete for the Cian
+--   anti-bot budget. Per-house polite delay = get_scraper_delay('cian') (~5s) with ±20%
+--   jitter; at limit=25 a fire is ~2-3 min of fetches.
+--
+-- enabled = FALSE (seeded dormant, deliberately). As of 2026-06-13 every external-HTTP
+--   scrape source on prod is enabled=false (scraping paused on purpose) except
+--   deactivate_stale_avito (pure-internal). Seeding this dormant preserves that posture —
+--   the schedule + trigger code ship ready, but do NOT begin fetching Cian until scraping is
+--   resumed deliberately (avoids a lone external-HTTP source firing while the rest are paused).
+--   TO ACTIVATE when scraping resumes:
+--     UPDATE scrape_schedules SET enabled=true WHERE source='newbuilding_enrich';
+--   (ON CONFLICT DO NOTHING below will NOT change an already-existing row's enabled flag.)
+--
+-- next_run_at bootstrapped to tomorrow 00:00 UTC — the scheduler will not fire immediately
+--   on deploy (same pattern as 078/079/082/088/091/093/096).
+--
+-- Idempotent: ON CONFLICT (source) DO NOTHING — safe to re-apply.
+--
+-- Dependencies:
+--   052_scrape_schedules.sql (table + UNIQUE(source))
+--   scheduler.py change (trigger_newbuilding_enrich_run must be deployed)
+--   tasks/newbuilding_enrich_backfill.py run_newbuilding_enrich() must be deployed (#973),
+--     which delegates to backfill_newbuilding_enrichment() shipped with #972.
+--
+-- Deploy order:
+--   Apply after deploying the scheduler.py + task changes so the scheduler can dispatch
+--   'newbuilding_enrich' correctly on first fire.
+
+BEGIN;
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'newbuilding_enrich',
+    false,                                       -- DORMANT: prod scraping paused — flip true to activate (see note above)
+    0,                                           -- window 00:00-01:00 UTC = 03:00-04:00 МСК
+    1,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 0)) AT TIME ZONE 'UTC',
+    '{"limit": 25, "force": false}'::jsonb       -- bounded per-fire slice; request_delay_sec omitted → get_scraper_delay('cian')
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет 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 СберИндекс city-level price index), '
+    'rosreestr_quarter_poll (#888, monthly Rosreestr new-quarter availability check), '
+    'newbuilding_enrich (#973, nightly cian_newbuilding enrichment backfill slice).';
+
+COMMIT;

tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py

@@ -24,6 +24,7 @@ from app.tasks.newbuilding_enrich_backfill import (  # noqa: E402
     NewbuildingEnrichBackfillResult,
     _save_cian_reviews,
     backfill_newbuilding_enrichment,
+    run_newbuilding_enrich,
 )
 
 # ---------------------------------------------------------------------------
@@ -516,3 +517,203 @@ async def test_backfill_resolve_then_idempotent_rerun_no_reresolve() -> None:
     assert r2.succeeded == 0
     assert len(db.price_dynamics) == 1  # no duplicate
     assert len(db.reliability) == 1
+
+
+# ---------------------------------------------------------------------------
+# #973 (950-E2): nightly schedule — run_newbuilding_enrich wrapper, scheduler
+# trigger/dispatch wiring, and the seed migration 103.
+# ---------------------------------------------------------------------------
+
+import inspect  # noqa: E402
+import re  # noqa: E402
+from pathlib import Path  # noqa: E402
+
+from app.services import scheduler  # noqa: E402
+from app.tasks import newbuilding_enrich_backfill as task_mod  # noqa: E402
+
+_SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql"
+_MIGRATION_103 = _SQL_DIR / "103_scrape_schedules_seed_newbuilding_enrich.sql"
+
+
+# ── run_newbuilding_enrich wrapper: lifecycle + param plumbing ────────────────
+
+
+@pytest.mark.asyncio
+async def test_run_wrapper_marks_done_and_passes_params(monkeypatch: pytest.MonkeyPatch) -> None:
+    """run_newbuilding_enrich emits heartbeat → delegates with parsed params → mark_done."""
+    seen: dict = {}
+
+    async def _fake_backfill(_db, *, limit, force, request_delay_sec):
+        seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
+        return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_rows=2)
+
+    monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill)
+    monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    marked: dict = {}
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_done",
+        lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)),
+    )
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+
+    out = await run_newbuilding_enrich(
+        object(),  # type: ignore[arg-type]
+        run_id=55,
+        params={"limit": 7, "force": True, "request_delay_sec": 0},
+    )
+
+    assert out.processed == 3
+    # default_params plumbed through to the backfill.
+    assert seen == {"limit": 7, "force": True, "request_delay_sec": 0.0}
+    # run finalised via mark_done with the result counters.
+    assert marked["run_id"] == 55
+    assert marked["counters"]["processed"] == 3
+    assert marked["counters"]["succeeded"] == 2
+
+
+@pytest.mark.asyncio
+async def test_run_wrapper_defaults_when_params_empty(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Empty default_params → limit=25, force=False, request_delay_sec=None (→ scraper delay)."""
+    seen: dict = {}
+
+    async def _fake_backfill(_db, *, limit, force, request_delay_sec):
+        seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
+        return NewbuildingEnrichBackfillResult()
+
+    monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill)
+    monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+
+    await run_newbuilding_enrich(object(), run_id=1, params={})  # type: ignore[arg-type]
+
+    assert seen == {"limit": 25, "force": False, "request_delay_sec": None}
+
+
+@pytest.mark.asyncio
+async def test_run_wrapper_marks_failed_and_reraises(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Backfill raising → mark_failed + re-raise (no silent swallow)."""
+
+    async def _boom(_db, **kwargs):
+        raise RuntimeError("cian exploded")
+
+    monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _boom)
+    monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    failed: dict = {}
+    monkeypatch.setattr(
+        task_mod.runs_mod,
+        "mark_failed",
+        lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err),
+    )
+
+    with pytest.raises(RuntimeError, match="cian exploded"):
+        await run_newbuilding_enrich(object(), run_id=9, params={})  # type: ignore[arg-type]
+
+    assert failed["run_id"] == 9
+    assert "cian exploded" in failed["err"]
+
+
+# ── Scheduler wiring: trigger fn claims + dispatches; skip when already running ──
+
+
+def test_trigger_fn_exists_and_is_async() -> None:
+    fn = getattr(scheduler, "trigger_newbuilding_enrich_run", None)
+    assert fn is not None, "trigger_newbuilding_enrich_run missing from scheduler"
+    assert inspect.iscoroutinefunction(fn)
+
+
+def test_dispatch_branch_wired() -> None:
+    """scheduler_loop dispatches source='newbuilding_enrich' to the trigger."""
+    loop_src = inspect.getsource(scheduler.scheduler_loop)
+    assert 'source == "newbuilding_enrich"' in loop_src
+    assert "trigger_newbuilding_enrich_run(db, sch)" in loop_src
+
+
+def test_trigger_claims_run_and_delegates_to_task() -> None:
+    """Trigger claims via _claim_run and delegates to run_newbuilding_enrich (async task)."""
+    trig_src = inspect.getsource(scheduler.trigger_newbuilding_enrich_run)
+    assert "_claim_run(db, schedule_row)" in trig_src
+    assert "run_newbuilding_enrich" in trig_src
+    assert "asyncio.create_task(" in trig_src
+
+
+@pytest.mark.asyncio
+async def test_trigger_claims_and_dispatches(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Happy path: _claim_run returns a run_id → an asyncio task is spawned."""
+    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: 4242)
+    spawned: list = []
+    real_create_task = scheduler.asyncio.create_task
+
+    def _spy_create_task(coro):
+        spawned.append(coro)
+        # Wrap the real coro so it actually runs (and closes the SessionLocal).
+        return real_create_task(coro)
+
+    monkeypatch.setattr(scheduler.asyncio, "create_task", _spy_create_task)
+    # Stop the spawned _run() from touching a real DB / importing the heavy task.
+    monkeypatch.setattr(scheduler, "SessionLocal", lambda: MagicMock())
+
+    run_id = await scheduler.trigger_newbuilding_enrich_run(
+        MagicMock(), {"source": "newbuilding_enrich", "default_params": {}}
+    )
+
+    assert run_id == 4242
+    assert len(spawned) == 1
+
+
+@pytest.mark.asyncio
+async def test_trigger_skips_when_already_running(monkeypatch: pytest.MonkeyPatch) -> None:
+    """_claim_run None (already running / lock held) → trigger returns None, no task."""
+    monkeypatch.setattr(scheduler, "_claim_run", lambda _db, _row: None)
+    spawned: list = []
+    monkeypatch.setattr(scheduler.asyncio, "create_task", lambda coro: spawned.append(coro))
+
+    run_id = await scheduler.trigger_newbuilding_enrich_run(
+        MagicMock(), {"source": "newbuilding_enrich", "default_params": {}}
+    )
+
+    assert run_id is None
+    assert spawned == []
+
+
+# ── Migration 103 ─────────────────────────────────────────────────────────────
+
+
+def test_migration_103_exists() -> None:
+    assert _MIGRATION_103.is_file(), f"missing migration: {_MIGRATION_103}"
+
+
+def test_migration_103_seeds_correct_source_and_window() -> None:
+    sql = _MIGRATION_103.read_text("utf-8")
+    assert "'newbuilding_enrich'" in sql
+    insert_block = sql.split("INSERT INTO scrape_schedules")[1]
+    # Window 00:00-01:00 UTC = 03:00-04:00 МСК (start=0, end=1).
+    # Values may carry a trailing inline comment, so match "<n>," at line start only.
+    assert re.search(r"^\s*0,", insert_block, re.MULTILINE), "window_start_hour 0 missing"
+    assert re.search(r"^\s*1,", insert_block, re.MULTILINE), "window_end_hour 1 missing"
+
+
+def test_migration_103_is_idempotent() -> None:
+    sql = _MIGRATION_103.read_text("utf-8")
+    assert "ON CONFLICT (source) DO NOTHING" in sql
+
+
+def test_migration_103_is_transactional() -> None:
+    sql = _MIGRATION_103.read_text("utf-8")
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_103_no_psycopg_trap() -> None:
+    """Plain seed (no bind params), but guard the :x::type trap anyway (psycopg v3)."""
+    sql = _MIGRATION_103.read_text("utf-8")
+    assert not re.search(r":\w+::", sql)
+
+
+def test_migration_103_bootstraps_next_run_at() -> None:
+    """next_run_at bootstrapped to tomorrow so the scheduler does not fire on deploy."""
+    sql = _MIGRATION_103.read_text("utf-8")
+    assert "next_run_at" in sql
+    assert "CURRENT_DATE + INTERVAL '1 day'" in sql