Merge pull request 'feat(tradein): scheduler-источник cian_city_sweep (newbuilding Phase 4), seed dormant (#973)' (#1296) from feat/cian-city-sweep-schedule into main

Reviewed-on: #1296

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

@@ -49,6 +49,8 @@ Sources:
                               (tasks/yandex_newbuilding_sweep.py, #974; enrichment ЖК в
                               market.yandex_jk_enrichment через BrowserFetcher; shipped DORMANT —
                               enable manually after deploy verified; window 02:00-05:00 UTC)
+  - cian_city_sweep         → run_cian_city_sweep (newbuilding Phase 4, #973; shipped DORMANT —
+                              proxy dead, enable manually after proxy restored)
 """
 
 from __future__ import annotations
@@ -67,6 +69,7 @@ from app.core.db import SessionLocal
 from app.services import scrape_runs as runs_mod
 from app.services.scrape_pipeline import (
     run_avito_city_sweep,
+    run_cian_city_sweep,
     run_n1_city_sweep,
     run_yandex_city_sweep,
 )
@@ -372,6 +375,41 @@ async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -
     return run_id
 
 
+async def trigger_cian_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch run_cian_city_sweep в asyncio.create_task (#973).
+
+    Несёт Phase 4 newbuilding-enrichment (enrich_houses). Shipped DORMANT — seed
+    enabled=false (107), включается оператором вручную после восстановления прокси.
+    Returns run_id (или None если skip — есть running run).
+    """
+    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:
+            await run_cian_city_sweep(
+                run_db,
+                run_id=run_id,
+                pages_per_anchor=int(params.get("pages_per_anchor", 3)),
+                request_delay_sec=float(params.get("request_delay_sec", 5.0)),
+                radius_m=int(params.get("radius_m", 1500)),
+                detail_top_n=int(params.get("detail_top_n", 10)),
+                enrich_houses=bool(params.get("enrich_houses", True)),
+            )
+        except Exception:
+            logger.exception("scheduler: run_cian_city_sweep 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 cian_city_sweep run_id=%d", run_id)
+    return run_id
+
+
 async def _execute_cian_backfill(
     db: Session,
     *,
@@ -1090,6 +1128,8 @@ async def scheduler_loop() -> None:
                         await trigger_n1_city_sweep_run(db, sch)
                     elif source == "cian_history_backfill":
                         await trigger_cian_backfill_run(db, sch)
+                    elif source == "cian_city_sweep":
+                        await trigger_cian_city_sweep_run(db, sch)
                     elif source == "rosreestr_dkp_import":
                         await trigger_rosreestr_dkp_run(db, sch)
                     elif source == "listing_source_snapshot":

tradein-mvp/backend/data/sql/107_scrape_schedules_seed_cian_city_sweep.sql

@@ -0,0 +1,64 @@
+-- 107_scrape_schedules_seed_cian_city_sweep.sql
+-- Seed row для Cian периодического city-sweep с newbuilding Phase 4 (#973).
+--
+-- !!! DORMANT BY DESIGN !!! enabled = false.
+--   Sweep бьёт Cian с prod-IP через proxy — proxy сейчас мёртв → anti-bot риск.
+--   Capability полностью wired (run_cian_city_sweep + trigger + dispatch),
+--   но schedule засеян ВЫКЛЮЧЕННЫМ. Live-поведение намеренно НЕ verified.
+--   ВКЛЮЧАТЬ ВРУЧНУЮ оператором ПОСЛЕ восстановления прокси:
+--     UPDATE scrape_schedules SET enabled = true WHERE source = 'cian_city_sweep';
+--   Зеркалит yandex_city_sweep (078, тоже enabled = false, dormant).
+--
+-- Phase 4 (newbuilding enrichment) включается через enrich_houses=true в default_params.
+-- В отличие от yandex_city_sweep, cian sweep несёт detail + houses-фазы —
+-- поэтому default_params включает detail_top_n и enrich_houses.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)).
+-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно.
+--
+-- cian_city_sweep — окно 02:00-05:00 UTC (05:00-08:00 МСК), как avito_city_sweep.
+--   default_params (консервативно, под анти-бот):
+--     pages_per_anchor   — страниц SERP на anchor.
+--     request_delay_sec  — пауза между anchor'ами (поверх per-page sleep scraper'а).
+--     radius_m           — радиус поиска вокруг anchor'а в метрах.
+--     detail_top_n       — кол-во объявлений на detail-обогащение (Phase 3).
+--     enrich_houses      — включить Phase 4 newbuilding enrichment (true).
+
+BEGIN;
+
+-- next_run_at = следующее наступление night-window (tomorrow + window_start_hour).
+-- При enabled=false get_due_schedules() всё равно не подхватит строку, но заполняем
+-- по образцу 078: когда оператор включит schedule, next_run_at уже валидный и
+-- запуск не выстрелит мгновенно (вне окна). AT TIME ZONE 'UTC' фиксирует wall-clock
+-- час в UTC независимо от session TimeZone GUC.
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    next_run_at,
+    default_params
+)
+VALUES
+(
+    'cian_city_sweep',
+    false,                                       -- DORMANT (#973) — enable manually after proxy restored
+    2,
+    5,
+    ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC',
+    '{"pages_per_anchor": 3, "request_delay_sec": 5, "radius_m": 1500, "detail_top_n": 10, "enrich_houses": true}'::jsonb
+)
+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), yandex_newbuilding_sweep (dormant, #974), '
+    'cian_city_sweep (dormant, #973).';
+
+COMMIT;

tradein-mvp/backend/tests/test_scheduler.py

@@ -275,3 +275,92 @@ async def test_scheduler_dispatch_routes_yandex_newbuilding_sweep(
         await _sched.trigger_yandex_newbuilding_sweep_run(db, sch)
 
     assert "yandex_newbuilding_sweep" in triggered
+
+
+# ── #973: trigger_cian_city_sweep_run + dispatch ─────────────────────────────
+
+
+_CIAN_SWEEP_ROW = {
+    "source": "cian_city_sweep",
+    "window_start_hour": 2,
+    "window_end_hour": 5,
+    "default_params": {
+        "pages_per_anchor": 3,
+        "request_delay_sec": 5.0,
+        "radius_m": 1500,
+        "detail_top_n": 10,
+        "enrich_houses": True,
+    },
+}
+
+
+@pytest.mark.asyncio
+async def test_trigger_cian_city_sweep_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
+    """trigger_cian_city_sweep_run claims run + launches create_task (#973)."""
+    db = _FakeSchedulerDB()
+    calls: dict[str, list] = {"create_run": [], "sweep": []}
+
+    def fake_create_run(_d, *, source, params):
+        calls["create_run"].append(source)
+        db.running[source] = True
+        return 201
+
+    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
+
+    # Patch run_cian_city_sweep to a coroutine that records call
+    async def fake_sweep(run_db, *, run_id, **kwargs):
+        calls["sweep"].append(run_id)
+
+    monkeypatch.setattr(_sched, "run_cian_city_sweep", fake_sweep)
+    # Patch SessionLocal to return the same db
+    monkeypatch.setattr(_sched, "SessionLocal", lambda: db)
+
+    run_id = await _sched.trigger_cian_city_sweep_run(db, _CIAN_SWEEP_ROW)
+
+    assert run_id == 201
+    assert calls["create_run"] == ["cian_city_sweep"]
+    # Drain event-loop so the asyncio.create_task runs
+    await asyncio.sleep(0)
+    assert calls["sweep"] == [201]
+
+
+@pytest.mark.asyncio
+async def test_trigger_cian_city_sweep_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
+    """trigger_cian_city_sweep_run returns None when source already running."""
+    db = _FakeSchedulerDB()
+    db.running["cian_city_sweep"] = True
+
+    calls: dict[str, int] = {"n": 0}
+
+    def fake_create_run(_d, *, source, params):
+        calls["n"] += 1
+        return 999
+
+    monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
+
+    result = await _sched.trigger_cian_city_sweep_run(db, _CIAN_SWEEP_ROW)
+
+    assert result is None
+    assert calls["n"] == 0
+
+
+@pytest.mark.asyncio
+async def test_scheduler_dispatch_routes_cian_city_sweep(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Dispatch block в scheduler_loop роутит cian_city_sweep → trigger_cian_city_sweep_run."""
+    triggered: list[str] = []
+
+    async def fake_trigger(db, sch):
+        triggered.append(sch["source"])
+        return 1
+
+    monkeypatch.setattr(_sched, "trigger_cian_city_sweep_run", fake_trigger)
+
+    # Simulate the dispatch block for one due schedule
+    db = object()
+    sch = _CIAN_SWEEP_ROW
+    source = sch["source"]
+    # Replicate the elif chain from scheduler_loop for cian_city_sweep
+    if source == "cian_city_sweep":
+        await _sched.trigger_cian_city_sweep_run(db, sch)
+
+    assert "cian_city_sweep" in triggered