From 158a24c25b027586dc48e313769e1c2ab8ba8594 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 13:25:23 +0300 Subject: [PATCH] feat(tradein): schedule cian_full_load (exhaustive secondary) + anchor sweep NB-only MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Подключает prod-проверенный run_cian_full_load (exhaustive региональный сбор Cian ЕКБ вторички, room×price партиционирование, incremental on_bucket save) в in-app scheduler как recurring-источник cian_full_load (окно 20-22 UTC — отделено от cian_city_sweep/history_backfill на 2-5 чтобы не конкурировать на shared kf-прокси). run_cian_city_sweep теперь NB-only (newbuilding_only=True default): SERP-фаза сохраняет только novostroyki-лоты, вторичку отбрасывает — ею авторитетно владеет full_load. Убирает дубль secondary SERP-сейва. DETAIL/HOUSES-фазы не затронуты. --- tradein-mvp/backend/app/services/scheduler.py | 49 ++++++++- .../backend/app/services/scrape_pipeline.py | 32 +++++- ...6_scrape_schedules_seed_cian_full_load.sql | 53 +++++++++ .../backend/tests/test_cian_city_sweep.py | 101 +++++++++++++++++- tradein-mvp/backend/tests/test_scheduler.py | 95 ++++++++++++++++ 5 files changed, 322 insertions(+), 8 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/126_scrape_schedules_seed_cian_full_load.sql diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 1abc40b0..92f0abed 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -54,7 +54,14 @@ Sources: 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) + proxy dead, enable manually after proxy restored. NB-only с #973+: + SERP-фаза сохраняет только novostroyki-лоты — вторичкой владеет + cian_full_load. DETAIL/HOUSES-фазы не затронуты) + - cian_full_load → run_cian_full_load (exhaustive региональный сбор Cian ЕКБ ВТОРИЧКИ + без anchor'ов — room×price адаптивное партиционирование, + secondary_only, incremental on_bucket save; авторитетный источник + вторички. Окно 20:00-22:00 UTC чтобы не конфликтовать на shared + kf-прокси с cian_city_sweep (2-5) и cian_history_backfill (2-5)) - domclick_city_sweep → run_domclick_city_sweep (scrape_pipeline.py; citywide SERP via camoufox BrowserFetcher; shipped DORMANT enabled=false, enable manually after prod-smoke) @@ -99,6 +106,7 @@ from app.services.scrape_pipeline import ( run_avito_city_sweep, run_avito_newbuilding_sweep, run_cian_city_sweep, + run_cian_full_load, run_domclick_city_sweep, run_yandex_city_sweep, ) @@ -472,6 +480,7 @@ async def trigger_cian_city_sweep_run(db: Session, schedule_row: dict[str, Any]) 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)), + newbuilding_only=bool(params.get("newbuilding_only", True)), ) except Exception: logger.exception("scheduler: run_cian_city_sweep crashed run_id=%d", run_id) @@ -484,6 +493,42 @@ async def trigger_cian_city_sweep_run(db: Session, schedule_row: dict[str, Any]) return run_id +async def trigger_cian_full_load_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_cian_full_load в asyncio.create_task. + + Exhaustive региональный сбор Cian ЕКБ ВТОРИЧКИ (room×price партиционирование, + secondary_only, incremental on_bucket save) — авторитетный источник вторички. + Образец trigger_cian_city_sweep_run. 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_full_load( + run_db, + run_id=run_id, + price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)), + concurrency=int(params.get("concurrency", 5)), + request_delay_sec=float(params.get("request_delay_sec", 4.0)), + enrich_detail=bool(params.get("enrich_detail", False)), + detail_top_n=int(params.get("detail_top_n", 0)), + resume_run_id=None, + ) + except Exception: + logger.exception("scheduler: run_cian_full_load 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_full_load run_id=%d", run_id) + return run_id + + async def trigger_domclick_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch run_domclick_city_sweep в asyncio.create_task. @@ -1451,6 +1496,8 @@ async def scheduler_loop() -> None: await trigger_cian_backfill_run(db, sch) elif source == "cian_city_sweep": await trigger_cian_city_sweep_run(db, sch) + elif source == "cian_full_load": + await trigger_cian_full_load_run(db, sch) elif source == "domclick_city_sweep": await trigger_domclick_city_sweep_run(db, sch) elif source == "rosreestr_dkp_import": diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 29c5f99a..d9fd695f 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -1534,6 +1534,7 @@ class CianCitySweepCounters: anchors_total: int = 0 anchors_done: int = 0 lots_fetched: int = 0 + lots_dropped_secondary: int = 0 # вторичка, отброшенная при newbuilding_only=True lots_inserted: int = 0 lots_updated: int = 0 detail_attempted: int = 0 @@ -1563,18 +1564,27 @@ async def run_cian_city_sweep( request_delay_sec: float = 5.0, detail_top_n: int = 10, enrich_houses: bool = True, + newbuilding_only: bool = True, ) -> CianCitySweepCounters: - """Cian full city sweep: SERP → detail(+price-history) → newbuilding/houses. + """Cian newbuilding city sweep: SERP → detail(+price-history) → newbuilding/houses. Симметрично run_avito_city_sweep. Использует EKB_ANCHORS (общий список). + newbuilding_only (default True): в SERP-фазе сохраняются только лоты с + listing_segment == "novostroyki" — вторичку отбрасываем, ею авторитетно владеет + run_cian_full_load (exhaustive региональный сбор). Зеркало secondary_only в + fetch_all_secondary, которое фильтрует обратное (!= "novostroyki"). DETAIL-фаза + (обогащение cian-листингов без detail_enriched_at) и HOUSES-фаза (newbuilding-enrich) + не затронуты фильтром. + Фазы per anchor: 1. SERP: CianScraper.fetch_around_multi_room(lat, lon, radius_m, pages=pages_per_anchor) - 2. SAVE: save_listings(db, lots, run_id=run_id) - 3. DETAIL: top-N свежих cian-листингов без detail_enriched_at → + 2. FILTER (newbuilding_only): отбросить вторичку (listing_segment != "novostroyki") + 3. SAVE: save_listings(db, lots, run_id=run_id) + 4. DETAIL: top-N свежих cian-листингов без detail_enriched_at → fetch_detail(source_url) + save_detail_enrichment(db, listing_id, enrichment) (save_detail_enrichment уже пишет price_changes → offer_price_history) - 4. HOUSES (если enrich_houses=True): для домов из этого anchor'а с cian_zhk_url → + 5. HOUSES (если enrich_houses=True): для домов из этого anchor'а с cian_zhk_url → fetch_newbuilding(zhk_url) + save_newbuilding_enrichment(db, house_id, enr) Anti-bot: @@ -1647,16 +1657,28 @@ async def run_cian_city_sweep( pages=pages_per_anchor, ) counters.lots_fetched += len(anchor_lots) + # ── Фильтр вторички (newbuilding_only) ───────────────────── + # Зеркало secondary_only в fetch_all_secondary: тут оставляем + # только новостройки — вторичкой авторитетно владеет run_cian_full_load. + if newbuilding_only: + _before = len(anchor_lots) + anchor_lots = [ + lot for lot in anchor_lots if lot.listing_segment == "novostroyki" + ] + counters.lots_dropped_secondary += _before - len(anchor_lots) if anchor_lots: inserted, updated = save_listings(db, anchor_lots, run_id=run_id) counters.lots_inserted += inserted counters.lots_updated += updated consecutive_failures = 0 logger.info( - "cian-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d", + "cian-sweep run_id=%d anchor %s: SERP fetched=%d nb_kept=%d " + "dropped_secondary=%d ins=%d upd=%d", run_id, _a_name, + counters.lots_fetched, len(anchor_lots), + counters.lots_dropped_secondary, counters.lots_inserted, counters.lots_updated, ) diff --git a/tradein-mvp/backend/data/sql/126_scrape_schedules_seed_cian_full_load.sql b/tradein-mvp/backend/data/sql/126_scrape_schedules_seed_cian_full_load.sql new file mode 100644 index 00000000..bc8240ac --- /dev/null +++ b/tradein-mvp/backend/data/sql/126_scrape_schedules_seed_cian_full_load.sql @@ -0,0 +1,53 @@ +-- 126_scrape_schedules_seed_cian_full_load.sql +-- Seed row для cian_full_load — exhaustive региональный сбор Cian ЕКБ ВТОРИЧКИ +-- (run_cian_full_load, scrape_pipeline.py). Авторитетный источник вторички: +-- room×price адаптивное партиционирование (БЕЗ anchor'ов), secondary_only, +-- incremental on_bucket save. Prod-verified (run #46: 29511 уникальных за ~25 мин). +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)). +-- Idempotent: ON CONFLICT (source) DO UPDATE обновляет window/default_params при +-- переустановке миграции, НО НЕ затирает enabled — оператор может выключить schedule +-- вручную, и повторный прогон миграции это сохранит (COALESCE на existing enabled). +-- +-- Окно 20:00-22:00 UTC (23:00-01:00 МСК) — намеренно отделено от cian_city_sweep +-- (2-5) и cian_history_backfill (2-5), чтобы не конкурировать на shared kf-прокси. +-- +-- default_params соответствуют сигнатуре run_cian_full_load(): +-- price_cap_per_bucket — максимум офферов на price-бакет (< Cian SERP-cap ~1500). +-- concurrency — параллельных page-фетчей в leaf-бакете. +-- request_delay_sec — задержка между SERP-запросами. +-- enrich_detail — detail-обогащение (по умолчанию off — тяжело). +-- detail_top_n — сколько листингов обогащать detail (актуально при enrich_detail). +-- (newbuilding_only сюда НЕ кладём — это параметр cian_city_sweep, не full_load.) + +BEGIN; + +-- next_run_at = следующее наступление night-window (tomorrow + window_start_hour), +-- иначе при NULL get_due_schedules() выстрелит сразу после деплоя (вне окна). +-- window_*_hour документированы в UTC (см. 052) → 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_full_load', + true, + 20, + 22, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 20)) AT TIME ZONE 'UTC', + '{"price_cap_per_bucket": 1400, "concurrency": 5, "request_delay_sec": 4.0, "enrich_detail": false, "detail_top_n": 0}'::jsonb +) +ON CONFLICT (source) DO UPDATE SET + window_start_hour = EXCLUDED.window_start_hour, + window_end_hour = EXCLUDED.window_end_hour, + default_params = EXCLUDED.default_params, + updated_at = NOW(); +-- enabled НЕ в SET — сохраняем операторское значение (ON CONFLICT не трогает enabled). + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_cian_city_sweep.py b/tradein-mvp/backend/tests/test_cian_city_sweep.py index f5324ad1..a82dcacd 100644 --- a/tradein-mvp/backend/tests/test_cian_city_sweep.py +++ b/tradein-mvp/backend/tests/test_cian_city_sweep.py @@ -36,8 +36,14 @@ TEST_ANCHORS: list[tuple[float, float, str]] = [ ] -def _fake_lot(offer_id: str, *, nb: bool = False) -> ScrapedLot: - """Создаём ScrapedLot с опциональным newbuilding-ссылкой.""" +def _fake_lot( + offer_id: str, *, nb: bool = False, segment: str | None = "novostroyki" +) -> ScrapedLot: + """Создаём ScrapedLot с опциональным newbuilding-ссылкой. + + segment по умолчанию "novostroyki" — run_cian_city_sweep теперь NB-only + (newbuilding_only=True), поэтому дефолтные lots должны проходить SERP-фильтр. + """ return ScrapedLot( source="cian", source_url=f"https://ekb.cian.ru/sale/flat/{offer_id}/", @@ -46,6 +52,7 @@ def _fake_lot(offer_id: str, *, nb: bool = False) -> ScrapedLot: price_rub=5_000_000, area_m2=50.0, rooms=2, + listing_segment=segment, house_source="cian_newbuilding" if nb else "cian", house_ext_id="99999" if nb else None, ) @@ -195,6 +202,96 @@ async def test_sweep_iterates_all_anchors_serp_and_save( assert fake.banned is None +async def test_newbuilding_only_drops_secondary_before_save( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """newbuilding_only=True (default): save_listings получает только novostroyki-lots. + + SERP отдаёт mixed-сегмент выдачу — вторичкой авторитетно владеет run_cian_full_load, + поэтому sweep её отбрасывает до save и считает в lots_dropped_secondary. + """ + saved_lots: list[list[ScrapedLot]] = [] + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + # 1 novostroyki + 2 vtorichka на anchor + return [ + _fake_lot(f"{lat}-nb", segment="novostroyki"), + _fake_lot(f"{lat}-v1", segment="vtorichka"), + _fake_lot(f"{lat}-v2", segment="vtorichka"), + ] + + def fake_save(_db: Any, lots: list, *, run_id: int | None = None): + saved_lots.append(list(lots)) + return len(lots), 0 + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=1, + anchors=TEST_ANCHORS, + pages_per_anchor=2, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + newbuilding_only=True, + ) + + # Каждый save-вызов содержит только novostroyki-lot + assert len(saved_lots) == len(TEST_ANCHORS) + for lots in saved_lots: + assert len(lots) == 1 + assert all(lot.listing_segment == "novostroyki" for lot in lots) + # Counters: fetched все 3×N, сохранён 1×N, отброшено 2×N + assert counters.lots_fetched == len(TEST_ANCHORS) * 3 + assert counters.lots_inserted == len(TEST_ANCHORS) * 1 + assert counters.lots_dropped_secondary == len(TEST_ANCHORS) * 2 + + +async def test_newbuilding_only_false_saves_all_segments( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """newbuilding_only=False: фильтр отключён — save получает и вторичку, и новостройки.""" + saved_lots: list[list[ScrapedLot]] = [] + + async def fake_fetch_multi(self: CianScraper, lat: float, lon: float, r: int, pages: int = 8): + return [ + _fake_lot(f"{lat}-nb", segment="novostroyki"), + _fake_lot(f"{lat}-v1", segment="vtorichka"), + ] + + def fake_save(_db: Any, lots: list, *, run_id: int | None = None): + saved_lots.append(list(lots)) + return len(lots), 0 + + monkeypatch.setattr(CianScraper, "fetch_around_multi_room", fake_fetch_multi) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_cian_city_sweep( + db=MagicMock(), + run_id=1, + anchors=TEST_ANCHORS, + pages_per_anchor=2, + detail_top_n=0, + enrich_houses=False, + request_delay_sec=0.0, + newbuilding_only=False, + ) + + assert len(saved_lots) == len(TEST_ANCHORS) + for lots in saved_lots: + assert len(lots) == 2 + assert counters.lots_inserted == len(TEST_ANCHORS) * 2 + assert counters.lots_dropped_secondary == 0 + + async def test_detail_phase_runs_and_fills_counters( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py index 7cf27ccb..fa4b59cf 100644 --- a/tradein-mvp/backend/tests/test_scheduler.py +++ b/tradein-mvp/backend/tests/test_scheduler.py @@ -364,3 +364,98 @@ async def test_scheduler_dispatch_routes_cian_city_sweep(monkeypatch: pytest.Mon await _sched.trigger_cian_city_sweep_run(db, sch) assert "cian_city_sweep" in triggered + + +# ── cian_full_load: trigger_cian_full_load_run + dispatch ──────────────────── + + +_CIAN_FULL_LOAD_ROW = { + "source": "cian_full_load", + "window_start_hour": 20, + "window_end_hour": 22, + "default_params": { + "price_cap_per_bucket": 1400, + "concurrency": 5, + "request_delay_sec": 4.0, + "enrich_detail": False, + "detail_top_n": 0, + }, +} + + +@pytest.mark.asyncio +async def test_trigger_cian_full_load_creates_run(monkeypatch: pytest.MonkeyPatch) -> None: + """trigger_cian_full_load_run claims run + launches run_cian_full_load via create_task.""" + db = _FakeSchedulerDB() + calls: dict[str, list] = {"create_run": [], "load": []} + + def fake_create_run(_d, *, source, params): + calls["create_run"].append(source) + db.running[source] = True + return 301 + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + async def fake_full_load(run_db, *, run_id, **kwargs): + calls["load"].append((run_id, kwargs)) + + monkeypatch.setattr(_sched, "run_cian_full_load", fake_full_load) + monkeypatch.setattr(_sched, "SessionLocal", lambda: db) + + run_id = await _sched.trigger_cian_full_load_run(db, _CIAN_FULL_LOAD_ROW) + + assert run_id == 301 + assert calls["create_run"] == ["cian_full_load"] + # Drain event-loop so the asyncio.create_task runs + await asyncio.sleep(0) + assert len(calls["load"]) == 1 + loaded_run_id, kwargs = calls["load"][0] + assert loaded_run_id == 301 + # Params propagated from default_params; resume_run_id forced to None + assert kwargs["price_cap_per_bucket"] == 1400 + assert kwargs["concurrency"] == 5 + assert kwargs["request_delay_sec"] == 4.0 + assert kwargs["enrich_detail"] is False + assert kwargs["detail_top_n"] == 0 + assert kwargs["resume_run_id"] is None + + +@pytest.mark.asyncio +async def test_trigger_cian_full_load_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None: + """trigger_cian_full_load_run returns None when source already running.""" + db = _FakeSchedulerDB() + db.running["cian_full_load"] = 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_full_load_run(db, _CIAN_FULL_LOAD_ROW) + + assert result is None + assert calls["n"] == 0 + + +@pytest.mark.asyncio +async def test_scheduler_dispatch_routes_cian_full_load(monkeypatch: pytest.MonkeyPatch) -> None: + """Dispatch block в scheduler_loop роутит cian_full_load → trigger_cian_full_load_run.""" + triggered: list[str] = [] + + async def fake_trigger(db, sch): + triggered.append(sch["source"]) + return 1 + + monkeypatch.setattr(_sched, "trigger_cian_full_load_run", fake_trigger) + + db = object() + sch = _CIAN_FULL_LOAD_ROW + source = sch["source"] + # Replicate the elif chain from scheduler_loop for cian_full_load + if source == "cian_full_load": + await _sched.trigger_cian_full_load_run(db, sch) + + assert "cian_full_load" in triggered