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

POST /api/v1/admin/scrape/kn
Body: {"region_code": 66, "developers": ["6208_0"]?, "fetch_flats": true?}
Header: X-Admin-Token: <SCRAPE_ADMIN_TOKEN env var>

Disabled if SCRAPE_ADMIN_TOKEN is empty (default).
Returns the Celery task id; poll /api/v1/admin/scrape/runs/{run_id} to track DB-side progress.
"""

from __future__ import annotations

from typing import Annotated, Any

from fastapi import APIRouter, Depends, HTTPException
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
from app.core.deps import AdminTokenAuth

router = APIRouter()


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,
    _: AdminTokenAuth,
) -> 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": result.date_done.isoformat() if result.date_done else None,
        "force": payload.force,
        "lock_was_released": lock_released,
    }


@router.get("/queue")
def queue_status(
    db: Annotated[Session, Depends(get_db)],
    _: AdminTokenAuth,
) -> 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):
        try:
            return fn()
        except Exception:
            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:
        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,
    _: AdminTokenAuth,
) -> 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,
    _: AdminTokenAuth,
) -> 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)],
    _: AdminTokenAuth,
    run_id: int | None = None,
    limit: int = 50,
) -> 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)],
    _: AdminTokenAuth,
    run_id: int | None = None,
    since_id: int | None = None,
    limit: int = 200,
) -> 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(
    _: AdminTokenAuth,
) -> 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,
    _: AdminTokenAuth,
) -> 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(
    _: AdminTokenAuth,
) -> 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"}


@router.post("/poi-sync")
def trigger_poi_sync(
    _: AdminTokenAuth,
) -> 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"}


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,
    _: AdminTokenAuth,
) -> 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)],
    _: AdminTokenAuth,
    limit: int = 20,
) -> 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,
    _: AdminTokenAuth,
) -> 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)],
    _: AdminTokenAuth,
) -> 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)],
    _: AdminTokenAuth,
) -> 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)],
    _: AdminTokenAuth,
) -> dict[str, Any]:
    """Что у нас в БД сейчас + что лежит в SQLite Антона (size, last modified).

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

    pg_row = (
        db.execute(
            text(
                """
                SELECT
                    (SELECT COUNT(*) FROM objective_lots) AS lots,
                    (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),
            "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)],
    _: AdminTokenAuth,
) -> 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"
        ),
    )


# Маппинг 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)],
    _: AdminTokenAuth,
) -> 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")

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

    for thematic_id in payload.thematic_ids:
        if thematic_id not in _THEMATIC_META:
            raise HTTPException(
                status_code=400,
                detail=f"thematic_id={thematic_id} не поддерживается (допустимы: 1, 2, 5)",
            )
        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,
                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)],
    _: AdminTokenAuth,
    limit: int = 30,
) -> 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)],
    _: AdminTokenAuth,
) -> 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)],
    _: AdminTokenAuth,
) -> 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}


# ── 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)],
    _: AdminTokenAuth,
    scraper_type: str | None = None,
    limit: int = 30,
) -> 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)],
    _: AdminTokenAuth,
    scraper_type: str | None = None,
    run_id: int | None = None,
    limit: int = 200,
) -> 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
    ]


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


@router.get("/runs")
def list_runs(
    db: Annotated[Session, Depends(get_db)],
    _: AdminTokenAuth,
    limit: int = 20,
) -> 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
    ]