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

"""In-app scheduler — periodically triggers configured scrapes based on scrape_schedules table.

Запускается в FastAPI lifespan (app.main lifespan context manager) при startup.
Логика:
1. Каждые 60 sec query scrape_schedules WHERE enabled=true AND next_run_at <= NOW().
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)
  - yandex_city_sweep       → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT)
  - cian_history_backfill   → run_cian_backfill (this module, #560)
  - rosreestr_dkp_import    → import_rosreestr_dkp (this module, #563)
  - listing_source_snapshot → snapshot_listing_sources (tasks/listing_source_snapshot.py, #570;
                              pure internal DB snapshot — no external calls, enabled by default)
  - asking_to_sold_ratio_refresh → recompute_asking_to_sold_ratios
                              (tasks/asking_to_sold_ratio.py, #648 Stage 4; pure internal DB
                              re-derivation of the asking→sold ratios — no external calls,
                              enabled by default; window after rosreestr_dkp_import)
"""
from __future__ import annotations

import asyncio
import logging
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

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_yandex_city_sweep

logger = logging.getLogger(__name__)

# Loop interval — check каждую минуту
SCHEDULER_TICK_SEC = 60
ZOMBIE_THRESHOLD_HOURS = 6


def compute_next_run_at(
    window_start_hour: int, window_end_hour: int, *, now: datetime | None = None
) -> datetime:
    """Pick random datetime в window [start, end) UTC, для СЛЕДУЮЩИХ суток после now.

    Если window_end_hour <= window_start_hour → cross-midnight window
    (например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59).
    """
    now = now or datetime.now(tz=UTC)
    tomorrow = (now + timedelta(days=1)).date()

    if window_end_hour > window_start_hour:
        # Обычное окно (например 2..5 → 02:00-04:59)
        start_seconds = window_start_hour * 3600
        end_seconds = window_end_hour * 3600
        rand_seconds = random.randint(start_seconds, end_seconds - 1)
        return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(
            seconds=rand_seconds
        )
    else:
        # Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59)
        # Длина окна = (24-start) + end часов
        total_seconds = ((24 - window_start_hour) + window_end_hour) * 3600
        rand_seconds = random.randint(0, total_seconds - 1)
        # Если rand попадает в первую часть (start..24)
        first_half = (24 - window_start_hour) * 3600
        if rand_seconds < first_half:
            # Текущая дата (если ещё не наступило окно) или next day
            base_date = now.date() if now.hour < window_start_hour else tomorrow
            return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta(
                seconds=window_start_hour * 3600 + rand_seconds
            )
        else:
            # Во второй части (0..end), следующего дня
            offset = rand_seconds - first_half
            return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset)


def has_running_run(db: Session, source: str) -> bool:
    """Есть ли активный run для source (status='running')."""
    row = db.execute(
        text(
            """
            SELECT 1 FROM scrape_runs
            WHERE source = :source AND status = 'running'
            LIMIT 1
            """
        ),
        {"source": source},
    ).fetchone()
    return row is not None


def reap_zombies(db: Session) -> int:
    """Mark scrape_runs as 'zombie' если heartbeat не обновлялся > ZOMBIE_THRESHOLD_HOURS hours."""
    zombie_interval = f"{ZOMBIE_THRESHOLD_HOURS} hours"
    result = db.execute(
        text(
            """
            UPDATE scrape_runs
            SET status = 'zombie', finished_at = NOW()
            WHERE status = 'running'
              AND (heartbeat_at IS NULL
                   OR heartbeat_at < NOW() - CAST(:interval AS interval))
            RETURNING id
            """
        ),
        {"interval": zombie_interval},
    )
    rows = result.fetchall()
    db.commit()
    if rows:
        logger.warning("scheduler: reaped %d zombie runs: %s", len(rows), [r.id for r in rows])
    return len(rows)


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 если уже есть running run для этого source.
    Общий helper для всех source-handler'ов.
    """
    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=source, params=params)

    next_at = compute_next_run_at(
        schedule_row["window_start_hour"],
        schedule_row["window_end_hour"],
    )
    db.execute(
        text(
            """
            UPDATE scrape_schedules
            SET last_run_id = :run_id, last_run_at = NOW(),
                next_run_at = :next_at, updated_at = NOW()
            WHERE source = :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:
        run_db = SessionLocal()
        try:
            await run_avito_city_sweep(
                run_db,
                run_id=run_id,
                pages_per_anchor=int(params.get("pages_per_anchor", 3)),
                detail_top_n=int(params.get("detail_top_n", 20)),
                request_delay_sec=float(params.get("request_delay_sec", 7.0)),
                enrich_houses=bool(params.get("enrich_houses", True)),
                radius_m=int(params.get("radius_m", 1500)),
            )
        except Exception:
            logger.exception("scheduler: run_avito_city_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    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", run_id)
    return run_id


async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
    """Создать новый scrape_runs + launch run_yandex_city_sweep в asyncio.create_task.

    Зеркало trigger_avito_city_sweep_run, но проще (#561): у Yandex sweep нет
    houses/detail-параметров. Shipped DORMANT — schedule seed enabled=false (078),
    включается оператором вручную после #623 (egress-IP rotation).

    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:
        run_db = SessionLocal()
        try:
            await run_yandex_city_sweep(
                run_db,
                run_id=run_id,
                pages_per_anchor=int(params.get("pages_per_anchor", 2)),
                request_delay_sec=float(params.get("request_delay_sec", 9.0)),
                radius_m=int(params.get("radius_m", 1500)),
            )
        except Exception:
            logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id)
        finally:
            run_db.close()

    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 yandex_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


async def trigger_listing_source_snapshot_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch snapshot_listing_sources в executor (sync DB-only task).

    Per-source price-history daily snapshot (#570). Mirror trigger_rosreestr_dkp_run:
    задача синхронная (чистый internal DB-snapshot, БЕЗ внешних HTTP-вызовов / анти-бота),
    поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (079),
    в отличие от avito/yandex sweep (dormant).

    Returns run_id (или None если skip — есть running run).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            # snapshot_listing_sources is synchronous (DB-only, no async HTTP)
            from app.tasks.listing_source_snapshot import snapshot_listing_sources

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, snapshot_listing_sources, run_db, run_id)
        except Exception:
            logger.exception("scheduler: snapshot_listing_sources 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 listing_source_snapshot run_id=%d", run_id)
    return run_id


async def trigger_asking_to_sold_ratio_run(
    db: Session, schedule_row: dict[str, Any]
) -> int | None:
    """Создать scrape_runs + launch recompute_asking_to_sold_ratios в executor (sync DB-only task).

    Daily TRUE-MIRROR refresh асking→sold коэффициентов (#648 Stage 4): DELETE WHERE
    district='' + re-derive INSERT (та же 080-derivation) в одной транзакции, чтобы ratio
    не устаревал по мере ночного импорта ДКП. Mirror trigger_listing_source_snapshot_run:
    задача синхронная (чистый internal DB re-derive, БЕЗ внешних HTTP-вызовов / анти-бота),
    поэтому гоняем её в run_in_executor. SAFE to enable — schedule seed enabled=true (082),
    в отличие от avito/yandex sweep (dormant).

    Returns run_id (или None если skip — есть running run).
    """
    run_id = _claim_run(db, schedule_row)
    if run_id is None:
        return None

    async def _run() -> None:
        run_db = SessionLocal()
        try:
            # recompute_asking_to_sold_ratios is synchronous (DB-only, no async HTTP)
            from app.tasks.asking_to_sold_ratio import recompute_asking_to_sold_ratios

            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, recompute_asking_to_sold_ratios, run_db, run_id)
        except Exception:
            logger.exception(
                "scheduler: recompute_asking_to_sold_ratios 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 asking_to_sold_ratio_refresh 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: 'ros:dkp:' || id — плоский натуральный ключ (инъективный, без коллизий,
      human-readable). До #576 здесь был md5('ros:dkp:' || id); миграция 077 конвертировала
      существующие строки. source_id хранит исходный rosreestr id (дедуп переустанавливаем).
    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,
                        id                                       AS source_id_src,
                        '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, source_id, address, rooms, area_m2,
                                    floor, year_built, price_rub, price_per_m2, deal_date
                                )
                                VALUES (
                                    'rosreestr',
                                    CAST(:dedup_hash AS text),
                                    CAST(:source_id 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"],
                                "source_id": str(row["source_id_src"]),
                                "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(
        text(
            """
            SELECT id, source, enabled, window_start_hour, window_end_hour,
                   default_params, last_run_id, last_run_at, next_run_at
            FROM scrape_schedules
            WHERE enabled = true
              AND (next_run_at IS NULL OR next_run_at <= NOW())
            """
        ),
    ).mappings().all()
    return [dict(r) for r in rows]


async def scheduler_loop() -> None:
    """Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд."""
    logger.info("scheduler: started (tick=%ds)", SCHEDULER_TICK_SEC)
    # Initial sleep 30s чтобы дать FastAPI startup завершиться
    await asyncio.sleep(30)
    while True:
        try:
            db = SessionLocal()
            try:
                # Reap zombies first
                reap_zombies(db)
                # Process due schedules
                due = get_due_schedules(db)
                for sch in due:
                    source = sch["source"]
                    if source == "avito_city_sweep":
                        await trigger_avito_city_sweep_run(db, sch)
                    elif source == "yandex_city_sweep":
                        await trigger_yandex_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)
                    elif source == "listing_source_snapshot":
                        await trigger_listing_source_snapshot_run(db, sch)
                    elif source == "asking_to_sold_ratio_refresh":
                        await trigger_asking_to_sold_ratio_run(db, sch)
                    else:
                        logger.warning("scheduler: unknown source=%s, skip", source)
            finally:
                db.close()
        except Exception:
            logger.exception("scheduler: tick failed")
        await asyncio.sleep(SCHEDULER_TICK_SEC)