feat(tradein): schedule cian_full_load (exhaustive secondary) + anchor sweep NB-only

Подключает 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

@@ -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":

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,
                 )

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;

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:

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