feat(tradein): автоматизация cian-backfill + rosreestr-import через scheduler (#560 #563)

Оба источника данных до этого требовали ручного запуска (HTTP-запрос в admin UI
и bash-скрипт на хосте соответственно). Теперь оба интегрированы в in-app scheduler
(тот же механизм что avito_city_sweep) — запускаются автоматически ночью.

#560 (cian_history_backfill):
- Window 02:00–05:00 UTC; pre-check verify_session() перед claim'ом слота —
  если куки истекли, run не создаётся, отправляется GlitchTip capture_message.
- Вызывает существующий backfill_cian_history() как есть; checkpoint через
  update_heartbeat() до и после вызова (для zombie-detection).

#563 (rosreestr_dkp_import):
- Window 04:00–06:00 UTC; sync функция в run_in_executor (DB-only, нет HTTP).
- Cursor-based пагинация (WHERE id > last_id ORDER BY id LIMIT batch_size),
  heartbeat = checkpoint каждый батч (~2000 строк по умолчанию).
- SAVEPOINT per row через begin_nested() — один сбойный row не откатывает батч.
- Идемпотентен: ON CONFLICT (dedup_hash) DO NOTHING; dedup_hash = md5('ros:dkp:'+id)
  соответствует Fix #549.
- Cleanup при старте: удаляет legacy rows address='Екатеринбург, реальная сделка'.
- Координаты NULL — геокодинг остаётся ручным follow-up.

Migration 072: seed rows scrape_schedules + foreign table gendesign_rosreestr_deals
(через gendesign_remote FDW server, настроен в 060).

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

@@ -6,6 +6,11 @@
 2. Для каждого: проверить нет ли уже running run этого source — если нет, trigger.
 3. После trigger — recompute next_run_at в следующих суток.
 4. На old runs (orphaned status='running' > 6h без heartbeat) — пометить как 'zombie'.
+
+Sources:
+  - avito_city_sweep        → run_avito_city_sweep (scrape_pipeline.py)
+  - cian_history_backfill   → run_cian_backfill (this module, #560)
+  - rosreestr_dkp_import    → import_rosreestr_dkp (this module, #563)
 """
 from __future__ import annotations
 
@@ -15,6 +20,7 @@ import random
 from datetime import UTC, datetime, time, timedelta
 from typing import Any
 
+import sentry_sdk
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
@@ -105,19 +111,20 @@ def reap_zombies(db: Session) -> int:
     return len(rows)
 
 
-async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
-    """Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task.
+def _claim_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at.
 
-    Returns run_id (или None если skip — есть running run).
+    Returns run_id или None если уже есть running run для этого source.
+    Общий helper для всех source-handler'ов.
     """
-    if has_running_run(db, schedule_row["source"]):
-        logger.info("scheduler: skip — already running for source=%s", schedule_row["source"])
+    source = schedule_row["source"]
+    if has_running_run(db, source):
+        logger.info("scheduler: skip — already running for source=%s", source)
         return None
 
     params = schedule_row.get("default_params") or {}
-    run_id = runs_mod.create_run(db, source=schedule_row["source"], params=params)
+    run_id = runs_mod.create_run(db, source=source, params=params)
 
-    # Update schedule: last_run_id, last_run_at, recompute next_run_at
     next_at = compute_next_run_at(
         schedule_row["window_start_hour"],
         schedule_row["window_end_hour"],
@@ -131,9 +138,28 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]
             WHERE source = :source
             """
         ),
-        {"run_id": run_id, "next_at": next_at, "source": schedule_row["source"]},
+        {"run_id": run_id, "next_at": next_at, "source": source},
     )
     db.commit()
+    logger.info(
+        "scheduler: claimed run_id=%d source=%s next_run_at=%s",
+        run_id,
+        source,
+        next_at.isoformat(),
+    )
+    return run_id
+
+
+async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать новый scrape_runs + launch run_avito_city_sweep в asyncio.create_task.
+
+    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 {}
 
     # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick)
     async def _run() -> None:
@@ -156,12 +182,378 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any]
     task = asyncio.create_task(_run())
     # Keep reference to avoid GC before task completes (RUF006)
     task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None)
-    logger.info(
-        "scheduler: triggered city_sweep run_id=%d next_run_at=%s", run_id, next_at.isoformat()
-    )
+    logger.info("scheduler: triggered city_sweep run_id=%d", run_id)
     return run_id
 
 
+async def trigger_cian_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch run_cian_backfill в asyncio.create_task.
+
+    Pre-checks Cian session validity before claiming the run.
+    Marks nothing as failed if cookies are absent/expired — just logs + GlitchTip warning.
+    Returns run_id или None (skip / expired cookies).
+    """
+    from app.services.cian_session import load_session, verify_session
+
+    # Pre-check: load and verify cookies before claiming the run slot
+    cookies = load_session(db)
+    if cookies is None:
+        logger.warning("scheduler: cian_history_backfill skipped — no valid session cookies in DB")
+        return None
+
+    state = await verify_session(cookies)
+    if state is None:
+        logger.warning(
+            "scheduler: cian_history_backfill — cookies expired or invalid, skipping run"
+        )
+        try:
+            sentry_sdk.capture_message(
+                "cian_history_backfill skipped: Cian session cookies expired — "
+                "please re-upload via admin UI",
+                level="warning",
+            )
+        except Exception:
+            pass  # sentry_sdk not initialised in dev
+        return None
+
+    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 _execute_cian_backfill(run_db, run_id=run_id, params=params)
+        except Exception:
+            logger.exception("scheduler: _execute_cian_backfill 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_history_backfill run_id=%d", run_id)
+    return run_id
+
+
+async def _execute_cian_backfill(
+    db: Session,
+    *,
+    run_id: int,
+    params: dict[str, Any],
+) -> None:
+    """Orchestrate Cian history backfill with heartbeat + checkpoint.
+
+    Wraps backfill_cian_history(), updating scrape_runs counters (via update_heartbeat)
+    before and after the batch call for zombie-detection visibility.
+
+    Checkpoint/resume semantics: backfill_cian_history() queries rows WHERE history IS
+    NULL via LEFT JOIN — so re-running after a partial completion naturally skips
+    already-processed rows (idempotent by design).
+
+    Params (from default_params jsonb):
+        batch_size: int — rows per run (listings + houses counted separately).
+    """
+    from app.tasks.cian_history_backfill import backfill_cian_history
+
+    batch_size = int(params.get("batch_size", 100))
+    counters: dict[str, int] = {
+        "listings_processed": 0,
+        "listings_succeeded": 0,
+        "listings_failed": 0,
+        "houses_processed": 0,
+        "houses_succeeded": 0,
+        "houses_failed": 0,
+    }
+
+    try:
+        runs_mod.update_heartbeat(db, run_id, counters)
+
+        result = await backfill_cian_history(
+            db,
+            batch_size=batch_size,
+            do_listings=True,
+            do_houses=True,
+            do_valuations=False,
+        )
+
+        counters = {
+            "listings_processed": result.listings_processed,
+            "listings_succeeded": result.listings_succeeded,
+            "listings_failed": result.listings_failed_fetch + result.listings_failed_save,
+            "houses_processed": result.houses_processed,
+            "houses_succeeded": result.houses_succeeded,
+            "houses_failed": result.houses_failed_fetch + result.houses_failed_save,
+            "duration_sec": int(result.duration_sec),
+        }
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "scheduler: cian_history_backfill run_id=%d done — "
+            "listings=%d/%d houses=%d/%d %.1fs",
+            run_id,
+            result.listings_succeeded,
+            result.listings_total,
+            result.houses_succeeded,
+            result.houses_total,
+            result.duration_sec,
+        )
+    except Exception as exc:
+        logger.exception("scheduler: cian_history_backfill run_id=%d failed", run_id)
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise
+
+
+async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
+    """Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task)."""
+    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:
+            # import_rosreestr_dkp is synchronous (DB-only, no async HTTP)
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, import_rosreestr_dkp, run_db, run_id, params)
+        except Exception:
+            logger.exception("scheduler: import_rosreestr_dkp 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 rosreestr_dkp_import run_id=%d", run_id)
+    return run_id
+
+
+def import_rosreestr_dkp(
+    db: Session,
+    run_id: int,
+    params: dict[str, Any],
+) -> None:
+    """Import ДКП-сделок из gendesign rosreestr_deals через postgres_fdw.
+
+    Python-порт import-rosreestr.sh (Variant C из #563).
+
+    Источник: foreign table gendesign_rosreestr_deals (создана в migration 072).
+    SERVER gendesign_remote настроен в 060_postgres_fdw_extension.sql.
+    USER MAPPING создаётся при startup в core/fdw.py (tradein_fdw_reader).
+
+    Фильтры (совпадают с import-rosreestr.sh + Fix_Rosreestr_Dkp_Filter_May24):
+      - region_code = 66 (Свердловская область)
+      - city ILIKE '%катеринбург%'
+      - realestate_type_code = '002001003000' (квартира)
+      - area BETWEEN 18 AND 200
+      - deal_price BETWEEN 1000000 AND 100000000
+      - street IS NOT NULL AND trim(street) != ''
+      - doc_type = 'ДКП'  (только вторичка — #549 / Fix_Rosreestr_Dkp_Filter_May24)
+      - period_start_date >= since  (default '2024-01-01')
+
+    dedup_hash: md5('ros:dkp:' || id) — соответствует #549 / Fix_Rosreestr_Dkp_Filter_May24.
+    Rooms: выводятся из площади (Росреестр не отдаёт кол-во комнат).
+
+    Batch-процессинг: читаем из FDW батчами по batch_size через cursor-based пагинацию
+    (WHERE id > last_id ORDER BY id). Heartbeat обновляется каждый батч (= checkpoint).
+    SAVEPOINT per row — один сбойный row не откатывает батч.
+
+    Координаты: NULL после импорта — геокодинг остаётся follow-up (geocode-deals).
+    TODO (follow-up): запустить geocode backfill после import.
+
+    Cleanup: удаляет legacy строки address='Екатеринбург, реальная сделка' (pre-#549).
+    """
+    since: str = str(params.get("since", "2024-01-01"))
+    batch_size: int = int(params.get("batch_size", 2000))
+
+    counters: dict[str, int] = {
+        "rows_fetched": 0,
+        "rows_inserted": 0,
+        "rows_skipped": 0,
+        "batches_done": 0,
+    }
+
+    # Cleanup legacy synthetic rows (pre-#549, idempotent)
+    try:
+        deleted = db.execute(
+            text(
+                "DELETE FROM deals WHERE address = CAST(:addr AS text) RETURNING id"
+            ),
+            {"addr": "Екатеринбург, реальная сделка"},
+        ).fetchall()
+        db.commit()
+        if deleted:
+            logger.info(
+                "rosreestr_dkp_import run_id=%d: removed %d legacy synthetic rows",
+                run_id,
+                len(deleted),
+            )
+    except Exception as exc:
+        logger.warning(
+            "rosreestr_dkp_import run_id=%d: cleanup failed (non-fatal): %s", run_id, exc
+        )
+        db.rollback()
+
+    last_id: int = 0
+    total_batches = 0
+
+    try:
+        while True:
+            if runs_mod.is_cancelled(db, run_id):
+                logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id)
+                runs_mod.mark_cancelled(db, run_id)
+                return
+
+            # Cursor-based pagination via foreign table gendesign_rosreestr_deals.
+            # FDW pushes WHERE + ORDER BY + LIMIT to gendesign-postgres automatically
+            # (postgres_fdw 'use_remote_estimate' is off by default but clause push-down
+            # still happens for simple predicates on the remote side).
+            batch_rows = db.execute(
+                text("""
+                    SELECT
+                        id,
+                        md5('ros:dkp:' || CAST(id AS text))     AS dedup_hash,
+                        'Екатеринбург, ' || trim(street)         AS address,
+                        CASE
+                            WHEN area < 30 THEN 0
+                            WHEN area < 44 THEN 1
+                            WHEN area < 62 THEN 2
+                            WHEN area < 85 THEN 3
+                            ELSE 4
+                        END                                      AS rooms,
+                        round(area, 2)                           AS area_m2,
+                        LEAST(
+                            NULLIF(
+                                substring(floor FROM '[0-9]+'), ''
+                            )::int,
+                            100
+                        )                                        AS floor_num,
+                        year_build                               AS year_built,
+                        round(deal_price)::bigint                AS price_rub,
+                        round(price_per_sqm)::int                AS price_per_m2,
+                        period_start_date                        AS deal_date
+                    FROM gendesign_rosreestr_deals
+                    WHERE region_code = 66
+                      AND city ILIKE '%катеринбург%'
+                      AND realestate_type_code = '002001003000'
+                      AND area BETWEEN 18 AND 200
+                      AND deal_price BETWEEN 1000000 AND 100000000
+                      AND street IS NOT NULL AND trim(street) <> ''
+                      AND doc_type = 'ДКП'
+                      AND period_start_date >= CAST(:since AS date)
+                      AND id > CAST(:last_id AS bigint)
+                    ORDER BY id
+                    LIMIT CAST(:batch_size AS int)
+                """),
+                {"since": since, "last_id": last_id, "batch_size": batch_size},
+            ).mappings().all()
+
+            if not batch_rows:
+                break
+
+            total_batches += 1
+            batch_inserted = 0
+            batch_skipped = 0
+            batch_max_id = last_id
+
+            for row in batch_rows:
+                row_id: int = int(row["id"])
+                if row_id > batch_max_id:
+                    batch_max_id = row_id
+
+                try:
+                    with db.begin_nested():  # SAVEPOINT per row
+                        inserted = db.execute(
+                            text("""
+                                INSERT INTO deals (
+                                    source, dedup_hash, address, rooms, area_m2,
+                                    floor, year_built, price_rub, price_per_m2, deal_date
+                                )
+                                VALUES (
+                                    'rosreestr',
+                                    CAST(:dedup_hash AS text),
+                                    CAST(:address AS text),
+                                    CAST(:rooms AS int),
+                                    CAST(:area_m2 AS numeric),
+                                    CAST(:floor_num AS int),
+                                    CAST(:year_built AS int),
+                                    CAST(:price_rub AS bigint),
+                                    CAST(:price_per_m2 AS int),
+                                    CAST(:deal_date AS date)
+                                )
+                                ON CONFLICT (dedup_hash) DO NOTHING
+                                RETURNING id
+                            """),
+                            {
+                                "dedup_hash": row["dedup_hash"],
+                                "address": row["address"],
+                                "rooms": row["rooms"],
+                                "area_m2": row["area_m2"],
+                                "floor_num": row["floor_num"],
+                                "year_built": row["year_built"],
+                                "price_rub": row["price_rub"],
+                                "price_per_m2": row["price_per_m2"],
+                                "deal_date": row["deal_date"],
+                            },
+                        ).fetchone()
+                        if inserted is not None:
+                            batch_inserted += 1
+                        else:
+                            batch_skipped += 1
+                except Exception as exc:
+                    logger.warning(
+                        "rosreestr_dkp_import run_id=%d: row id=%d INSERT failed: %s",
+                        run_id,
+                        row_id,
+                        exc,
+                    )
+                    batch_skipped += 1
+
+            last_id = batch_max_id
+            db.commit()
+
+            counters["rows_fetched"] += len(batch_rows)
+            counters["rows_inserted"] += batch_inserted
+            counters["rows_skipped"] += batch_skipped
+            counters["batches_done"] = total_batches
+            counters["last_id"] = last_id  # type: ignore[assignment]
+
+            # Heartbeat = checkpoint: allows zombie detection + resume visibility
+            runs_mod.update_heartbeat(db, run_id, counters)
+            logger.info(
+                "rosreestr_dkp_import run_id=%d: batch=%d fetched=%d "
+                "inserted=%d skipped=%d last_id=%d",
+                run_id,
+                total_batches,
+                len(batch_rows),
+                batch_inserted,
+                batch_skipped,
+                last_id,
+            )
+
+            if len(batch_rows) < batch_size:
+                break  # Last partial batch — no more rows
+
+        runs_mod.mark_done(db, run_id, counters)
+        logger.info(
+            "rosreestr_dkp_import run_id=%d done: "
+            "total_fetched=%d inserted=%d skipped=%d batches=%d",
+            run_id,
+            counters["rows_fetched"],
+            counters["rows_inserted"],
+            counters["rows_skipped"],
+            total_batches,
+        )
+
+    except Exception as exc:
+        logger.exception(
+            "rosreestr_dkp_import run_id=%d failed at last_id=%d", run_id, last_id
+        )
+        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters)
+        raise
+
+
 def get_due_schedules(db: Session) -> list[dict[str, Any]]:
     """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
     rows = db.execute(
@@ -192,10 +584,15 @@ async def scheduler_loop() -> None:
                 # Process due schedules
                 due = get_due_schedules(db)
                 for sch in due:
-                    if sch["source"] == "avito_city_sweep":
+                    source = sch["source"]
+                    if source == "avito_city_sweep":
                         await trigger_avito_city_sweep_run(db, sch)
+                    elif source == "cian_history_backfill":
+                        await trigger_cian_backfill_run(db, sch)
+                    elif source == "rosreestr_dkp_import":
+                        await trigger_rosreestr_dkp_run(db, sch)
                     else:
-                        logger.warning("scheduler: unknown source=%s, skip", sch["source"])
+                        logger.warning("scheduler: unknown source=%s, skip", source)
             finally:
                 db.close()
         except Exception:

tradein-mvp/backend/data/sql/072_scrape_schedules_seed_cian_rosreestr.sql

@@ -0,0 +1,84 @@
+-- 072_scrape_schedules_seed_cian_rosreestr.sql
+-- Seed rows для двух новых автоматизированных задач (#560 + #563).
+-- Добавляет foreign table gendesign_rosreestr_deals для импорта через FDW.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql, 060_postgres_fdw_extension.sql.
+-- Idempotent: безопасно запускать повторно.
+--
+-- cian_history_backfill — 02:00-05:00 UTC (05:00-08:00 МСК)
+--   Аналог ручного POST /api/v1/admin/scrape/cian-backfill-history.
+--   default_params соответствуют API-параметрам backfill_cian_history():
+--     batch_size  — кол-во listing/house-строк за один run.
+--   Если Cian-сессия истекла — run помечается failed + отправляется capture_message.
+--
+-- rosreestr_dkp_import — 04:00-06:00 UTC (07:00-09:00 МСК)
+--   Аналог ручного ./deploy/import-rosreestr.sh через FDW + pure-Python psycopg v3.
+--   SINCE по умолчанию '2024-01-01' — резонное начало истории ДКП.
+--   Idempotent: ON CONFLICT (dedup_hash) DO NOTHING.
+--
+-- gendesign_rosreestr_deals — foreign table через gendesign_remote FDW server.
+--   Колонки: только те, что используются импортёром (минимальный набор).
+--   Партиционированная таблица на gendesign-стороне; FDW пушит WHERE вниз автоматически.
+
+BEGIN;
+
+-- ── Foreign table для rosreestr_deals ─────────────────────────────────────────
+-- server gendesign_remote создан в 060_postgres_fdw_extension.sql.
+-- USER MAPPING управляется backend startup (core/fdw.py, role tradein_fdw_reader).
+-- tradein_fdw_reader должен иметь SELECT на rosreestr_deals в gendesign-БД.
+-- Колонки подобраны под SELECT в import_rosreestr_dkp() в scheduler.py.
+
+DROP FOREIGN TABLE IF EXISTS gendesign_rosreestr_deals;
+
+CREATE FOREIGN TABLE gendesign_rosreestr_deals (
+    id                    bigint,
+    region_code           int,
+    city                  text,
+    street                text,
+    realestate_type_code  text,
+    doc_type              text,
+    area                  numeric,
+    floor                 text,
+    year_build            int,
+    deal_price            numeric,
+    price_per_sqm         numeric,
+    period_start_date     date
+)
+SERVER gendesign_remote
+OPTIONS (schema_name 'public', table_name 'rosreestr_deals');
+
+COMMENT ON FOREIGN TABLE gendesign_rosreestr_deals IS
+    'Live read из gendesign.rosreestr_deals для rosreestr_dkp_import (FDW, #563). '
+    'USER MAPPING управляется backend startup (core/fdw.py).';
+
+-- ── Seed rows для scrape_schedules ────────────────────────────────────────────
+
+INSERT INTO scrape_schedules (
+    source,
+    enabled,
+    window_start_hour,
+    window_end_hour,
+    default_params
+)
+VALUES
+(
+    'cian_history_backfill',
+    true,
+    2,
+    5,
+    '{"batch_size": 100}'::jsonb
+),
+(
+    'rosreestr_dkp_import',
+    true,
+    4,
+    6,
+    '{"since": "2024-01-01", "batch_size": 2000}'::jsonb
+)
+ON CONFLICT (source) DO NOTHING;
+
+COMMENT ON TABLE scrape_schedules IS
+    'In-app scheduler config (заменяет cron-script setup). '
+    'Sources: avito_city_sweep, cian_history_backfill, rosreestr_dkp_import.';
+
+COMMIT;