"""Shared scheduler helpers (#2397 Part C — legacy scheduler_loop removed).

История: до #2192 этот модуль нёс полный in-app asyncio scheduler (`scheduler_loop`,
27-веточный `if/elif` dispatch, `trigger_*_run` per-source launchers, `get_due_schedules`,
`reap_zombies`, `_claim_run` advisory-lock claim, `_spawn_tracked`/`_drain_inflight`
graceful-drain machinery). #2192 представил kit-путь (`scraper_kit.orchestration.scheduler`
+ `app.services.product_handlers`) как ship-dark alternative; прод переключился на него
(`USE_KIT_SCHEDULER=true` в tradein-scraper, см. docker-compose.prod.yml). #2397 Part C
убрал legacy `scheduler_loop` и все `trigger_*`/dispatch-функции — kit теперь единственный
scheduling-путь (`app/scheduler_main.py` безусловно запускает `_run_kit_scheduler()`).

Что осталось в этом модуле — НЕ scheduler-loop, а функции с живыми потребителями вне
удалённой machinery:
  - `compute_next_run_at`  — читается admin.py (операторский предпросмотр "next run").
  - `has_running_run`      — читается admin.py (UI-индикатор "уже бежит").
  - `import_rosreestr_dkp` — job-тело, вызываемое kit-handler'ом
                              product_handlers._job_rosreestr_dkp (lazy import).
  - `_execute_cian_backfill` — job-тело, вызываемое kit-handler'ом
                              product_handlers._job_cian_history_backfill (lazy import).

Zombie-reap, advisory-lock claim и tick-loop теперь целиком в
`scraper_kit/orchestration/scheduler.py` (см. test_scraper_kit_scheduler_parity.py).
"""

from __future__ import annotations

import logging
import random
from datetime import UTC, datetime, time, timedelta
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.shutdown import shutdown_requested
from app.services import scrape_runs as runs_mod

logger = logging.getLogger(__name__)


def compute_next_run_at(
    window_start_hour: int,
    window_end_hour: int,
    *,
    now: datetime | None = None,
    interval_days: int = 1,
) -> datetime:
    """Pick random datetime в window [start, end) UTC, через interval_days суток после now.

    interval_days задаёт каденс источника: 1 (default) = daily (back-compat), 7 = weekly.
    Берётся из schedule.default_params["interval_days"] вызывающим кодом; отсутствие ключа
    → 1 → прежнее ежедневное поведение.

    Если 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)
    interval_days = max(1, int(interval_days))
    # Целевая дата = now + interval_days суток (interval_days=1 → завтра, как раньше).
    target = (now + timedelta(days=interval_days)).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(target, 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:
            # interval_days=1: текущая дата (если окно ещё не наступило сегодня) или next day.
            # interval_days>1: всегда целевая дата (стаггер на N суток вперёд).
            today_ok = interval_days == 1 and now.hour < window_start_hour
            base_date = now.date() if today_ok else target
            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(target, 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


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


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:
            cancelled = runs_mod.is_cancelled(db, run_id)
            if cancelled or shutdown_requested():
                if cancelled:
                    # User-cancel: семантика без изменений — mark_cancelled.
                    logger.info("rosreestr_dkp_import run_id=%d: cancelled by user", run_id)
                    runs_mod.mark_cancelled(db, run_id)
                    return
                # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
                # Это НЕ user-cancel → mark_done (partial), не mark_cancelled. Курсор
                # уже зафиксирован update_heartbeat'ом каждый батч; следующий run
                # пере-сканирует с id=0 (INSERT ... ON CONFLICT DO NOTHING идемпотентен)
                # → уже вставленные строки пропускаются, остаток до-импортируется.
                # mark_done выводит run из 'running' → reap_zombies его не тронет.
                logger.info(
                    "rosreestr_dkp_import run_id=%d: SIGTERM-drain — committing partial "
                    "(last_id=%d, batches=%d) and exiting",
                    run_id,
                    last_id,
                    total_batches,
                )
                runs_mod.update_heartbeat(db, run_id, counters)
                runs_mod.mark_done(db, run_id, counters)
                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(
                                -- #1525: '-?[0-9]+' сохраняет знак минус, иначе цоколь/подвал
                                -- '-1' импортируется как 1. Для '5/9' по-прежнему берётся 5.
                                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