"""Admin trigger for ad-hoc kn-API sweeps.

POST /api/v1/admin/scrape/kn
Body: {"region_code": 66, "developers": ["6208_0"]?, "fetch_flats": true?}

Auth: gendsgn.ru-wide Caddy basic_auth gate (PR #426). App-level X-Admin-Token
header removed 2026-05-23 — двойная auth избыточна для pilot.

Returns the Celery task id; poll /api/v1/admin/scrape/runs/{run_id} to track DB-side progress.
"""

from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Annotated, Any

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.db import get_db

logger = logging.getLogger(__name__)

router = APIRouter()


def _utcnow() -> datetime:
    """Timezone-aware UTC now. timestamptz-колонки приходят aware, поэтому naive
    datetime.utcnow() дал бы TypeError при вычитании — всегда aware."""
    return datetime.now(UTC)


class TriggerKnRequest(BaseModel):
    region_code: int = Field(..., ge=1, le=99)
    developers: list[str] | None = Field(default=None, max_length=20)
    fetch_flats: bool = True
    extras: bool = True
    download_photos: bool = False
    # Если True — ПЕРЕД apply_async удалить existing Redis lock. Эмердженси-режим
    # для случая «зомби-worker оставил lock и блокирует новые задачи».
    force: bool = False


@router.post("/kn")
def trigger_kn_sweep(
    payload: TriggerKnRequest,
) -> dict[str, Any]:
    # Defer Celery import so the API works without a broker in dev.
    from app.workers.tasks.scrape_kn import (
        force_release_lock,
        scrape_kn_region,
    )

    lock_released = False
    if payload.force:
        lock_released = force_release_lock(payload.region_code, payload.developers)

    result = scrape_kn_region.apply_async(
        args=[payload.region_code, payload.developers],
        kwargs={
            "skip_jitter": True,
            "extras": payload.extras,
            "download_photos": payload.download_photos,
            "fetch_flats": payload.fetch_flats,
        },
    )
    return {
        "task_id": result.id,
        "region_code": payload.region_code,
        "developers": payload.developers,
        "queued_at": "now",
        "force": payload.force,
        "lock_was_released": lock_released,
    }


@router.get("/queue")
def queue_status(
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Snapshot of Celery state, optimised for UI poll latency.

    - active: from kn_scrape_runs WHERE status='running' (instant, richer info
      than celery inspect — includes objects_count/flats_count/requests_count).
    - reserved + workers: celery inspect with short timeout (500 ms), called in
      a thread so a missing worker can't block the request.
    - queue_depth: Redis LLEN on default queue (single round-trip).
    - scheduled: omitted — we don't use ETA tasks.

    Worst-case latency ≈ 600 ms even if no worker is reachable, vs ~8 s with
    the previous 4×2 s blocking inspect calls.
    """

    import concurrent.futures
    import time

    from app.workers.celery_app import celery_app

    # 0) Reap zombies by heartbeat. After Resume_Checkpoint refactor the worker
    # writes heartbeat_at every ~30 sec (per 10 objects). 10 min без heartbeat
    # = точно мёртв. Используем COALESCE(heartbeat_at, started_at) чтобы
    # подхватить и legacy-runs (без heartbeat) старше 10 мин.
    db.execute(
        text(
            """
            UPDATE kn_scrape_runs
               SET status = 'zombie',
                   finished_at = NOW(),
                   error = 'auto-marked: no heartbeat 10+ min'
             WHERE status = 'running'
               AND finished_at IS NULL
               AND COALESCE(heartbeat_at, started_at)
                   < NOW() - INTERVAL '10 minutes'
            """
        )
    )
    db.commit()

    # 1) Active runs straight from DB — no broker round-trip.
    active_rows = (
        db.execute(
            text(
                """
                SELECT run_id, started_at, region_codes, developer_ids,
                       objects_count, flats_count, requests_count
                  FROM kn_scrape_runs
                 WHERE status = 'running'
                 ORDER BY started_at DESC
                 LIMIT 20
                """
            )
        )
        .mappings()
        .all()
    )
    active: list[dict[str, Any]] = [
        {
            "worker": "db",
            "task_id": f"run-{r['run_id']}",
            "name": "scrape_kn_region",
            "args": [list(r["region_codes"] or []), list(r["developer_ids"] or [])],
            "kwargs": {
                "objects": r["objects_count"],
                "flats": r["flats_count"],
                "requests": r["requests_count"],
            },
            "time_start": (int(r["started_at"].timestamp()) if r["started_at"] else None),
            "eta": None,
        }
        for r in active_rows
    ]

    # 2) Run reserved + ping in parallel with a tight timeout so the endpoint
    # never hangs when the broker is unreachable or worker is gone.
    inspect = celery_app.control.inspect(timeout=0.5)

    def _flatten(per_worker: dict[str, list[dict[str, Any]]] | None) -> list[dict[str, Any]]:
        if not per_worker:
            return []
        out: list[dict[str, Any]] = []
        for worker, tasks in per_worker.items():
            for t in tasks or []:
                out.append(
                    {
                        "worker": worker,
                        "task_id": t.get("id"),
                        "name": t.get("name"),
                        "args": t.get("args"),
                        "kwargs": t.get("kwargs"),
                        "time_start": t.get("time_start"),
                        "eta": t.get("eta"),
                    }
                )
        return out

    def _safe(fn):
        # Деградация (return None при недоступном broker) намеренна для UI-poll
        # эндпоинта, но логируем чтобы устойчиво сломанный broker был виден в логах.
        try:
            return fn()
        except Exception:
            logger.warning(
                "queue_status: celery inspect call %s failed",
                getattr(fn, "__name__", fn),
                exc_info=True,
            )
            return None

    deadline = time.monotonic() + 0.8
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as ex:
        f_reserved = ex.submit(_safe, inspect.reserved)
        f_ping = ex.submit(_safe, inspect.ping)
        try:
            reserved_raw = f_reserved.result(timeout=max(0.1, deadline - time.monotonic()))
        except concurrent.futures.TimeoutError:
            reserved_raw = None
        try:
            ping_resp = f_ping.result(timeout=max(0.1, deadline - time.monotonic()))
        except concurrent.futures.TimeoutError:
            ping_resp = None

    reserved = _flatten(reserved_raw)
    workers = list((ping_resp or {}).keys())

    # 3) Pending in broker queue (not yet picked up by any worker).
    queue_depth: int | None = None
    try:
        with celery_app.connection_or_acquire() as conn:
            with conn.channel() as channel:
                queue_depth = channel.client.llen("celery")
    except Exception:
        # Намеренная деградация для UI-poll; логируем чтобы недоступный broker
        # не был невидим в логах (см. .claude/rules/backend.md).
        logger.warning("queue_status: broker queue_depth probe failed", exc_info=True)
        queue_depth = None

    return {
        "workers": workers,
        "queue_depth": queue_depth,
        "active": active,
        "reserved": reserved,
        "scheduled": [],
    }


class ReleaseLockRequest(BaseModel):
    region_code: int = Field(..., ge=1, le=99)
    developers: list[str] | None = Field(default=None, max_length=20)


@router.post("/release-lock")
def release_lock(
    payload: ReleaseLockRequest,
) -> dict[str, Any]:
    """Принудительно удалить Redis-lock для (region, devs). Использовать когда
    зомби-worker оставил lock и новые задачи скипаются с reason='lock_held'."""
    from app.workers.tasks.scrape_kn import force_release_lock

    released = force_release_lock(payload.region_code, payload.developers)
    return {
        "region_code": payload.region_code,
        "developers": payload.developers,
        "released": released,
    }


class RevokeRequest(BaseModel):
    task_id: str = Field(..., min_length=1, max_length=255)
    terminate: bool = False  # SIGTERM running task vs just unmark from queue


@router.post("/revoke")
def revoke_task(
    payload: RevokeRequest,
) -> dict[str, Any]:
    """Cancel a queued or running task by id."""
    from app.workers.celery_app import celery_app

    celery_app.control.revoke(payload.task_id, terminate=payload.terminate, signal="SIGTERM")
    return {"task_id": payload.task_id, "revoked": True, "terminate": payload.terminate}


@router.get("/failures")
def list_failures(
    db: Annotated[Session, Depends(get_db)],
    run_id: int | None = None,
    limit: int = Query(default=50, ge=1),
) -> list[dict[str, Any]]:
    """Per-request failure log for manual browser verification."""
    where = ""
    params: dict[str, Any] = {"lim": min(limit, 200)}
    if run_id is not None:
        where = "WHERE run_id = :rid"
        params["rid"] = run_id
    rows = (
        db.execute(
            text(
                f"""
            SELECT failure_id, run_id, obj_id, endpoint, full_url,
                   http_status, error, body_preview, occurred_at
              FROM kn_scrape_failures
              {where}
             ORDER BY occurred_at DESC
             LIMIT :lim
            """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [
        {
            "failure_id": r["failure_id"],
            "run_id": r["run_id"],
            "obj_id": r["obj_id"],
            "endpoint": r["endpoint"],
            "full_url": r["full_url"],
            "http_status": r["http_status"],
            "error": r["error"],
            "body_preview": r["body_preview"],
            "occurred_at": r["occurred_at"].isoformat() if r["occurred_at"] else None,
        }
        for r in rows
    ]


@router.get("/logs")
def list_logs(
    db: Annotated[Session, Depends(get_db)],
    run_id: int | None = None,
    since_id: int | None = None,
    limit: int = Query(default=200, ge=0),
) -> list[dict[str, Any]]:
    """Per-run progress events. Use since_id to poll incrementally:
    pass the highest log_id seen → returns only newer rows."""
    where: list[str] = []
    params: dict[str, Any] = {"lim": min(limit, 1000)}
    if run_id is not None:
        where.append("run_id = :rid")
        params["rid"] = run_id
    if since_id is not None:
        where.append("log_id > :sid")
        params["sid"] = since_id
    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
    rows = (
        db.execute(
            text(
                f"""
                SELECT log_id, run_id, ts, level, stage, obj_id, message
                  FROM kn_scrape_log
                  {where_sql}
                 ORDER BY log_id DESC
                 LIMIT :lim
                """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [
        {
            "log_id": r["log_id"],
            "run_id": r["run_id"],
            "ts": r["ts"].isoformat() if r["ts"] else None,
            "level": r["level"],
            "stage": r["stage"],
            "obj_id": r["obj_id"],
            "message": r["message"],
        }
        for r in rows
    ]


# NSPD-скрапер УДАЛЁН 2026-05-11: вместо Playwright-based nspd_kn.py теперь
# используется bulk geo-fetcher через rosreestr2coord. См. /admin/scrape/geo
# и backend/app/workers/tasks/nspd_geo.py + nspd_lite.py (urllib).
# История прогонов осталась в nspd_scrape_runs/log + v_scrape_runs_unified.


# ── Objective ETL (Антоновский SQLite → PG) ─────────────────────────────────


@router.post("/noise-sync")
def trigger_noise_sync() -> dict[str, Any]:
    """Manual trigger для синхронизации OSM шумовых источников ЕКБ из Overpass API.

    Обычно запускается еженедельно через beat (понед 3:30 МСК).
    Этот endpoint — для ad-hoc запуска (например после первого деплоя или
    после добавления миграции 84_osm_noise_sources_ekb).
    """
    from app.workers.tasks.noise_sync import sync_osm_noise_sources_ekb

    result = sync_osm_noise_sources_ekb.apply_async()
    return {"task_id": result.id, "queued_at": "now"}


class HarvestQuarterRequest(BaseModel):
    """Параметры ручного запуска harvest одного квартала."""

    quarter_cad: str = Field(
        ...,
        min_length=5,
        pattern=r"^\d+:\d+:\d+$",
        description="3-сегментный кадастровый номер квартала, например '66:41:0204016'",
    )
    region_code: int = Field(default=66, ge=1, le=99)
    include_zouit: bool = Field(default=True)
    include_risks: bool = Field(default=False)


@router.post("/nspd/harvest-quarter")
def trigger_harvest_quarter(
    payload: HarvestQuarterRequest,
) -> dict[str, Any]:
    """Manual trigger одного quarter harvest (для testing / конкретного квартала).

    Запускает harvest_quarter Celery task и возвращает task_id.
    Для получения результата — poll через Celery inspect или нождите строку
    в nspd_quarter_dumps WHERE quarter_cad = payload.quarter_cad.
    """
    from app.workers.tasks.nspd_sync import harvest_quarter

    task = harvest_quarter.apply_async(
        args=[payload.quarter_cad, payload.region_code],
        kwargs={
            "include_zouit": payload.include_zouit,
            "include_risks": payload.include_risks,
        },
    )
    return {
        "task_id": task.id,
        "quarter_cad": payload.quarter_cad,
        "region_code": payload.region_code,
        "include_zouit": payload.include_zouit,
        "include_risks": payload.include_risks,
    }


@router.post("/pzz-sync")
def trigger_pzz_sync() -> dict[str, Any]:
    """Manual trigger для импорта ПЗЗ территориальных зон ЕКБ из Росреестр PKK6.

    Запускать после деплоя миграции 85_pzz_zones_ekb.sql и при необходимости
    обновить данные о зонировании (обычно раз в несколько месяцев).
    """
    from app.workers.tasks.pzz_sync import sync_pzz_zones_ekb

    result = sync_pzz_zones_ekb.apply_async()
    return {"task_id": result.id, "queued_at": "now"}


class TriggerZoneRegulationsBackfillRequest(BaseModel):
    """Параметры ручного backfill ПЗЗ-регламента терзон ЕКБ (эпик #1881)."""

    limit: int | None = Field(
        default=None,
        ge=1,
        le=1000,
        description="Максимум уникальных незакэшированных зон к фетчу (None = все). Smoke-тест.",
    )
    rate_delay_s: float = Field(
        default=1.0,
        ge=0.0,
        le=10.0,
        description="Пауза (сек) между live-fetch'ами зон — вежливость к геопорталу.",
    )


@router.post("/zone-regulations/backfill")
def trigger_zone_regulations_backfill(
    payload: TriggerZoneRegulationsBackfillRequest,
) -> dict[str, Any]:
    """Backfill кэша ПЗЗ-регламента ВСЕХ терзон ЕКБ → zone_regulation_cache (эпик #1881).

    Чтобы financial_estimate фаерил для большинства участков (max_far из кэша, без live-urbanCard
    в hot-пути). Идемпотентно: уже закэшированные зоны пропускаются, повторный запуск дёшев.

    estimate_minutes — груба: ~120 уникальных терзон ЕКБ × (rate_delay + ~2с live-urbanCard).
    """
    from app.workers.tasks.backfill_zone_regulations import backfill_zone_regulations

    result = backfill_zone_regulations.apply_async(
        kwargs={
            "limit": payload.limit,
            "rate_delay_s": payload.rate_delay_s,
        },
    )
    # ~120 уникальных терзон ЕКБ; на зону ≈ rate_delay + ~2с urbanCard round-trip.
    n_zones = payload.limit if payload.limit is not None else 120
    estimate_minutes = round(n_zones * (payload.rate_delay_s + 2.0) / 60.0, 1)
    return {
        "task_id": result.id,
        "limit": payload.limit,
        "rate_delay_s": payload.rate_delay_s,
        "estimate_minutes": estimate_minutes,
        "queued_at": "now",
    }


@router.post("/poi-sync")
def trigger_poi_sync() -> dict[str, Any]:
    """Manual trigger для синхронизации OSM POI ЕКБ из Overpass API.

    Обычно запускается еженедельно через beat (понед 3:00 МСК).
    Этот endpoint — для ad-hoc запуска (например после первого деплоя
    или при создании нового тестового участка).
    """
    from app.workers.tasks.poi_sync import sync_osm_poi_ekb

    result = sync_osm_poi_ekb.apply_async()
    return {"task_id": result.id, "queued_at": "now"}


@router.post("/gisogd-permits-sync")
def trigger_gisogd_permits_sync() -> dict[str, Any]:
    """Manual trigger инкрементальной загрузки РНС/РВЭ ГИСОГД-СО → gisogd_permits (#2367).

    Обычно запускается еженедельно через beat (вторник 06:30 МСК). Этот endpoint —
    для ad-hoc запуска (например после деплоя миграции 187_gisogd_permits.sql или для
    внеочередного обновления). Инкрементально: карточки тянутся только для новых/
    изменившихся документов, повторный запуск дёшев.
    """
    from app.workers.tasks.gisogd_permits_sync import sync_gisogd_permits

    result = sync_gisogd_permits.apply_async()
    return {"task_id": result.id, "queued_at": "now"}


class TriggerObjectiveEtlRequest(BaseModel):
    sqlite_path: str | None = Field(
        default=None,
        description="Override settings.objective_anton_sqlite_path (abs file path)",
    )


@router.post("/objective")
def trigger_objective_etl(
    payload: TriggerObjectiveEtlRequest,
) -> dict[str, Any]:
    """Запустить ETL Антоновского SQLite → наша PG.

    SQLite-файл должен быть доступен внутри Celery-worker контейнера:
      - на проде: смонтирован bind-mount-ом из /opt/gendesign/site-finder/
      - локально: путь передать через payload.sqlite_path
    """
    from app.workers.tasks.objective_etl import import_anton_objective

    result = import_anton_objective.apply_async(
        kwargs={
            "triggered_by": "manual",
            "sqlite_path": payload.sqlite_path,
        },
    )
    return {
        "task_id": result.id,
        "sqlite_path": payload.sqlite_path or settings.objective_anton_sqlite_path,
    }


@router.get("/objective/runs")
def list_objective_runs(
    db: Annotated[Session, Depends(get_db)],
    limit: int = Query(default=20, ge=0),
) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
                SELECT run_id, started_at, finished_at, heartbeat_at,
                       group_name, reports_ok, reports_failed,
                       rows_corpus_room, rows_lots, rows_history,
                       requests_count, status, error, triggered_by
                  FROM objective_scrape_runs
                 ORDER BY started_at DESC
                 LIMIT :lim
                """
            ),
            {"lim": min(limit, 100)},
        )
        .mappings()
        .all()
    )
    return [
        {
            "run_id": r["run_id"],
            "started_at": r["started_at"].isoformat() if r["started_at"] else None,
            "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
            "heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None,
            "group_name": r["group_name"],
            "reports_ok": r["reports_ok"],
            "reports_failed": r["reports_failed"],
            "rows_corpus_room": r["rows_corpus_room"],
            "rows_lots": r["rows_lots"],
            # Поле в Schema 68 v2 называется rows_history — мы переиспользуем его
            # под количество загруженных mapping'ов из jk_objective_match.
            "rows_mappings": r["rows_history"],
            "requests_count": r["requests_count"],
            "status": r["status"],
            "error": r["error"],
            "triggered_by": r["triggered_by"],
        }
        for r in rows
    ]


class TriggerObjectiveSyncRequest(BaseModel):
    """Ручной триггер нашего sync_all_groups (платный API Объектива)."""

    groups: list[str] | None = Field(
        default=None,
        description="Override settings.objective_sync_groups (csv). "
        "Если None — берёт дефолт из settings.",
        max_length=20,
    )


@router.post("/objective/sync-our")
def trigger_objective_sync_our(
    payload: TriggerObjectiveSyncRequest,
) -> dict[str, Any]:
    """Запустить НАШ sync_all_groups (тратит limits Объектива).

    В отличие от ETL (POST /objective) — это вызовы api.objctv.ru напрямую,
    с парсингом payload через app.workers.tasks.scrape_objective.
    Полезно когда нужно расширить покрытие за пределов Антоновского ЕКБ
    (Свердл.обл целиком + Челябинск + Тюмень + Пермь).
    """
    if not settings.objective_api_key:
        raise HTTPException(
            status_code=503,
            detail="OBJECTIVE_API_KEY не задан (settings.objective_api_key пуст)",
        )
    from app.workers.tasks.scrape_objective import sync_all_groups

    groups_eff = payload.groups or [
        g.strip() for g in settings.objective_sync_groups.split(",") if g.strip()
    ]
    result = sync_all_groups.apply_async(
        kwargs={
            "groups": payload.groups,  # None → task возьмёт дефолт
            "triggered_by": "manual",
        },
    )
    return {
        "task_id": result.id,
        "groups_planned": groups_eff,
        "estimate_minutes": len(groups_eff)
        * 3,  # ~3 мин на группу (corp_sum + lots_pf + 30s pause)
    }


class ObjectiveConfigUpdateRequest(BaseModel):
    """PATCH-style — обновляет только переданные поля."""

    cron_schedule: str | None = None
    groups_csv: str | None = None
    use_ddu: bool | None = None
    use_dkp: bool | None = None
    period_months_back: int | None = Field(default=None, ge=1, le=24)
    inter_group_delay_s: int | None = Field(default=None, ge=0, le=300)
    rate_ms: int | None = Field(default=None, ge=0, le=10000)
    retries: int | None = Field(default=None, ge=0, le=5)


@router.get("/objective/config")
def get_objective_config(
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Текущая динамическая конфигурация Objective sync (single row)."""
    from app.services.objective_sync_config import get_config

    return get_config(db).to_dict()


@router.put("/objective/config")
def update_objective_config(
    payload: ObjectiveConfigUpdateRequest,
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """PATCH-style update. После изменения cron_schedule нужен restart beat,
    остальные поля применяются на следующем sync-вызове автоматически."""
    from app.services.objective_sync_config import update_config

    cfg = update_config(
        db,
        cron_schedule=payload.cron_schedule,
        groups_csv=payload.groups_csv,
        use_ddu=payload.use_ddu,
        use_dkp=payload.use_dkp,
        period_months_back=payload.period_months_back,
        inter_group_delay_s=payload.inter_group_delay_s,
        rate_ms=payload.rate_ms,
        retries=payload.retries,
        updated_by="admin-ui",
    )
    return {
        **cfg.to_dict(),
        "_note": (
            "cron_schedule applied только после restart beat-контейнера; "
            "остальные параметры подхватятся при следующем запуске sync_all_groups"
        ),
    }


@router.get("/objective/coverage")
def objective_coverage(
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Что у нас в БД сейчас + что лежит в SQLite Антона (size, last modified).

    Помогает понять стоит ли запускать ETL: если SQLite старше последнего
    успешного run'а — данные не изменятся.
    """
    from app.services.objective_etl import get_sqlite_info

    # #1964: `lots` = СЫРОЙ COUNT(*) — намеренно. Этот счётчик сравнивается с
    # SQLite-источником Антона (ETL-fidelity: «свежее ли SQLite, чем БД»), а SQLite
    # хранит сырые строки → дедуп-view сломал бы сравнение. Рядом отдаём
    # `lots_physflat` (v_objective_lots_latest — по физлоту) для прозрачности
    # инфляции (~2.91×: сырых ~1.75M vs физлотов ~603k).
    pg_row = (
        db.execute(
            text(
                """
                SELECT
                    (SELECT COUNT(*) FROM objective_lots) AS lots,
                    (SELECT COUNT(*) FROM v_objective_lots_latest) AS lots_physflat,
                    (SELECT COUNT(*) FROM objective_corpus_room_month) AS crm,
                    (SELECT COUNT(*) FROM objective_complex_mapping) AS mappings,
                    (SELECT MAX(snapshot_date) FROM objective_lots) AS last_lot_snapshot,
                    (SELECT MAX(finished_at) FROM objective_scrape_runs
                       WHERE status = 'done'
                         AND triggered_by IN ('manual','beat','api')) AS last_run_at
                """
            )
        )
        .mappings()
        .first()
    )
    sqlite_info = get_sqlite_info(settings.objective_anton_sqlite_path)
    return {
        "pg": {
            "lots": int(pg_row["lots"] or 0),
            "lots_physflat": int(pg_row["lots_physflat"] or 0),
            "corp_room_month": int(pg_row["crm"] or 0),
            "mappings": int(pg_row["mappings"] or 0),
            "last_lot_snapshot": pg_row["last_lot_snapshot"].isoformat()
            if pg_row["last_lot_snapshot"]
            else None,
            "last_run_at": pg_row["last_run_at"].isoformat() if pg_row["last_run_at"] else None,
        },
        "sqlite": sqlite_info,
    }


# ── NSPD geo bulk-jobs (resume-friendly) ────────────────────────────────────


class EnqueueGeoJobRequest(BaseModel):
    """Создать новый bulk geo-job."""

    name: str | None = Field(default=None, max_length=200)
    job_kind: str = Field(..., pattern="^(quarters|parcels|buildings|mixed)$")
    source_kind: str = Field(
        default="manual_list", pattern="^(manual_list|rosreestr_pending|region_bbox)$"
    )
    cad_nums: list[str] = Field(
        default_factory=list, max_length=50000, description="Список cad-номеров для обработки"
    )
    thematic_id: int = Field(default=2, ge=1, le=15, description="1=parcel, 2=quarter, 5=building")
    region_codes: list[int] | None = Field(
        default=None, max_length=20, description="Для source=rosreestr_pending"
    )
    rate_ms: int = Field(default=600, ge=100, le=10000)


@router.post("/geo/jobs")
def enqueue_geo_job(
    payload: EnqueueGeoJobRequest,
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Создать bulk geo-job и поставить в очередь.

    source_kind:
      - manual_list: cad_nums передан явно
      - rosreestr_pending: автоматически набрать из rosreestr_deals.quarter_cad_number
        которых ещё нет в cad_quarters_geom (для region_codes если задано)
      - region_bbox: TODO (для будущего spatial-bulk)
    """
    from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper
    from app.workers.tasks.nspd_geo import process_nspd_geo_job

    cad_with_thematic: list[tuple[str, int]] = []
    if payload.source_kind == "manual_list":
        cad_with_thematic = [
            (c.strip(), payload.thematic_id) for c in payload.cad_nums if c.strip()
        ]
    elif payload.source_kind == "rosreestr_pending":
        regions = payload.region_codes or [66]
        rows = db.execute(
            text(
                """
                SELECT DISTINCT quarter_cad_number AS cad
                  FROM rosreestr_deals
                 WHERE region_code = ANY(:rc)
                   AND doc_type = 'ДДУ'
                   AND realestate_type_code = '002001003000'
                   AND quarter_cad_number IS NOT NULL
                   AND quarter_cad_number NOT LIKE :bp
                   AND quarter_cad_number NOT LIKE :bs
                   AND NOT EXISTS (SELECT 1 FROM cad_quarters_geom g
                                    WHERE g.cad_number = rosreestr_deals.quarter_cad_number)
                 LIMIT 50000
                """
            ),
            {"rc": regions, "bp": "00:00:%", "bs": "%:0000000"},
        ).all()
        cad_with_thematic = [(r[0], 2) for r in rows]
    else:
        raise HTTPException(
            status_code=400,
            detail=f"source_kind={payload.source_kind} not implemented yet",
        )

    if not cad_with_thematic:
        raise HTTPException(status_code=400, detail="no targets to process")

    job_id = enqueue_helper(
        name=payload.name,
        job_kind=payload.job_kind,
        source_kind=payload.source_kind,
        source_params={"region_codes": payload.region_codes, "thematic_id": payload.thematic_id},
        cad_nums_with_thematic=cad_with_thematic,
        rate_ms=payload.rate_ms,
        triggered_by="manual",
    )
    from app.services.job_settings import get_setting_value

    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
    process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
    return {
        "job_id": job_id,
        "targets_total": len(cad_with_thematic),
        "estimate_minutes": round(len(cad_with_thematic) * payload.rate_ms / 1000 / 60, 1),
    }


class BulkGeoEnqueueRequest(BaseModel):
    """Параметры для параллельного backfill geo по Свердловской обл."""

    parallelism: int = Field(default=5, ge=1, le=10)
    thematic_ids: list[int] = Field(
        default=[2],
        description="Список thematic_id: 1=parcel, 2=quarter, 5=building. Можно несколько.",
    )
    region_codes: list[int] = Field(default=[66], description="Список region кодов")
    only_ddu: bool = Field(
        default=False,
        description="True = только ДДУ (002001003000); False = все cad-кварталы",
    )
    source: str = Field(
        default="rosreestr_pending",
        pattern="^(rosreestr_pending|all_in_region)$",
        description=(
            "rosreestr_pending — cad-номера из rosreestr_deals которых нет в geo-таблице; "
            "all_in_region — UNION всех cad из rosreestr_deals + cad_buildings + complexes"
        ),
    )
    rate_ms: int = Field(default=600, ge=100, le=10000)


# Маппинг thematic_id → таблица проверки существования + колонка + job_kind-метка
_THEMATIC_META: dict[int, dict[str, str]] = {
    1: {"exists_table": "cad_parcels_geom", "exists_col": "cad_num", "label": "parcels"},
    2: {"exists_table": "cad_quarters_geom", "exists_col": "cad_number", "label": "quarters"},
    5: {"exists_table": "cad_buildings", "exists_col": "cad_num", "label": "buildings"},
}


def _collect_pending_cad(
    db: Session,
    thematic_id: int,
    region_codes: list[int],
    only_ddu: bool,
) -> list[str]:
    """Собрать cad-номера из rosreestr_deals которых нет в соответствующей geo-таблице."""
    meta = _THEMATIC_META.get(thematic_id)
    if meta is None:
        raise HTTPException(
            status_code=400,
            detail=f"thematic_id={thematic_id} не поддерживается (допустимы: 1, 2, 5)",
        )
    ddu_filter = (
        "AND doc_type = 'ДДУ' AND realestate_type_code = '002001003000'" if only_ddu else ""
    )
    exists_table = meta["exists_table"]
    exists_col = meta["exists_col"]
    rows = db.execute(
        text(
            f"""
            SELECT DISTINCT quarter_cad_number AS cad
              FROM rosreestr_deals
             WHERE region_code = ANY(:rc)
               {ddu_filter}
               AND quarter_cad_number IS NOT NULL
               AND quarter_cad_number NOT LIKE :bp
               AND quarter_cad_number NOT LIKE :bs
               AND NOT EXISTS (
                   SELECT 1 FROM {exists_table} g
                    WHERE g.{exists_col} = rosreestr_deals.quarter_cad_number
               )
             LIMIT 50000
            """
        ),
        {"rc": region_codes, "bp": "00:00:%", "bs": "%:0000000"},
    ).all()
    return [r[0] for r in rows]


def _collect_all_in_region(
    db: Session,
    region_codes: list[int],
) -> list[str]:
    """UNION всех cad-номеров из rosreestr_deals + cad_buildings + complexes.cad_quarter
    с фильтром по region prefix (66xx для кодов региона 66)."""
    rows = db.execute(
        text("""
            SELECT DISTINCT cad FROM (
                SELECT quarter_cad_number AS cad
                  FROM rosreestr_deals
                 WHERE quarter_cad_number IS NOT NULL
                   AND region_code = ANY(:rc)
                UNION
                SELECT cad_num AS cad
                  FROM cad_buildings
                 WHERE cad_num IS NOT NULL
                   AND cad_num ~ :region_re
                UNION
                SELECT cad_quarter AS cad
                  FROM complexes
                 WHERE cad_quarter IS NOT NULL
                   AND cad_quarter ~ :region_re
            ) sub
            WHERE cad NOT LIKE :bp
              AND cad NOT LIKE :bs
            LIMIT 100000
        """),
        {
            "rc": region_codes,
            "region_re": "^(" + "|".join(str(rc) for rc in region_codes) + "):",
            "bp": "00:00:%",
            "bs": "%:0000000",
        },
    ).all()
    return [r[0] for r in rows]


@router.post("/geo/bulk")
def bulk_enqueue_geo(
    payload: BulkGeoEnqueueRequest,
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Разбить pending cad-номера на N чанков и запустить N×len(thematic_ids) geo-jobs.

    source=rosreestr_pending: для каждого thematic_id выбирает cad из rosreestr_deals
      которых нет в соответствующей geo-таблице.
    source=all_in_region: UNION из rosreestr_deals + cad_buildings + complexes — один
      набор для всех thematic_ids (каждый thematic_id обрабатывает весь список).
    """
    from app.services.job_settings import get_setting_value
    from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper
    from app.workers.tasks.nspd_geo import process_nspd_geo_job

    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")

    # Валидируем ВСЕ thematic_ids ДО любых сайд-эффектов (создание jobs / apply_async),
    # иначе невалидный id в середине списка приводит к partial execution: для предыдущих
    # валидных id строки в nspd_geo_jobs уже созданы и задачи улетели в очередь geo,
    # а клиент получает 400 без идемпотентного отката (#1562).
    invalid_ids = [tid for tid in payload.thematic_ids if tid not in _THEMATIC_META]
    if invalid_ids:
        raise HTTPException(
            status_code=400,
            detail=f"thematic_id={invalid_ids} не поддерживается (допустимы: 1, 2, 5)",
        )

    jobs_summary: list[dict[str, Any]] = []

    for thematic_id in payload.thematic_ids:
        meta = _THEMATIC_META[thematic_id]

        # 1) Собрать cad-номера
        if payload.source == "rosreestr_pending":
            all_cad = _collect_pending_cad(db, thematic_id, payload.region_codes, payload.only_ddu)
        else:  # all_in_region
            all_cad = _collect_all_in_region(db, payload.region_codes)

        if not all_cad:
            jobs_summary.append(
                {
                    "thematic_id": thematic_id,
                    "job_ids": [],
                    "targets_total": 0,
                    "parallelism": 0,
                    "note": "нет cad-номеров для backfill",
                }
            )
            continue

        # 2) Разбить на чанки
        pending_total = len(all_cad)
        n_jobs = min(payload.parallelism, pending_total)
        chunk_size, remainder = divmod(pending_total, n_jobs)
        chunks: list[list[str]] = []
        start = 0
        for i in range(n_jobs):
            end = start + chunk_size + (1 if i < remainder else 0)
            chunks.append(all_cad[start:end])
            start = end

        # 3) Создать jobs
        job_ids: list[int] = []
        label = meta["label"]
        for idx, chunk in enumerate(chunks):
            cad_with_thematic = [(c, thematic_id) for c in chunk]
            job_id = enqueue_helper(
                name=f"bulk_{label}_t{thematic_id}_{idx + 1}/{n_jobs}",
                job_kind=label,
                source_kind=f"{payload.source}_chunk",
                source_params={
                    "region_codes": payload.region_codes,
                    "thematic_id": thematic_id,
                    "source": payload.source,
                },
                cad_nums_with_thematic=cad_with_thematic,
                rate_ms=payload.rate_ms,
                triggered_by="bulk_admin",
            )
            process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
            job_ids.append(job_id)

        jobs_summary.append(
            {
                "thematic_id": thematic_id,
                "job_ids": job_ids,
                "targets_total": pending_total,
                "parallelism": n_jobs,
            }
        )

    if not any(j["targets_total"] > 0 for j in jobs_summary):
        raise HTTPException(status_code=400, detail="Нет cad-номеров для backfill (все thematic)")

    return {"jobs": jobs_summary}


@router.get("/geo/jobs")
def list_geo_jobs(
    db: Annotated[Session, Depends(get_db)],
    limit: int = Query(default=30, ge=0),
) -> list[dict[str, Any]]:
    """Список последних geo-jobs (для UI dashboard)."""
    rows = (
        db.execute(
            text(
                """
            SELECT job_id, name, job_kind, source_kind, status, triggered_by,
                   started_at, finished_at, heartbeat_at,
                   targets_total, targets_done, targets_failed, targets_skipped,
                   requests_count, waf_blocked_count, error,
                   rate_ms, created_at
              FROM nspd_geo_jobs
             ORDER BY created_at DESC
             LIMIT :lim
            """
            ),
            {"lim": min(limit, 100)},
        )
        .mappings()
        .all()
    )
    return [
        {
            "job_id": r["job_id"],
            "name": r["name"],
            "job_kind": r["job_kind"],
            "source_kind": r["source_kind"],
            "status": r["status"],
            "triggered_by": r["triggered_by"],
            "started_at": r["started_at"].isoformat() if r["started_at"] else None,
            "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
            "heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None,
            "targets_total": r["targets_total"],
            "targets_done": r["targets_done"],
            "targets_failed": r["targets_failed"],
            "targets_skipped": r["targets_skipped"],
            "requests_count": r["requests_count"],
            "waf_blocked_count": r["waf_blocked_count"],
            "error": r["error"],
            "rate_ms": r["rate_ms"],
            "created_at": r["created_at"].isoformat() if r["created_at"] else None,
            "progress_pct": round(
                100.0 * (r["targets_done"] or 0) / max(r["targets_total"] or 1, 1), 1
            ),
        }
        for r in rows
    ]


@router.post("/geo/jobs/{job_id}/cancel")
def cancel_geo_job(
    job_id: int,
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Пометить job как cancelled. Worker увидит при следующей итерации."""
    db.execute(
        text(
            """
            UPDATE nspd_geo_jobs SET status = 'cancelled', finished_at = NOW(),
                                     error = COALESCE(error, 'cancelled by admin')
             WHERE job_id = :id AND status IN ('queued','running','paused')
            """
        ),
        {"id": job_id},
    )
    db.commit()
    return {"job_id": job_id, "cancelled": True}


@router.post("/geo/jobs/{job_id}/resume")
def resume_geo_job(
    job_id: int,
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """Re-enqueue paused/failed job. Resume idempotent через pending targets."""
    from app.services.job_settings import get_setting_value
    from app.workers.tasks.nspd_geo import process_nspd_geo_job

    db.execute(
        text("UPDATE nspd_geo_jobs SET status='queued', error=NULL WHERE job_id=:id"),
        {"id": job_id},
    )
    db.commit()
    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
    process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
    return {"job_id": job_id, "resumed": True}


# ── Newbuilding cross-load ETL (#976) ────────────────────────────────────────


@router.post("/newbuilding-crossload")
def trigger_newbuilding_crossload() -> dict[str, Any]:
    """Ручной запуск cross-load ETL tradein.houses → newbuilding_listings (#976).

    Обычно запускается ночью через beat (03:30 МСК).
    Если TRADEIN_DATABASE_URL не задан в settings — возвращает 503.
    """
    if not settings.tradein_database_url:
        raise HTTPException(
            status_code=503,
            detail=(
                "TRADEIN_DATABASE_URL не задан — cross-load ETL отключён. "
                "Добавь переменную в backend/.env.runtime и перезапусти worker."
            ),
        )
    from app.workers.tasks.etl_newbuilding_crossload import etl_newbuilding_crossload

    result = etl_newbuilding_crossload.apply_async()
    return {"task_id": result.id, "queued_at": "now"}


# ── Unified scrape dashboard (поверх v_scrape_runs_unified / v_scrape_log_unified) ──


@router.get("/all/runs")
def list_all_runs(
    db: Annotated[Session, Depends(get_db)],
    scraper_type: str | None = None,
    limit: int = Query(default=30, ge=0),
) -> list[dict[str, Any]]:
    """Унифицированный список прогонов (kn + nspd + objective).

    Поверх v_scrape_runs_unified. Если scraper_type не задан — все 3.
    Используется главной dashboard /admin/scrape/all.
    """
    where = "" if not scraper_type else "WHERE scraper_type = :st"
    params: dict[str, Any] = {"lim": min(limit, 200)}
    if scraper_type:
        params["st"] = scraper_type
    rows = (
        db.execute(
            text(
                f"""
                SELECT scraper_type, run_id, started_at, finished_at, heartbeat_at,
                       status, error, triggered_by, scope, requests_count,
                       items_ok, items_failed, sub_items_ok,
                       progress_idx, progress_total, extra
                  FROM v_scrape_runs_unified
                  {where}
                 ORDER BY started_at DESC NULLS LAST
                 LIMIT :lim
                """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [
        {
            "scraper_type": r["scraper_type"],
            "run_id": r["run_id"],
            "started_at": r["started_at"].isoformat() if r["started_at"] else None,
            "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
            "heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None,
            "status": r["status"],
            "error": r["error"],
            "triggered_by": r["triggered_by"],
            "scope": r["scope"],
            "requests_count": r["requests_count"],
            "items_ok": r["items_ok"],
            "items_failed": r["items_failed"],
            "sub_items_ok": r["sub_items_ok"],
            "progress_idx": r["progress_idx"],
            "progress_total": r["progress_total"],
            "extra": r["extra"],
        }
        for r in rows
    ]


@router.get("/all/logs")
def list_all_logs(
    db: Annotated[Session, Depends(get_db)],
    scraper_type: str | None = None,
    run_id: int | None = None,
    limit: int = Query(default=200, ge=0),
) -> list[dict[str, Any]]:
    """Унифицированный список логов (kn + nspd). Objective пока не пишет log."""
    where: list[str] = []
    params: dict[str, Any] = {"lim": min(limit, 1000)}
    if scraper_type:
        where.append("scraper_type = :st")
        params["st"] = scraper_type
    if run_id is not None:
        where.append("run_id = :rid")
        params["rid"] = run_id
    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
    rows = (
        db.execute(
            text(
                f"""
                SELECT scraper_type, log_id, run_id, ts, level, stage, entity_id, message
                  FROM v_scrape_log_unified
                  {where_sql}
                 ORDER BY log_id DESC
                 LIMIT :lim
                """
            ),
            params,
        )
        .mappings()
        .all()
    )
    return [
        {
            "scraper_type": r["scraper_type"],
            "log_id": r["log_id"],
            "run_id": r["run_id"],
            "ts": r["ts"].isoformat() if r["ts"] else None,
            "level": r["level"],
            "stage": r["stage"],
            "entity_id": r["entity_id"],
            "message": r["message"],
        }
        for r in rows
    ]


# ── Data-freshness monitoring (#73 B5-3) ─────────────────────────────────────
#
# GET /api/v1/admin/scrape/freshness — для каждого источника данных:
#   last_success_at / last_attempt_at / last_status / objects_updated_24h|7d /
#   age_days / freshness (fresh|stale|critical) / status (ok|stale|failed).
#
# Источник истины — run-таблицы (НЕ контент-таблицы): на проде проверено, что
# идемпотентный re-scrape НЕ бампит complexes.updated_at (newest 2026-05-11 при
# kn-run 3 дня назад), поэтому "objects_updated" считаем по счётчикам прогонов
# (objects_count / quarters_ok / rows_lots / targets_done) за done-прогоны в окне.
#
# Пороги (fresh/stale/critical в днях) выведены из beat-расписания каждого job'а
# (app/workers/beat_schedule.py). Sentry-алёрты по критичным job'ам шлёт отдельный
# beat-task scrape_freshness_check (tasks.scrape_freshness_check) — endpoint только
# читает, не алертит, чтобы UI-поллинг не спамил issue.


class FreshnessSource(BaseModel):
    """Описание одного источника данных в реестре freshness-мониторинга."""

    source: str  # стабильный ключ (kn / nspd / objective / nspd_geo / cadastre)
    label: str  # человекочитаемое имя для UI
    table: str  # run-таблица ЛИБО data-таблица (если задан timestamp_col)
    # Колонка-счётчик "сколько объектов обновил done-прогон". Агрегируется как
    # COALESCE(SUM(<work_col>) FILTER (...), 0) — FILTER применяется к SUM, не к COALESCE.
    # В timestamp-режиме (timestamp_col задан) work_col игнорируется.
    work_col: str
    started_col: str = "started_at"
    # Колонка-fallback для last_attempt когда started_at NULL (queued-but-not-run).
    attempt_fallback_col: str | None = None
    # Data-table режим: если задан, источник — это ПЛОСКАЯ data-таблица (не run-ledger,
    # без status/started/finished). Свежесть = MAX(timestamp_col); upd_24h/_7d = COUNT(*)
    # по строкам в окне. work_col/started_col/attempt_fallback_col при этом игнорируются.
    timestamp_col: str | None = None
    fresh_days: float = 7.0  # < fresh_days → fresh
    stale_days: float = 30.0  # < stale_days → stale, иначе critical
    critical: bool = False  # критичный источник (для Sentry-алёртов в beat-таске)
    # True → источник, который прогнался успешно И недавно (был бы fresh по возрасту),
    # НО SUM(work_col) по done-прогонам цикла (окно fresh_days) < min_output_rows,
    # ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-почти-ничего-не-извлёк"
    # (напр. WAF-break / тихий частичный extract: status=done, но flats_count суммарно
    # 9 при здоровых ~376k). По умолчанию OFF — включаем только там, где малый выход =
    # поломка (не legitimate-low, как nspd/nspd_geo/cadastre). Применимо только к
    # run-ledger (не data-table) режиму.
    alert_on_zero_output: bool = False
    # Пол (floor) для recent_output = SUM(work_col) по done-прогонам ЦИКЛА в окне
    # fresh_days (НЕ один прогон — цикл объединяет несколько прогонов: objective по
    # group_name, kn по region_codes+resume). Проверяется ТОЛЬКО при
    # alert_on_zero_output=True: recent_output < min_output_rows → даунгрейд (строгое <).
    # Окно = fresh_days, поэтому здоровый недельный прогон возрастом 7–8d остаётся
    # внутри окна и не ложно-срабатывает (#1947 fix). Default 1 → флагует только если
    # суммарный выход цикла = 0 (безопасный минимальный catch).
    min_output_rows: int = 1


# Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на
# проде (information_schema) — статусы done присутствуют. nspd — data-table режим:
# смотрит nspd_quarter_dumps.fetched_at_utc (живой harvest harvest_stale_quarters +
# write-back /analyze-резолвера #1891), а НЕ defunct ledger nspd_scrape_runs (3 WAF-
# забаненных manual-прогона 2026-04-30, никогда не шедулился → ложный "no successful runs").
# Пороги:
#   kn          — beat ~еженедельно (scrape_kn cron, Mon) → fresh<8d, stale<14d
#   kn_flats    — та же run-таблица, work_col=flats_count + alert_on_zero_output +
#                 min_output_rows=50000 (issue #1945: kn-run status=done но flats_count
#                 мал/нестабилен — монитор смотрел objects_count ~1548, не flats_count;
#                 #1947 fix: floor сравнивается с SUM(flats_count) по done-прогонам цикла
#                 в окне fresh_days=8d — ловит low-output 9/2296/3670 и не ложит на 7–8d
#                 aging) → fresh<8d, stale<14d
#   objective   — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d
#   nspd        — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d
#   nspd_geo    — ручной bulk geo-job → fresh<7d, stale<30d
#   cadastre    — ручной cadastre-job → fresh<14d, stale<45d
_FRESHNESS_SOURCES: list[FreshnessSource] = [
    FreshnessSource(
        source="kn",
        label="DOM.РФ kn-API (ЖК)",
        table="kn_scrape_runs",
        work_col="objects_count",
        # kn шедулится ЕЖЕНЕДЕЛЬНО (settings.scrape_kn_cron='15 4 * * mon', Mon 04:15 МСК),
        # поэтому 2/5 дней давали ложный "stale" каждую среду-воскресенье. Калибруем под
        # недельную каденцию (как objective 7/14): fresh до 8 дней (полная неделя + добег
        # понедельничного прогона), stale при пропуске ~двух понедельников.
        fresh_days=8.0,
        stale_days=14.0,
        critical=True,
    ),
    FreshnessSource(
        # kn-run пишет ДВА счётчика: objects_count (ЖК, ~1548 — выше) и flats_count
        # (квартиры). Источник истины для "лоадер прогнался, но почти ничего не извлёк"
        # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count мал/нестабилен,
        # монитор молчал — objects_count оставался здоровым). Та же run-таблица и пороги,
        # что у kn, но work_col=flats_count + alert_on_zero_output + min_output_rows.
        #
        # Floor сравнивается с recent_output = SUM(flats_count) по ВСЕМ done-прогонам в
        # окне fresh_days=8d (kn пишет строку на region_codes + resume — это НЕ один
        # прогон). Магнитуды (прод gendesign-postgres-1, #1947): healthy цикл ~376k
        # квартир (snapshot 05-17=376604); сломанное состояние — суммарно ≤3670, сейчас
        # SUM(flats_count,8d)=9 (latest done = resume, считает только хвост; ground truth
        # snapshot 2026-06-22=9, сломан 5+ недель). Порог 50000 чисто разделяет:
        # healthy ≫ 50k ≫ broken ≤3670.
        #
        # KNOWN LIMITATION: zombie-resume, загрузивший лишь малый хвост БЕЗ полного
        # прогона в окне, может транзиентно недосчитать → максимум одно-цикловый ложный
        # алёрт, самоисцеляется следующим полным прогоном. Приемлемо (а в текущем реальном
        # resume-кейсе snapshot реально сломан=9 → флаг корректен).
        source="kn_flats",
        label="DOM.РФ kn-API (квартиры)",
        table="kn_scrape_runs",
        work_col="flats_count",
        fresh_days=8.0,
        stale_days=14.0,
        critical=True,
        alert_on_zero_output=True,
        min_output_rows=50000,
    ),
    FreshnessSource(
        source="objective",
        label="Объектив (лоты/цены)",
        table="objective_scrape_runs",
        work_col="rows_lots",
        fresh_days=7.0,
        stale_days=14.0,
        critical=True,
        # Объектив-sync, прогнавшийся с почти нулём лотов — тоже поломка (тихий
        # частичный extract), а не легитимный low. objective пишет строку на group_name
        # (4 группы: Свердловская/Тюмень/Пермь/Челябинск), поэтому floor сравнивается с
        # recent_output = SUM(rows_lots) по группам в окне fresh_days=7d (прод: SUM 7d=
        # 946264; даже наименьшая отдельная группа Челябинск=68387). Floor 1000 ловит
        # вырожденный цикл с огромным запасом, не задевая нормальный разброс — и
        # робастен к тому, КАКАЯ группа дописалась последней (single-newest был бы
        # хрупок: 324570 или 68387 в зависимости от порядка финиша).
        alert_on_zero_output=True,
        min_output_rows=1000,
    ),
    FreshnessSource(
        source="nspd",
        label="NSPD quarter dumps (harvest+resolver)",
        # Data-table режим: nspd_quarter_dumps — это контент-таблица квартальных
        # дампов (NO status/started/finished). Живой harvest пишет fetched_at_utc;
        # defunct nspd_scrape_runs (manual WAF-ban 2026-04-30) больше НЕ источник истины.
        table="nspd_quarter_dumps",
        timestamp_col="fetched_at_utc",
        # В timestamp-режиме не используется — оставляем валидное имя колонки.
        work_col="total_features",
        # Медленный кадастровый + lazy-refresh источник: дампы освежаются по мере
        # устаревания кварталов / on-demand из /analyze. Текущий возраст ~4d → fresh.
        fresh_days=14.0,
        stale_days=30.0,
    ),
    FreshnessSource(
        source="nspd_geo",
        label="NSPD geo bulk-fetcher",
        table="nspd_geo_jobs",
        work_col="targets_done",
        attempt_fallback_col="created_at",
        # On-demand источник БЕЗ cron (admin UI / CLI / lazy из analyze) — штучные
        # фетчи по активности пользователя. fresh_days=7 флагал каждую неделю
        # тишины ложным stale-алертом (расследование 2026-07-04); пороги — под
        # реальную каденцию, critical=False и так не трогает overall.
        fresh_days=30.0,
        stale_days=90.0,
    ),
    FreshnessSource(
        source="cadastre",
        label="Кадастровые jobs",
        table="cadastre_jobs",
        work_col="targets_done",
        attempt_fallback_col="created_at",
        fresh_days=14.0,
        stale_days=45.0,
    ),
    FreshnessSource(
        # #2367: реестр РНС/РВЭ ГИСОГД-СО. Data-table режим (плоская таблица без run-
        # ledger): свежесть = MAX(fetched_at), upd_24h/_7d = COUNT(*) по окну. Источник
        # обновляется ежедневно, тянем еженедельно (beat вторник) → fresh<14d, stale<45d
        # (широкий запас на пропуск одного-двух вторников, как cadastre). critical=False.
        source="gisogd_permits",
        label="ГИСОГД-СО РНС/РВЭ (реестр разрешений)",
        table="gisogd_permits",
        timestamp_col="fetched_at",
        # В timestamp-режиме work_col не используется — валидное имя колонки.
        work_col="id",
        fresh_days=14.0,
        stale_days=45.0,
    ),
]


def _classify_freshness(age_days: float | None, fresh_days: float, stale_days: float) -> str:
    """fresh / stale / critical / unknown по возрасту последнего успешного прогона."""
    if age_days is None:
        return "unknown"
    if age_days < fresh_days:
        return "fresh"
    if age_days < stale_days:
        return "stale"
    return "critical"


def compute_freshness(db: Session) -> dict[str, Any]:
    """Свежесть данных по каждому источнику (для dashboard + alerting).

    Чистая функция (один источник истины для HTTP-эндпоинта и beat-таски
    scrape_freshness_check). Принимает открытую сессию, НЕ коммитит/закрывает её —
    жизненным циклом сессии управляет вызывающий (FastAPI-зависимость или таска).

    Для run-ledger источников один проход по run-таблице агрегирует:
      - last_success_at  — max(COALESCE(finished_at, started_at)) WHERE status='done'
      - last_attempt_at  — max(started_at) (или created_at fallback) по всем прогонам
      - last_status      — status последнего прогона
      - objects_updated_24h / _7d — сумма run-счётчика по done-прогонам в окне
      - recent_output    — SUM(work_col) по done-прогонам цикла в окне fresh_days (NULL
                           если успехов в окне нет)
      - age_days         — возраст last_success в днях (NULL если успехов не было)
      - freshness        — fresh / stale / critical / unknown (по порогам источника)
      - status           — ok / stale / failed (для UI-светофора)
      - reason           — пояснение к output-floor-даунгрейду (иначе None)

    Output-floor (issue #1945, redesigned #1947): для run-ledger источников с
    alert_on_zero_output=True источник, который иначе был бы fresh→ok (недавний
    успешный прогон), но SUM(work_col) по ВСЕМ done-прогонам цикла в окне fresh_days
    (recent_output) < min_output_rows, ДАУНГРЕЙДИТСЯ до failed — ловит
    "лоадер-прогнался-но-почти-ничего-не-извлёк" (kn flats_count суммарно 9 при
    здоровых ~376k). Cycle-SUM (не один прогон) робастен к multi-run циклам: objective
    пишет строку на group_name, kn — на region_codes + resume. Окно fresh_days (не
    хардкод 7d) → здоровый недельный прогон 7–8d остаётся в окне (нет aging-FP).
    Только когда возраст НЕ покрывает поломку (status=='ok'), чтобы не двойнить.

    Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps)
    нет run-ledger семантики (status/started/finished отсутствуют), поэтому:
      - last_success_at = last_attempt_at = MAX(timestamp_col)
      - objects_updated_24h / _7d = COUNT(*) строк, обновлённых в окне
      - last_status = NULL (косметика только для run-ledger'ов)
    Остальной downstream (age_days / _classify_freshness / status-маппинг) — общий.

    `overall_status` агрегирует худший статус по критичным источникам.
    """
    sources_out: list[dict[str, Any]] = []
    worst_critical = "ok"
    severity = {"ok": 0, "stale": 1, "failed": 2}

    for src in _FRESHNESS_SOURCES:
        # table / column-имена — из доверенного серверного реестра (НЕ из
        # пользовательского ввода), поэтому f-string в SQL здесь безопасен;
        # все временные границы передаются параметрами (:d1/:d7).
        if src.timestamp_col is not None:
            # Data-table режим: плоская контент-таблица без run-ledger семантики
            # (нет status/started/finished). Свежесть = MAX(timestamp_col),
            # upd_24h/_7d = COUNT(*) строк, обновлённых в окне. last_status=NULL
            # (косметика только для run-ledger'ов).
            ts = src.timestamp_col
            row = (
                db.execute(
                    text(
                        f"""
                        SELECT
                            MAX({ts})                               AS last_success_at,
                            MAX({ts})                               AS last_attempt_at,
                            COALESCE(COUNT(*) FILTER (
                                WHERE {ts} > NOW() - CAST(:d1 AS interval)
                            ), 0)                                   AS upd_24h,
                            COALESCE(COUNT(*) FILTER (
                                WHERE {ts} > NOW() - CAST(:d7 AS interval)
                            ), 0)                                   AS upd_7d,
                            NULL                                    AS last_status
                          FROM {src.table}
                        """
                    ),
                    {"d1": "24 hours", "d7": "7 days"},
                )
                .mappings()
                .first()
            )
        else:
            started = src.started_col
            attempt_expr = (
                f"MAX(COALESCE({started}, {src.attempt_fallback_col}))"
                if src.attempt_fallback_col
                else f"MAX({started})"
            )
            # last_status — отдельным скаляром: status последнего по времени прогона.
            order_col = (
                f"COALESCE({started}, {src.attempt_fallback_col})"
                if src.attempt_fallback_col
                else started
            )
            # FILTER применяется к SUM(...), а COALESCE(...,0) оборачивает результат —
            # COALESCE(...) FILTER(...) — синтаксическая ошибка в PostgreSQL.
            row = (
                db.execute(
                    text(
                        f"""
                        SELECT
                            MAX(COALESCE(finished_at, {started}))
                                FILTER (WHERE status = 'done')      AS last_success_at,
                            {attempt_expr}                          AS last_attempt_at,
                            COALESCE(SUM({src.work_col}) FILTER (
                                WHERE status = 'done'
                                  AND COALESCE(finished_at, {started})
                                      > NOW() - CAST(:d1 AS interval)
                            ), 0)                                   AS upd_24h,
                            COALESCE(SUM({src.work_col}) FILTER (
                                WHERE status = 'done'
                                  AND COALESCE(finished_at, {started})
                                      > NOW() - CAST(:d7 AS interval)
                            ), 0)                                   AS upd_7d,
                            (
                                SELECT status FROM {src.table}
                                 ORDER BY {order_col} DESC NULLS LAST
                                 LIMIT 1
                            )                                       AS last_status,
                            -- Cycle-sum: SUM(work_col) по ВСЕМ done-прогонам в окне
                            -- fresh_days. Логический цикл загрузки = НЕСКОЛЬКО прогонов
                            -- (objective — по строке на group_name, kn — по region_codes
                            -- + resume), поэтому суммируем, а не берём один прогон.
                            -- Окно = fresh_days (per-source), а НЕ хардкод 7d: здоровый
                            -- недельный прогон возрастом 7–8d остаётся ВНУТРИ окна →
                            -- нет aging-FP (#1947). Inner COALESCE(work_col,0) — NULL-
                            -- счётчик на done-прогоне вносит 0, не NULL. fresh_days —
                            -- float из доверенного реестра → f-string безопасен. Окно
                            -- через make_interval(secs => fresh_days*86400): secs —
                            -- double precision, поэтому дробный fresh_days валиден (у
                            -- days int-only → 7.5 упал бы). NB без двоеточийного cast.
                            SUM(COALESCE({src.work_col}, 0)) FILTER (
                                WHERE status = 'done'
                                  AND COALESCE(finished_at, {started})
                                      > now()
                                        - make_interval(secs => {src.fresh_days} * 86400)
                            )                                       AS recent_output
                          FROM {src.table}
                        """
                    ),
                    {"d1": "24 hours", "d7": "7 days"},
                )
                .mappings()
                .first()
            )

        last_success = row["last_success_at"] if row else None
        last_attempt = row["last_attempt_at"] if row else None
        last_status = row["last_status"] if row else None
        # SUM(work_col) по done-прогонам в окне fresh_days (run-ledger режим = весь
        # цикл загрузки). В data-table режиме колонки нет → None; .get() терпит
        # отсутствие ключа в mapping-строке. None также когда done-прогонов в окне нет.
        recent_output_raw = row.get("recent_output") if row else None
        recent_output: int | None = (
            int(recent_output_raw) if recent_output_raw is not None else None
        )

        age_days: float | None = None
        if last_success is not None:
            age_seconds = (_utcnow() - last_success).total_seconds()
            age_days = round(age_seconds / 86400.0, 2)

        freshness = _classify_freshness(age_days, src.fresh_days, src.stale_days)

        upd_24h = int(row["upd_24h"] or 0) if row else 0
        upd_7d = int(row["upd_7d"] or 0) if row else 0

        # status (UI-светофор): нет успехов вообще → failed; иначе мапим freshness.
        if last_success is None:
            status = "failed"
        elif freshness == "fresh":
            status = "ok"
        elif freshness == "stale":
            status = "stale"
        else:  # critical
            status = "failed"

        # Output-floor downgrade (issue #1945, redesigned #1947): источник прогнался
        # успешно И недавно (был бы fresh→ok по возрасту), НО SUM(work_col) по всем
        # done-прогонам ЦИКЛА (окно fresh_days) ниже порога min_output_rows → ДАУНГРЕЙД
        # до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила.
        #   - cycle-SUM (а не один прогон): цикл = несколько done-прогонов (objective —
        #     по group_name, kn — по region_codes + resume), сумма даёт магнитуду цикла
        #     (objective ~946k, kn_flats ~376k healthy) → робастно к multi-run.
        #   - окно fresh_days (а не хардкод 7d): здоровый недельный прогон 7–8d остаётся
        #     внутри окна → нет aging-FP первой версии #1947.
        # Только run-ledger (timestamp_col is None) c alert_on_zero_output=True и ТОЛЬКО
        # когда источник иначе был бы fresh/ok (по возрасту stale/failed уже покрыт — не
        # двойним). status=='ok' гарантирует недавний успех (last_success is not None).
        # Строгое '<': SUM ровно на пороге → ok (порог = минимально-приемлемый объём).
        reason: str | None = None
        if (
            src.alert_on_zero_output
            and src.timestamp_col is None
            and status == "ok"
            and recent_output is not None
            and recent_output < src.min_output_rows
        ):
            status = "failed"
            reason = (
                f"low-output: done-прогоны за {src.fresh_days:g}d дали "
                f"SUM({src.work_col})={recent_output} < порога {src.min_output_rows} "
                f"(#1945, #1947)"
            )

        if src.critical and severity[status] > severity[worst_critical]:
            worst_critical = status

        sources_out.append(
            {
                "source": src.source,
                "label": src.label,
                "critical": src.critical,
                "last_success_at": last_success.isoformat() if last_success else None,
                "last_attempt_at": last_attempt.isoformat() if last_attempt else None,
                "last_status": last_status,
                "objects_updated_24h": upd_24h,
                "objects_updated_7d": upd_7d,
                # SUM(work_col) по done-прогонам цикла в окне fresh_days (None для
                # data-table режима / без done-прогонов в окне). Additive — база для
                # output-floor-проверки, полезно и для UI-диагностики.
                "recent_output": recent_output,
                "age_days": age_days,
                "thresholds": {"fresh_days": src.fresh_days, "stale_days": src.stale_days},
                "freshness": freshness,
                "status": status,
                # Доп. контекст при output-floor-даунгрейде (None в обычном случае).
                # Additive — не ломает существующий UI/JSON-контракт.
                "reason": reason,
            }
        )

    return {
        "generated_at": _utcnow().isoformat(),
        "overall_status": worst_critical,
        "sources": sources_out,
    }


@router.get("/freshness")
def scrape_freshness(
    db: Annotated[Session, Depends(get_db)],
) -> dict[str, Any]:
    """GET-обёртка над compute_freshness — только читает, не алертит.

    Вся агрегация живёт в compute_freshness (один источник истины); эндпоинт
    лишь прокидывает сессию и возвращает результат без изменений формы payload.
    Sentry-алёрты по критичным источникам шлёт beat-таска scrape_freshness_check.
    """
    return compute_freshness(db)


# ── DomRF (kn) — runs list ───────────────────────────────────────────────────


@router.get("/runs")
def list_runs(
    db: Annotated[Session, Depends(get_db)],
    limit: int = Query(default=20, ge=0),
) -> list[dict[str, Any]]:
    rows = (
        db.execute(
            text(
                """
            SELECT run_id, started_at, finished_at, region_codes, developer_ids,
                   objects_count, flats_count, requests_count, status, error, snapshot_date
              FROM kn_scrape_runs
             ORDER BY started_at DESC
             LIMIT :lim
            """
            ),
            {"lim": min(limit, 100)},
        )
        .mappings()
        .all()
    )
    return [
        {
            "run_id": r["run_id"],
            "started_at": r["started_at"].isoformat() if r["started_at"] else None,
            "finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
            "region_codes": list(r["region_codes"]) if r["region_codes"] else [],
            "developer_ids": list(r["developer_ids"]) if r["developer_ids"] else [],
            "objects_count": r["objects_count"],
            "flats_count": r["flats_count"],
            "requests_count": r["requests_count"],
            "status": r["status"],
            "error": r["error"],
            "snapshot_date": r["snapshot_date"].isoformat() if r["snapshot_date"] else None,
        }
        for r in rows
    ]


# ── Ekburg construction permits (РНС/РВЭ ЕКБ) ────────────────────────────────


class TriggerEkburgPermitsRequest(BaseModel):
    year: int | None = Field(
        default=None,
        ge=2022,
        le=2030,
        description=(
            "Если None — refresh_all (все 5 лет 2022-2026). "
            "Иначе refresh_year(year) для конкретного года."
        ),
    )


@router.post("/ekburg-permits")
def trigger_ekburg_permits(
    payload: TriggerEkburgPermitsRequest,
) -> dict[str, Any]:
    """Manual trigger для синхронизации РНС/РВЭ ЕКБ из екатеринбург.рф (xlsx).

    Обычно запускается раз в месяц через beat (1-го числа 02:00 UTC).
    Этот endpoint — для ad-hoc запуска (после первого деплоя или для smoke-теста).

    - year=None → refresh_all: скачивает 5 xlsx (2022-2026), ~3-8 мин.
    - year=2026 → refresh_year(2026): один файл, ~1 мин.
    """
    from app.workers.tasks.ekburg_permits_sync import refresh_all, refresh_year

    if payload.year is None:
        result = refresh_all.apply_async()
        scope = "all_years_2022_2026"
    else:
        result = refresh_year.apply_async(args=[payload.year])
        scope = f"year_{payload.year}"

    return {"task_id": result.id, "scope": scope, "queued_at": "now"}


class TriggerKnCatalogObjectsRequest(BaseModel):
    region_code: int = Field(default=66, ge=1, le=99)
    max_objects: int | None = Field(default=None, ge=1, le=2000)
    force: bool = Field(
        default=False,
        description=(
            "True — игнорировать фильтр 'уже сегодня обновлён' и грузить ВСЕ объекты "
            "последнего snapshot ('Загрузить все'). По умолчанию пропускает то, "
            "что уже скраплено сегодня."
        ),
    )


@router.post("/kn-catalog-objects")
def trigger_kn_catalog_objects(
    payload: TriggerKnCatalogObjectsRequest,
) -> dict[str, Any]:
    """Manual trigger для catalog-OBJECT scraper (заполняет wall_type, energy_eff,
    ceiling_height_m, parking_*, playground_*, scores из SSR __NEXT_DATA__).

    Beat schedule: Tuesday 04:00 UTC, batch 300/run. Этот endpoint — для ad-hoc
    запуска (smoke-тест после деплоя или повторный pass для свежесозданных
    объектов до next beat fire).

    - max_objects=None → дефолтный лимит таска (300).
    - max_objects=3 → smoke-тест.
    - force=True → "Загрузить все": игнорирует skip-today, грузит всё подряд.
    """
    from app.workers.tasks.scrape_kn_catalog_objects import scrape_kn_catalog_objects

    kwargs: dict[str, Any] = {
        "region_code": payload.region_code,
        "force": payload.force,
    }
    if payload.max_objects is not None:
        kwargs["max_objects"] = payload.max_objects
    result = scrape_kn_catalog_objects.apply_async(kwargs=kwargs)

    return {
        "task_id": result.id,
        "region_code": payload.region_code,
        "max_objects": payload.max_objects,
        "force": payload.force,
        "queued_at": "now",
    }