1299 lines
47 KiB
Python
1299 lines
47 KiB
Python
"""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
|
||
]
|
||
|
||
|
||
# ── 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,
|
||
_: AdminTokenAuth,
|
||
) -> 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,
|
||
_: AdminTokenAuth,
|
||
) -> 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",
|
||
}
|