gendesign/backend/app/api/v1/admin_scrape.py

"""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, Header, 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

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


def _check_token(x_admin_token: str | None) -> None:
    if not settings.scrape_admin_token:
        raise HTTPException(
            status_code=503,
            detail="admin scrape disabled — set SCRAPE_ADMIN_TOKEN",
        )
    if x_admin_token != settings.scrape_admin_token:
        raise HTTPException(status_code=401, detail="invalid admin token")


@router.post("/kn")
def trigger_kn_sweep(
    payload: TriggerKnRequest,
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    _check_token(x_admin_token)
    # 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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> 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.
    """
    _check_token(x_admin_token)

    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,
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Принудительно удалить Redis-lock для (region, devs). Использовать когда
    зомби-worker оставил lock и новые задачи скипаются с reason='lock_held'."""
    _check_token(x_admin_token)
    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,
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Cancel a queued or running task by id."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    run_id: int | None = None,
    limit: int = 50,
) -> list[dict[str, Any]]:
    """Per-request failure log for manual browser verification."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    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."""
    _check_token(x_admin_token)
    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) ─────────────────────────────────


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,
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Запустить ETL Антоновского SQLite → наша PG.

    SQLite-файл должен быть доступен внутри Celery-worker контейнера:
      - на проде: смонтирован bind-mount-ом из /opt/gendesign/site-finder/
      - локально: путь передать через payload.sqlite_path
    """
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    limit: int = 20,
) -> list[dict[str, Any]]:
    _check_token(x_admin_token)
    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,
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Запустить НАШ sync_all_groups (тратит limits Объектива).

    В отличие от ETL (POST /objective) — это вызовы api.objctv.ru напрямую,
    с парсингом payload через app.workers.tasks.scrape_objective.
    Полезно когда нужно расширить покрытие за пределы Антоновского ЕКБ
    (Свердл.обл целиком + Челябинск + Тюмень + Пермь).
    """
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Текущая динамическая конфигурация Objective sync (single row)."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """PATCH-style update. После изменения cron_schedule нужен restart beat,
    остальные поля применяются на следующем sync-вызове автоматически."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Что у нас в БД сейчас + что лежит в SQLite Антона (size, last modified).

    Помогает понять стоит ли запускать ETL: если SQLite старше последнего
    успешного run'а — данные не изменятся.
    """
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> 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)
    """
    _check_token(x_admin_token)
    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_id: int = Field(default=2, ge=1, le=15, description="1=parcel, 2=quarter, 5=building")


@router.post("/geo/bulk")
def bulk_enqueue_geo(
    payload: BulkGeoEnqueueRequest,
    db: Annotated[Session, Depends(get_db)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Разбить pending cad-номера (region 66) на N чанков и запустить N geo-jobs параллельно.

    Логика выборки pending — те же правила что у source_kind='rosreestr_pending':
    quarter_cad_number из rosreestr_deals (region 66, ДДУ, тип 002001003000)
    которых нет в cad_quarters_geom.
    """
    _check_token(x_admin_token)
    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

    # 1) Собрать все pending cad-номера для region 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": [66], "bp": "00:00:%", "bs": "%:0000000"},
    ).all()

    all_cad: list[str] = [r[0] for r in rows]
    pending_total = len(all_cad)

    if pending_total == 0:
        raise HTTPException(status_code=400, detail="Нет cad-номеров для backfill")

    # 2) Разбить на чанки (numpy-style array_split — равные куски, хвост меньше)
    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) Создать job + enqueue для каждого чанка
    geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
    job_ids: list[int] = []
    for idx, chunk in enumerate(chunks):
        cad_with_thematic = [(c, payload.thematic_id) for c in chunk]
        job_id = enqueue_helper(
            name=f"bulk_svrd_{idx + 1}/{n_jobs}",
            job_kind="quarters",
            source_kind="rosreestr_pending_chunk",
            source_params={"region_codes": [66], "thematic_id": payload.thematic_id},
            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)

    return {
        "job_ids": job_ids,
        "targets_total": pending_total,
        "parallelism": n_jobs,
        "targets_per_job": chunk_size + (1 if remainder else 0),
    }


@router.get("/geo/jobs")
def list_geo_jobs(
    db: Annotated[Session, Depends(get_db)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    limit: int = 30,
) -> list[dict[str, Any]]:
    """Список последних geo-jobs (для UI dashboard)."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Пометить job как cancelled. Worker увидит при следующей итерации."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
) -> dict[str, Any]:
    """Re-enqueue paused/failed job. Resume idempotent через pending targets."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    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.
    """
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    scraper_type: str | None = None,
    run_id: int | None = None,
    limit: int = 200,
) -> list[dict[str, Any]]:
    """Унифицированный список логов (kn + nspd). Objective пока не пишет log."""
    _check_token(x_admin_token)
    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)],
    x_admin_token: Annotated[str | None, Header(alias="X-Admin-Token")] = None,
    limit: int = 20,
) -> list[dict[str, Any]]:
    _check_token(x_admin_token)
    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
    ]