"""Admin API для bulk cadastre harvest (GH issue #168, PR 3/5). Endpoints: POST /api/v1/admin/cadastre/jobs — создать job + enqueue GET /api/v1/admin/cadastre/jobs — список jobs GET /api/v1/admin/cadastre/jobs/{id} — детали job POST /api/v1/admin/cadastre/jobs/{id}/cancel — отменить job POST /api/v1/admin/cadastre/jobs/{id}/resume — возобновить paused job Scope variants: pilot — первые 50 кварталов ЕКБ (66:41:%) ekb_full — все ~2408 кварталов ЕКБ manual_list — явный список quarters в body.quarters Auth: gendsgn.ru-wide Caddy basic_auth gate (PR #426). App-level X-Admin-Token header removed 2026-05-23 — двойная auth избыточна для pilot. """ from __future__ import annotations import json import logging 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.db import get_db logger = logging.getLogger(__name__) router = APIRouter() # Ограничение пилота PILOT_LIMIT = 50 # Регион по умолчанию — ЕКБ (66:41:%) EKB_PREFIX = "66:41:%" # ── Request / response schemas ─────────────────────────────────────────────── class CreateCadastreJobRequest(BaseModel): """Параметры создания bulk cadastre job.""" scope: str = Field( ..., pattern="^(pilot|ekb_full|manual_list)$", description="pilot=50 кварталов ЕКБ, ekb_full=все 2408, manual_list=явный список", ) quarters: list[str] | None = Field( default=None, max_length=5000, description="Список кварталов для scope=manual_list", ) limit: int | None = Field( default=None, ge=1, le=10000, description="Ограничить количество кварталов (override для pilot/ekb_full)", ) name: str | None = Field(default=None, max_length=200) rate_ms: int = Field( default=333, ge=50, le=10000, description="Задержка между запросами в мс (333мс = ~3 req/s)", ) # ── Helpers ────────────────────────────────────────────────────────────────── def _resolve_quarters( db: Session, scope: str, quarters: list[str] | None, limit: int | None, ) -> list[str]: """Собрать список кварталов согласно scope.""" if scope == "manual_list": if not quarters: raise HTTPException( status_code=400, detail="scope=manual_list требует непустой список quarters", ) return [q.strip() for q in quarters if q.strip()] cap = limit or (PILOT_LIMIT if scope == "pilot" else 100000) rows = db.execute( text(""" SELECT cad_number FROM cad_quarters_geom WHERE cad_number LIKE :prefix ORDER BY cad_number LIMIT :cap """), {"prefix": EKB_PREFIX, "cap": cap}, ).all() if not rows: raise HTTPException( status_code=400, detail=( f"Нет кварталов с префиксом {EKB_PREFIX!r} в cad_quarters_geom. " "Убедитесь что geo-backfill кварталов завершён." ), ) return [str(r[0]) for r in rows] def _create_job( db: Session, *, name: str | None, scope: str, quarters: list[str], rate_ms: int, triggered_by: str = "manual", ) -> int: """INSERT в cadastre_jobs, вернуть job_id.""" job_id = db.execute( text(""" INSERT INTO cadastre_jobs ( name, job_kind, scope, status, rate_ms, triggered_by, targets_total, raw_targets ) VALUES ( :name, :job_kind, :scope, 'queued', :rate_ms, :triggered_by, :targets_total, CAST(:raw_targets AS jsonb) ) RETURNING job_id """), { "name": name, "job_kind": scope, "scope": scope, "rate_ms": rate_ms, "triggered_by": triggered_by, "targets_total": len(quarters), "raw_targets": json.dumps({"quarters": quarters}, ensure_ascii=False), }, ).scalar_one() db.commit() return int(job_id) def _serialize_job(row: Any) -> dict[str, Any]: """Сериализовать строку cadastre_jobs в dict для API response.""" targets_total = row["targets_total"] or 0 targets_done = row["targets_done"] or 0 return { "job_id": row["job_id"], "name": row["name"], "job_kind": row["job_kind"], "scope": row["scope"], "status": row["status"], "triggered_by": row["triggered_by"], "rate_ms": row["rate_ms"], "created_at": row["created_at"].isoformat() if row["created_at"] else None, "started_at": row["started_at"].isoformat() if row["started_at"] else None, "finished_at": row["finished_at"].isoformat() if row["finished_at"] else None, "heartbeat_at": row["heartbeat_at"].isoformat() if row["heartbeat_at"] else None, "targets_total": targets_total, "targets_done": targets_done, "targets_failed": row["targets_failed"] or 0, "targets_skipped": row["targets_skipped"] or 0, "requests_count": row["requests_count"] or 0, "waf_blocked_count": row["waf_blocked_count"] or 0, "error": row["error"], "phase_state": row["phase_state"], "progress_pct": round(100.0 * targets_done / max(targets_total, 1), 1), "estimate_minutes": round(targets_total * (row["rate_ms"] or 333) / 1000 / 60, 1), } # ── Endpoints ──────────────────────────────────────────────────────────────── @router.post("/jobs") def create_cadastre_job( body: CreateCadastreJobRequest, db: Annotated[Session, Depends(get_db)], ) -> dict[str, Any]: """Создать bulk cadastre harvest job и поставить в очередь. scope=pilot → первые 50 ЕКБ кварталов из cad_quarters_geom scope=ekb_full → все ~2408 ЕКБ кварталов scope=manual_list → body.quarters (явный список) Возвращает: {job_id, targets_total, estimate_minutes} """ from app.workers.tasks.scrape_cadastre import enqueue_cadastre_harvest quarters = _resolve_quarters(db, body.scope, body.quarters, body.limit) job_id = _create_job( db, name=body.name or f"cadastre_{body.scope}", scope=body.scope, quarters=quarters, rate_ms=body.rate_ms, ) # Enqueue таск который в свою очередь запустит per-quarter tasks enqueue_cadastre_harvest.apply_async( args=[job_id], queue="geo", ) estimate_minutes = round(len(quarters) * body.rate_ms / 1000 / 60, 1) logger.info( "create_cadastre_job: job_id=%d scope=%s quarters=%d estimate=%.1fmin", job_id, body.scope, len(quarters), estimate_minutes, ) return { "job_id": job_id, "scope": body.scope, "targets_total": len(quarters), "estimate_minutes": estimate_minutes, } @router.get("/jobs") def list_cadastre_jobs( db: Annotated[Session, Depends(get_db)], limit: int = 30, ) -> list[dict[str, Any]]: """Список последних cadastre harvest jobs.""" rows = ( db.execute( text(""" SELECT job_id, name, job_kind, scope, status, triggered_by, rate_ms, created_at, started_at, finished_at, heartbeat_at, targets_total, targets_done, targets_failed, targets_skipped, requests_count, waf_blocked_count, error, phase_state FROM cadastre_jobs ORDER BY created_at DESC LIMIT :lim """), {"lim": min(limit, 200)}, ) .mappings() .all() ) return [_serialize_job(r) for r in rows] @router.get("/jobs/{job_id}") def get_cadastre_job( job_id: int, db: Annotated[Session, Depends(get_db)], ) -> dict[str, Any]: """Детали одного cadastre harvest job.""" row = ( db.execute( text(""" SELECT job_id, name, job_kind, scope, status, triggered_by, rate_ms, created_at, started_at, finished_at, heartbeat_at, targets_total, targets_done, targets_failed, targets_skipped, requests_count, waf_blocked_count, error, phase_state, raw_targets FROM cadastre_jobs WHERE job_id = :id """), {"id": job_id}, ) .mappings() .first() ) if not row: raise HTTPException(status_code=404, detail=f"cadastre_job {job_id} не найден") result = _serialize_job(row) # Добавить сводку по raw_targets (количество + первые 10 для UI) raw_targets = row["raw_targets"] or {} quarters = raw_targets.get("quarters", []) result["raw_targets_count"] = len(quarters) result["raw_targets_preview"] = quarters[:10] return result @router.post("/jobs/{job_id}/cancel") def cancel_cadastre_job( job_id: int, db: Annotated[Session, Depends(get_db)], ) -> dict[str, Any]: """Отменить job. Воркеры увидят status='cancelled' при следующей итерации и skip.""" result = db.execute( text(""" UPDATE cadastre_jobs SET status = 'cancelled', finished_at = COALESCE(finished_at, NOW()), error = COALESCE(error, 'cancelled by admin') WHERE job_id = :id AND status IN ('queued', 'running', 'paused') RETURNING job_id """), {"id": job_id}, ).first() db.commit() if not result: raise HTTPException( status_code=404, detail=f"cadastre_job {job_id} не найден или уже завершён/отменён", ) logger.info("cancel_cadastre_job: job_id=%d cancelled", job_id) return {"job_id": job_id, "cancelled": True} @router.post("/jobs/{job_id}/resume") def resume_cadastre_job( job_id: int, db: Annotated[Session, Depends(get_db)], ) -> dict[str, Any]: """Возобновить paused/failed job. Re-enqueue enqueue_cadastre_harvest.""" from app.workers.tasks.scrape_cadastre import enqueue_cadastre_harvest result = db.execute( text(""" UPDATE cadastre_jobs SET status = 'queued', error = NULL WHERE job_id = :id AND status IN ('paused', 'failed') RETURNING job_id """), {"id": job_id}, ).first() db.commit() if not result: raise HTTPException( status_code=404, detail=f"cadastre_job {job_id} не найден или не в состоянии paused/failed", ) enqueue_cadastre_harvest.apply_async(args=[job_id], queue="geo") logger.info("resume_cadastre_job: job_id=%d re-enqueued", job_id) return {"job_id": job_id, "resumed": True}