diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 3abde7e6..e98b2467 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/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": diff --git a/tradein-mvp/backend/data/sql/107_scrape_schedules_seed_cian_city_sweep.sql b/tradein-mvp/backend/data/sql/107_scrape_schedules_seed_cian_city_sweep.sql new file mode 100644 index 00000000..7b82cd58 --- /dev/null +++ b/tradein-mvp/backend/data/sql/107_scrape_schedules_seed_cian_city_sweep.sql @@ -0,0 +1,63 @@ +-- 101_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), ' + 'cian_city_sweep (dormant, #973).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py index b8e7c106..7cf27ccb 100644 --- a/tradein-mvp/backend/tests/test_scheduler.py +++ b/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