bot-backend
fcd88aa756
Источник без cron (admin UI / CLI / lazy из analyze) — штучные фетчи по активности пользователя. fresh_days=7 давал ложный stale-алерт каждую неделю тишины (расследование 2026-07-04: 60 done-jobs, 0 failed, ledger целый, НСПД доступен — простой 9.5d это норма каденции, не поломка). critical=False — overall и так не трогался, чиним только шум.
1807 lines
79 KiB
Python
1807 lines
79 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?}
|
||
|
||
Auth: gendsgn.ru-wide Caddy basic_auth gate (PR #426). App-level X-Admin-Token
|
||
header removed 2026-05-23 — двойная auth избыточна для pilot.
|
||
|
||
Returns the Celery task id; poll /api/v1/admin/scrape/runs/{run_id} to track DB-side progress.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import UTC, datetime
|
||
from typing import Annotated, Any
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.core.db import get_db
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
def _utcnow() -> datetime:
|
||
"""Timezone-aware UTC now. timestamptz-колонки приходят aware, поэтому naive
|
||
datetime.utcnow() дал бы TypeError при вычитании — всегда aware."""
|
||
return datetime.now(UTC)
|
||
|
||
|
||
class TriggerKnRequest(BaseModel):
|
||
region_code: int = Field(..., ge=1, le=99)
|
||
developers: list[str] | None = Field(default=None, max_length=20)
|
||
fetch_flats: bool = True
|
||
extras: bool = True
|
||
download_photos: bool = False
|
||
# Если True — ПЕРЕД apply_async удалить existing Redis lock. Эмердженси-режим
|
||
# для случая «зомби-worker оставил lock и блокирует новые задачи».
|
||
force: bool = False
|
||
|
||
|
||
@router.post("/kn")
|
||
def trigger_kn_sweep(
|
||
payload: TriggerKnRequest,
|
||
) -> dict[str, Any]:
|
||
# Defer Celery import so the API works without a broker in dev.
|
||
from app.workers.tasks.scrape_kn import (
|
||
force_release_lock,
|
||
scrape_kn_region,
|
||
)
|
||
|
||
lock_released = False
|
||
if payload.force:
|
||
lock_released = force_release_lock(payload.region_code, payload.developers)
|
||
|
||
result = scrape_kn_region.apply_async(
|
||
args=[payload.region_code, payload.developers],
|
||
kwargs={
|
||
"skip_jitter": True,
|
||
"extras": payload.extras,
|
||
"download_photos": payload.download_photos,
|
||
"fetch_flats": payload.fetch_flats,
|
||
},
|
||
)
|
||
return {
|
||
"task_id": result.id,
|
||
"region_code": payload.region_code,
|
||
"developers": payload.developers,
|
||
"queued_at": "now",
|
||
"force": payload.force,
|
||
"lock_was_released": lock_released,
|
||
}
|
||
|
||
|
||
@router.get("/queue")
|
||
def queue_status(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Snapshot of Celery state, optimised for UI poll latency.
|
||
|
||
- active: from kn_scrape_runs WHERE status='running' (instant, richer info
|
||
than celery inspect — includes objects_count/flats_count/requests_count).
|
||
- reserved + workers: celery inspect with short timeout (500 ms), called in
|
||
a thread so a missing worker can't block the request.
|
||
- queue_depth: Redis LLEN on default queue (single round-trip).
|
||
- scheduled: omitted — we don't use ETA tasks.
|
||
|
||
Worst-case latency ≈ 600 ms even if no worker is reachable, vs ~8 s with
|
||
the previous 4×2 s blocking inspect calls.
|
||
"""
|
||
|
||
import concurrent.futures
|
||
import time
|
||
|
||
from app.workers.celery_app import celery_app
|
||
|
||
# 0) Reap zombies by heartbeat. After Resume_Checkpoint refactor the worker
|
||
# writes heartbeat_at every ~30 sec (per 10 objects). 10 min без heartbeat
|
||
# = точно мёртв. Используем COALESCE(heartbeat_at, started_at) чтобы
|
||
# подхватить и legacy-runs (без heartbeat) старше 10 мин.
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE kn_scrape_runs
|
||
SET status = 'zombie',
|
||
finished_at = NOW(),
|
||
error = 'auto-marked: no heartbeat 10+ min'
|
||
WHERE status = 'running'
|
||
AND finished_at IS NULL
|
||
AND COALESCE(heartbeat_at, started_at)
|
||
< NOW() - INTERVAL '10 minutes'
|
||
"""
|
||
)
|
||
)
|
||
db.commit()
|
||
|
||
# 1) Active runs straight from DB — no broker round-trip.
|
||
active_rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT run_id, started_at, region_codes, developer_ids,
|
||
objects_count, flats_count, requests_count
|
||
FROM kn_scrape_runs
|
||
WHERE status = 'running'
|
||
ORDER BY started_at DESC
|
||
LIMIT 20
|
||
"""
|
||
)
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
active: list[dict[str, Any]] = [
|
||
{
|
||
"worker": "db",
|
||
"task_id": f"run-{r['run_id']}",
|
||
"name": "scrape_kn_region",
|
||
"args": [list(r["region_codes"] or []), list(r["developer_ids"] or [])],
|
||
"kwargs": {
|
||
"objects": r["objects_count"],
|
||
"flats": r["flats_count"],
|
||
"requests": r["requests_count"],
|
||
},
|
||
"time_start": (int(r["started_at"].timestamp()) if r["started_at"] else None),
|
||
"eta": None,
|
||
}
|
||
for r in active_rows
|
||
]
|
||
|
||
# 2) Run reserved + ping in parallel with a tight timeout so the endpoint
|
||
# never hangs when the broker is unreachable or worker is gone.
|
||
inspect = celery_app.control.inspect(timeout=0.5)
|
||
|
||
def _flatten(per_worker: dict[str, list[dict[str, Any]]] | None) -> list[dict[str, Any]]:
|
||
if not per_worker:
|
||
return []
|
||
out: list[dict[str, Any]] = []
|
||
for worker, tasks in per_worker.items():
|
||
for t in tasks or []:
|
||
out.append(
|
||
{
|
||
"worker": worker,
|
||
"task_id": t.get("id"),
|
||
"name": t.get("name"),
|
||
"args": t.get("args"),
|
||
"kwargs": t.get("kwargs"),
|
||
"time_start": t.get("time_start"),
|
||
"eta": t.get("eta"),
|
||
}
|
||
)
|
||
return out
|
||
|
||
def _safe(fn):
|
||
# Деградация (return None при недоступном broker) намеренна для UI-poll
|
||
# эндпоинта, но логируем чтобы устойчиво сломанный broker был виден в логах.
|
||
try:
|
||
return fn()
|
||
except Exception:
|
||
logger.warning(
|
||
"queue_status: celery inspect call %s failed",
|
||
getattr(fn, "__name__", fn),
|
||
exc_info=True,
|
||
)
|
||
return None
|
||
|
||
deadline = time.monotonic() + 0.8
|
||
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as ex:
|
||
f_reserved = ex.submit(_safe, inspect.reserved)
|
||
f_ping = ex.submit(_safe, inspect.ping)
|
||
try:
|
||
reserved_raw = f_reserved.result(timeout=max(0.1, deadline - time.monotonic()))
|
||
except concurrent.futures.TimeoutError:
|
||
reserved_raw = None
|
||
try:
|
||
ping_resp = f_ping.result(timeout=max(0.1, deadline - time.monotonic()))
|
||
except concurrent.futures.TimeoutError:
|
||
ping_resp = None
|
||
|
||
reserved = _flatten(reserved_raw)
|
||
workers = list((ping_resp or {}).keys())
|
||
|
||
# 3) Pending in broker queue (not yet picked up by any worker).
|
||
queue_depth: int | None = None
|
||
try:
|
||
with celery_app.connection_or_acquire() as conn:
|
||
with conn.channel() as channel:
|
||
queue_depth = channel.client.llen("celery")
|
||
except Exception:
|
||
# Намеренная деградация для UI-poll; логируем чтобы недоступный broker
|
||
# не был невидим в логах (см. .claude/rules/backend.md).
|
||
logger.warning("queue_status: broker queue_depth probe failed", exc_info=True)
|
||
queue_depth = None
|
||
|
||
return {
|
||
"workers": workers,
|
||
"queue_depth": queue_depth,
|
||
"active": active,
|
||
"reserved": reserved,
|
||
"scheduled": [],
|
||
}
|
||
|
||
|
||
class ReleaseLockRequest(BaseModel):
|
||
region_code: int = Field(..., ge=1, le=99)
|
||
developers: list[str] | None = Field(default=None, max_length=20)
|
||
|
||
|
||
@router.post("/release-lock")
|
||
def release_lock(
|
||
payload: ReleaseLockRequest,
|
||
) -> dict[str, Any]:
|
||
"""Принудительно удалить Redis-lock для (region, devs). Использовать когда
|
||
зомби-worker оставил lock и новые задачи скипаются с reason='lock_held'."""
|
||
from app.workers.tasks.scrape_kn import force_release_lock
|
||
|
||
released = force_release_lock(payload.region_code, payload.developers)
|
||
return {
|
||
"region_code": payload.region_code,
|
||
"developers": payload.developers,
|
||
"released": released,
|
||
}
|
||
|
||
|
||
class RevokeRequest(BaseModel):
|
||
task_id: str = Field(..., min_length=1, max_length=255)
|
||
terminate: bool = False # SIGTERM running task vs just unmark from queue
|
||
|
||
|
||
@router.post("/revoke")
|
||
def revoke_task(
|
||
payload: RevokeRequest,
|
||
) -> dict[str, Any]:
|
||
"""Cancel a queued or running task by id."""
|
||
from app.workers.celery_app import celery_app
|
||
|
||
celery_app.control.revoke(payload.task_id, terminate=payload.terminate, signal="SIGTERM")
|
||
return {"task_id": payload.task_id, "revoked": True, "terminate": payload.terminate}
|
||
|
||
|
||
@router.get("/failures")
|
||
def list_failures(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
run_id: int | None = None,
|
||
limit: int = Query(default=50, ge=1),
|
||
) -> list[dict[str, Any]]:
|
||
"""Per-request failure log for manual browser verification."""
|
||
where = ""
|
||
params: dict[str, Any] = {"lim": min(limit, 200)}
|
||
if run_id is not None:
|
||
where = "WHERE run_id = :rid"
|
||
params["rid"] = run_id
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
f"""
|
||
SELECT failure_id, run_id, obj_id, endpoint, full_url,
|
||
http_status, error, body_preview, occurred_at
|
||
FROM kn_scrape_failures
|
||
{where}
|
||
ORDER BY occurred_at DESC
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
params,
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"failure_id": r["failure_id"],
|
||
"run_id": r["run_id"],
|
||
"obj_id": r["obj_id"],
|
||
"endpoint": r["endpoint"],
|
||
"full_url": r["full_url"],
|
||
"http_status": r["http_status"],
|
||
"error": r["error"],
|
||
"body_preview": r["body_preview"],
|
||
"occurred_at": r["occurred_at"].isoformat() if r["occurred_at"] else None,
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@router.get("/logs")
|
||
def list_logs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
run_id: int | None = None,
|
||
since_id: int | None = None,
|
||
limit: int = Query(default=200, ge=0),
|
||
) -> list[dict[str, Any]]:
|
||
"""Per-run progress events. Use since_id to poll incrementally:
|
||
pass the highest log_id seen → returns only newer rows."""
|
||
where: list[str] = []
|
||
params: dict[str, Any] = {"lim": min(limit, 1000)}
|
||
if run_id is not None:
|
||
where.append("run_id = :rid")
|
||
params["rid"] = run_id
|
||
if since_id is not None:
|
||
where.append("log_id > :sid")
|
||
params["sid"] = since_id
|
||
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
f"""
|
||
SELECT log_id, run_id, ts, level, stage, obj_id, message
|
||
FROM kn_scrape_log
|
||
{where_sql}
|
||
ORDER BY log_id DESC
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
params,
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"log_id": r["log_id"],
|
||
"run_id": r["run_id"],
|
||
"ts": r["ts"].isoformat() if r["ts"] else None,
|
||
"level": r["level"],
|
||
"stage": r["stage"],
|
||
"obj_id": r["obj_id"],
|
||
"message": r["message"],
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
# NSPD-скрапер УДАЛЁН 2026-05-11: вместо Playwright-based nspd_kn.py теперь
|
||
# используется bulk geo-fetcher через rosreestr2coord. См. /admin/scrape/geo
|
||
# и backend/app/workers/tasks/nspd_geo.py + nspd_lite.py (urllib).
|
||
# История прогонов осталась в nspd_scrape_runs/log + v_scrape_runs_unified.
|
||
|
||
|
||
# ── Objective ETL (Антоновский SQLite → PG) ─────────────────────────────────
|
||
|
||
|
||
@router.post("/noise-sync")
|
||
def trigger_noise_sync() -> dict[str, Any]:
|
||
"""Manual trigger для синхронизации OSM шумовых источников ЕКБ из Overpass API.
|
||
|
||
Обычно запускается еженедельно через beat (понед 3:30 МСК).
|
||
Этот endpoint — для ad-hoc запуска (например после первого деплоя или
|
||
после добавления миграции 84_osm_noise_sources_ekb).
|
||
"""
|
||
from app.workers.tasks.noise_sync import sync_osm_noise_sources_ekb
|
||
|
||
result = sync_osm_noise_sources_ekb.apply_async()
|
||
return {"task_id": result.id, "queued_at": "now"}
|
||
|
||
|
||
class HarvestQuarterRequest(BaseModel):
|
||
"""Параметры ручного запуска harvest одного квартала."""
|
||
|
||
quarter_cad: str = Field(
|
||
...,
|
||
min_length=5,
|
||
pattern=r"^\d+:\d+:\d+$",
|
||
description="3-сегментный кадастровый номер квартала, например '66:41:0204016'",
|
||
)
|
||
region_code: int = Field(default=66, ge=1, le=99)
|
||
include_zouit: bool = Field(default=True)
|
||
include_risks: bool = Field(default=False)
|
||
|
||
|
||
@router.post("/nspd/harvest-quarter")
|
||
def trigger_harvest_quarter(
|
||
payload: HarvestQuarterRequest,
|
||
) -> dict[str, Any]:
|
||
"""Manual trigger одного quarter harvest (для testing / конкретного квартала).
|
||
|
||
Запускает harvest_quarter Celery task и возвращает task_id.
|
||
Для получения результата — poll через Celery inspect или нождите строку
|
||
в nspd_quarter_dumps WHERE quarter_cad = payload.quarter_cad.
|
||
"""
|
||
from app.workers.tasks.nspd_sync import harvest_quarter
|
||
|
||
task = harvest_quarter.apply_async(
|
||
args=[payload.quarter_cad, payload.region_code],
|
||
kwargs={
|
||
"include_zouit": payload.include_zouit,
|
||
"include_risks": payload.include_risks,
|
||
},
|
||
)
|
||
return {
|
||
"task_id": task.id,
|
||
"quarter_cad": payload.quarter_cad,
|
||
"region_code": payload.region_code,
|
||
"include_zouit": payload.include_zouit,
|
||
"include_risks": payload.include_risks,
|
||
}
|
||
|
||
|
||
@router.post("/pzz-sync")
|
||
def trigger_pzz_sync() -> dict[str, Any]:
|
||
"""Manual trigger для импорта ПЗЗ территориальных зон ЕКБ из Росреестр PKK6.
|
||
|
||
Запускать после деплоя миграции 85_pzz_zones_ekb.sql и при необходимости
|
||
обновить данные о зонировании (обычно раз в несколько месяцев).
|
||
"""
|
||
from app.workers.tasks.pzz_sync import sync_pzz_zones_ekb
|
||
|
||
result = sync_pzz_zones_ekb.apply_async()
|
||
return {"task_id": result.id, "queued_at": "now"}
|
||
|
||
|
||
class TriggerZoneRegulationsBackfillRequest(BaseModel):
|
||
"""Параметры ручного backfill ПЗЗ-регламента терзон ЕКБ (эпик #1881)."""
|
||
|
||
limit: int | None = Field(
|
||
default=None,
|
||
ge=1,
|
||
le=1000,
|
||
description="Максимум уникальных незакэшированных зон к фетчу (None = все). Smoke-тест.",
|
||
)
|
||
rate_delay_s: float = Field(
|
||
default=1.0,
|
||
ge=0.0,
|
||
le=10.0,
|
||
description="Пауза (сек) между live-fetch'ами зон — вежливость к геопорталу.",
|
||
)
|
||
|
||
|
||
@router.post("/zone-regulations/backfill")
|
||
def trigger_zone_regulations_backfill(
|
||
payload: TriggerZoneRegulationsBackfillRequest,
|
||
) -> dict[str, Any]:
|
||
"""Backfill кэша ПЗЗ-регламента ВСЕХ терзон ЕКБ → zone_regulation_cache (эпик #1881).
|
||
|
||
Чтобы financial_estimate фаерил для большинства участков (max_far из кэша, без live-urbanCard
|
||
в hot-пути). Идемпотентно: уже закэшированные зоны пропускаются, повторный запуск дёшев.
|
||
|
||
estimate_minutes — груба: ~120 уникальных терзон ЕКБ × (rate_delay + ~2с live-urbanCard).
|
||
"""
|
||
from app.workers.tasks.backfill_zone_regulations import backfill_zone_regulations
|
||
|
||
result = backfill_zone_regulations.apply_async(
|
||
kwargs={
|
||
"limit": payload.limit,
|
||
"rate_delay_s": payload.rate_delay_s,
|
||
},
|
||
)
|
||
# ~120 уникальных терзон ЕКБ; на зону ≈ rate_delay + ~2с urbanCard round-trip.
|
||
n_zones = payload.limit if payload.limit is not None else 120
|
||
estimate_minutes = round(n_zones * (payload.rate_delay_s + 2.0) / 60.0, 1)
|
||
return {
|
||
"task_id": result.id,
|
||
"limit": payload.limit,
|
||
"rate_delay_s": payload.rate_delay_s,
|
||
"estimate_minutes": estimate_minutes,
|
||
"queued_at": "now",
|
||
}
|
||
|
||
|
||
@router.post("/poi-sync")
|
||
def trigger_poi_sync() -> dict[str, Any]:
|
||
"""Manual trigger для синхронизации OSM POI ЕКБ из Overpass API.
|
||
|
||
Обычно запускается еженедельно через beat (понед 3:00 МСК).
|
||
Этот endpoint — для ad-hoc запуска (например после первого деплоя
|
||
или при создании нового тестового участка).
|
||
"""
|
||
from app.workers.tasks.poi_sync import sync_osm_poi_ekb
|
||
|
||
result = sync_osm_poi_ekb.apply_async()
|
||
return {"task_id": result.id, "queued_at": "now"}
|
||
|
||
|
||
class TriggerObjectiveEtlRequest(BaseModel):
|
||
sqlite_path: str | None = Field(
|
||
default=None,
|
||
description="Override settings.objective_anton_sqlite_path (abs file path)",
|
||
)
|
||
|
||
|
||
@router.post("/objective")
|
||
def trigger_objective_etl(
|
||
payload: TriggerObjectiveEtlRequest,
|
||
) -> dict[str, Any]:
|
||
"""Запустить ETL Антоновского SQLite → наша PG.
|
||
|
||
SQLite-файл должен быть доступен внутри Celery-worker контейнера:
|
||
- на проде: смонтирован bind-mount-ом из /opt/gendesign/site-finder/
|
||
- локально: путь передать через payload.sqlite_path
|
||
"""
|
||
from app.workers.tasks.objective_etl import import_anton_objective
|
||
|
||
result = import_anton_objective.apply_async(
|
||
kwargs={
|
||
"triggered_by": "manual",
|
||
"sqlite_path": payload.sqlite_path,
|
||
},
|
||
)
|
||
return {
|
||
"task_id": result.id,
|
||
"sqlite_path": payload.sqlite_path or settings.objective_anton_sqlite_path,
|
||
}
|
||
|
||
|
||
@router.get("/objective/runs")
|
||
def list_objective_runs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
limit: int = Query(default=20, ge=0),
|
||
) -> list[dict[str, Any]]:
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT run_id, started_at, finished_at, heartbeat_at,
|
||
group_name, reports_ok, reports_failed,
|
||
rows_corpus_room, rows_lots, rows_history,
|
||
requests_count, status, error, triggered_by
|
||
FROM objective_scrape_runs
|
||
ORDER BY started_at DESC
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
{"lim": min(limit, 100)},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"run_id": r["run_id"],
|
||
"started_at": r["started_at"].isoformat() if r["started_at"] else None,
|
||
"finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
|
||
"heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None,
|
||
"group_name": r["group_name"],
|
||
"reports_ok": r["reports_ok"],
|
||
"reports_failed": r["reports_failed"],
|
||
"rows_corpus_room": r["rows_corpus_room"],
|
||
"rows_lots": r["rows_lots"],
|
||
# Поле в Schema 68 v2 называется rows_history — мы переиспользуем его
|
||
# под количество загруженных mapping'ов из jk_objective_match.
|
||
"rows_mappings": r["rows_history"],
|
||
"requests_count": r["requests_count"],
|
||
"status": r["status"],
|
||
"error": r["error"],
|
||
"triggered_by": r["triggered_by"],
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
class TriggerObjectiveSyncRequest(BaseModel):
|
||
"""Ручной триггер нашего sync_all_groups (платный API Объектива)."""
|
||
|
||
groups: list[str] | None = Field(
|
||
default=None,
|
||
description="Override settings.objective_sync_groups (csv). "
|
||
"Если None — берёт дефолт из settings.",
|
||
max_length=20,
|
||
)
|
||
|
||
|
||
@router.post("/objective/sync-our")
|
||
def trigger_objective_sync_our(
|
||
payload: TriggerObjectiveSyncRequest,
|
||
) -> dict[str, Any]:
|
||
"""Запустить НАШ sync_all_groups (тратит limits Объектива).
|
||
|
||
В отличие от ETL (POST /objective) — это вызовы api.objctv.ru напрямую,
|
||
с парсингом payload через app.workers.tasks.scrape_objective.
|
||
Полезно когда нужно расширить покрытие за пределов Антоновского ЕКБ
|
||
(Свердл.обл целиком + Челябинск + Тюмень + Пермь).
|
||
"""
|
||
if not settings.objective_api_key:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail="OBJECTIVE_API_KEY не задан (settings.objective_api_key пуст)",
|
||
)
|
||
from app.workers.tasks.scrape_objective import sync_all_groups
|
||
|
||
groups_eff = payload.groups or [
|
||
g.strip() for g in settings.objective_sync_groups.split(",") if g.strip()
|
||
]
|
||
result = sync_all_groups.apply_async(
|
||
kwargs={
|
||
"groups": payload.groups, # None → task возьмёт дефолт
|
||
"triggered_by": "manual",
|
||
},
|
||
)
|
||
return {
|
||
"task_id": result.id,
|
||
"groups_planned": groups_eff,
|
||
"estimate_minutes": len(groups_eff)
|
||
* 3, # ~3 мин на группу (corp_sum + lots_pf + 30s pause)
|
||
}
|
||
|
||
|
||
class ObjectiveConfigUpdateRequest(BaseModel):
|
||
"""PATCH-style — обновляет только переданные поля."""
|
||
|
||
cron_schedule: str | None = None
|
||
groups_csv: str | None = None
|
||
use_ddu: bool | None = None
|
||
use_dkp: bool | None = None
|
||
period_months_back: int | None = Field(default=None, ge=1, le=24)
|
||
inter_group_delay_s: int | None = Field(default=None, ge=0, le=300)
|
||
rate_ms: int | None = Field(default=None, ge=0, le=10000)
|
||
retries: int | None = Field(default=None, ge=0, le=5)
|
||
|
||
|
||
@router.get("/objective/config")
|
||
def get_objective_config(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Текущая динамическая конфигурация Objective sync (single row)."""
|
||
from app.services.objective_sync_config import get_config
|
||
|
||
return get_config(db).to_dict()
|
||
|
||
|
||
@router.put("/objective/config")
|
||
def update_objective_config(
|
||
payload: ObjectiveConfigUpdateRequest,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""PATCH-style update. После изменения cron_schedule нужен restart beat,
|
||
остальные поля применяются на следующем sync-вызове автоматически."""
|
||
from app.services.objective_sync_config import update_config
|
||
|
||
cfg = update_config(
|
||
db,
|
||
cron_schedule=payload.cron_schedule,
|
||
groups_csv=payload.groups_csv,
|
||
use_ddu=payload.use_ddu,
|
||
use_dkp=payload.use_dkp,
|
||
period_months_back=payload.period_months_back,
|
||
inter_group_delay_s=payload.inter_group_delay_s,
|
||
rate_ms=payload.rate_ms,
|
||
retries=payload.retries,
|
||
updated_by="admin-ui",
|
||
)
|
||
return {
|
||
**cfg.to_dict(),
|
||
"_note": (
|
||
"cron_schedule applied только после restart beat-контейнера; "
|
||
"остальные параметры подхватятся при следующем запуске sync_all_groups"
|
||
),
|
||
}
|
||
|
||
|
||
@router.get("/objective/coverage")
|
||
def objective_coverage(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Что у нас в БД сейчас + что лежит в SQLite Антона (size, last modified).
|
||
|
||
Помогает понять стоит ли запускать ETL: если SQLite старше последнего
|
||
успешного run'а — данные не изменятся.
|
||
"""
|
||
from app.services.objective_etl import get_sqlite_info
|
||
|
||
# #1964: `lots` = СЫРОЙ COUNT(*) — намеренно. Этот счётчик сравнивается с
|
||
# SQLite-источником Антона (ETL-fidelity: «свежее ли SQLite, чем БД»), а SQLite
|
||
# хранит сырые строки → дедуп-view сломал бы сравнение. Рядом отдаём
|
||
# `lots_physflat` (v_objective_lots_latest — по физлоту) для прозрачности
|
||
# инфляции (~2.91×: сырых ~1.75M vs физлотов ~603k).
|
||
pg_row = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT
|
||
(SELECT COUNT(*) FROM objective_lots) AS lots,
|
||
(SELECT COUNT(*) FROM v_objective_lots_latest) AS lots_physflat,
|
||
(SELECT COUNT(*) FROM objective_corpus_room_month) AS crm,
|
||
(SELECT COUNT(*) FROM objective_complex_mapping) AS mappings,
|
||
(SELECT MAX(snapshot_date) FROM objective_lots) AS last_lot_snapshot,
|
||
(SELECT MAX(finished_at) FROM objective_scrape_runs
|
||
WHERE status = 'done'
|
||
AND triggered_by IN ('manual','beat','api')) AS last_run_at
|
||
"""
|
||
)
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
sqlite_info = get_sqlite_info(settings.objective_anton_sqlite_path)
|
||
return {
|
||
"pg": {
|
||
"lots": int(pg_row["lots"] or 0),
|
||
"lots_physflat": int(pg_row["lots_physflat"] or 0),
|
||
"corp_room_month": int(pg_row["crm"] or 0),
|
||
"mappings": int(pg_row["mappings"] or 0),
|
||
"last_lot_snapshot": pg_row["last_lot_snapshot"].isoformat()
|
||
if pg_row["last_lot_snapshot"]
|
||
else None,
|
||
"last_run_at": pg_row["last_run_at"].isoformat() if pg_row["last_run_at"] else None,
|
||
},
|
||
"sqlite": sqlite_info,
|
||
}
|
||
|
||
|
||
# ── NSPD geo bulk-jobs (resume-friendly) ────────────────────────────────────
|
||
|
||
|
||
class EnqueueGeoJobRequest(BaseModel):
|
||
"""Создать новый bulk geo-job."""
|
||
|
||
name: str | None = Field(default=None, max_length=200)
|
||
job_kind: str = Field(..., pattern="^(quarters|parcels|buildings|mixed)$")
|
||
source_kind: str = Field(
|
||
default="manual_list", pattern="^(manual_list|rosreestr_pending|region_bbox)$"
|
||
)
|
||
cad_nums: list[str] = Field(
|
||
default_factory=list, max_length=50000, description="Список cad-номеров для обработки"
|
||
)
|
||
thematic_id: int = Field(default=2, ge=1, le=15, description="1=parcel, 2=quarter, 5=building")
|
||
region_codes: list[int] | None = Field(
|
||
default=None, max_length=20, description="Для source=rosreestr_pending"
|
||
)
|
||
rate_ms: int = Field(default=600, ge=100, le=10000)
|
||
|
||
|
||
@router.post("/geo/jobs")
|
||
def enqueue_geo_job(
|
||
payload: EnqueueGeoJobRequest,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Создать bulk geo-job и поставить в очередь.
|
||
|
||
source_kind:
|
||
- manual_list: cad_nums передан явно
|
||
- rosreestr_pending: автоматически набрать из rosreestr_deals.quarter_cad_number
|
||
которых ещё нет в cad_quarters_geom (для region_codes если задано)
|
||
- region_bbox: TODO (для будущего spatial-bulk)
|
||
"""
|
||
from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper
|
||
from app.workers.tasks.nspd_geo import process_nspd_geo_job
|
||
|
||
cad_with_thematic: list[tuple[str, int]] = []
|
||
if payload.source_kind == "manual_list":
|
||
cad_with_thematic = [
|
||
(c.strip(), payload.thematic_id) for c in payload.cad_nums if c.strip()
|
||
]
|
||
elif payload.source_kind == "rosreestr_pending":
|
||
regions = payload.region_codes or [66]
|
||
rows = db.execute(
|
||
text(
|
||
"""
|
||
SELECT DISTINCT quarter_cad_number AS cad
|
||
FROM rosreestr_deals
|
||
WHERE region_code = ANY(:rc)
|
||
AND doc_type = 'ДДУ'
|
||
AND realestate_type_code = '002001003000'
|
||
AND quarter_cad_number IS NOT NULL
|
||
AND quarter_cad_number NOT LIKE :bp
|
||
AND quarter_cad_number NOT LIKE :bs
|
||
AND NOT EXISTS (SELECT 1 FROM cad_quarters_geom g
|
||
WHERE g.cad_number = rosreestr_deals.quarter_cad_number)
|
||
LIMIT 50000
|
||
"""
|
||
),
|
||
{"rc": regions, "bp": "00:00:%", "bs": "%:0000000"},
|
||
).all()
|
||
cad_with_thematic = [(r[0], 2) for r in rows]
|
||
else:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"source_kind={payload.source_kind} not implemented yet",
|
||
)
|
||
|
||
if not cad_with_thematic:
|
||
raise HTTPException(status_code=400, detail="no targets to process")
|
||
|
||
job_id = enqueue_helper(
|
||
name=payload.name,
|
||
job_kind=payload.job_kind,
|
||
source_kind=payload.source_kind,
|
||
source_params={"region_codes": payload.region_codes, "thematic_id": payload.thematic_id},
|
||
cad_nums_with_thematic=cad_with_thematic,
|
||
rate_ms=payload.rate_ms,
|
||
triggered_by="manual",
|
||
)
|
||
from app.services.job_settings import get_setting_value
|
||
|
||
geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
|
||
process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
|
||
return {
|
||
"job_id": job_id,
|
||
"targets_total": len(cad_with_thematic),
|
||
"estimate_minutes": round(len(cad_with_thematic) * payload.rate_ms / 1000 / 60, 1),
|
||
}
|
||
|
||
|
||
class BulkGeoEnqueueRequest(BaseModel):
|
||
"""Параметры для параллельного backfill geo по Свердловской обл."""
|
||
|
||
parallelism: int = Field(default=5, ge=1, le=10)
|
||
thematic_ids: list[int] = Field(
|
||
default=[2],
|
||
description="Список thematic_id: 1=parcel, 2=quarter, 5=building. Можно несколько.",
|
||
)
|
||
region_codes: list[int] = Field(default=[66], description="Список region кодов")
|
||
only_ddu: bool = Field(
|
||
default=False,
|
||
description="True = только ДДУ (002001003000); False = все cad-кварталы",
|
||
)
|
||
source: str = Field(
|
||
default="rosreestr_pending",
|
||
pattern="^(rosreestr_pending|all_in_region)$",
|
||
description=(
|
||
"rosreestr_pending — cad-номера из rosreestr_deals которых нет в geo-таблице; "
|
||
"all_in_region — UNION всех cad из rosreestr_deals + cad_buildings + complexes"
|
||
),
|
||
)
|
||
rate_ms: int = Field(default=600, ge=100, le=10000)
|
||
|
||
|
||
# Маппинг thematic_id → таблица проверки существования + колонка + job_kind-метка
|
||
_THEMATIC_META: dict[int, dict[str, str]] = {
|
||
1: {"exists_table": "cad_parcels_geom", "exists_col": "cad_num", "label": "parcels"},
|
||
2: {"exists_table": "cad_quarters_geom", "exists_col": "cad_number", "label": "quarters"},
|
||
5: {"exists_table": "cad_buildings", "exists_col": "cad_num", "label": "buildings"},
|
||
}
|
||
|
||
|
||
def _collect_pending_cad(
|
||
db: Session,
|
||
thematic_id: int,
|
||
region_codes: list[int],
|
||
only_ddu: bool,
|
||
) -> list[str]:
|
||
"""Собрать cad-номера из rosreestr_deals которых нет в соответствующей geo-таблице."""
|
||
meta = _THEMATIC_META.get(thematic_id)
|
||
if meta is None:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"thematic_id={thematic_id} не поддерживается (допустимы: 1, 2, 5)",
|
||
)
|
||
ddu_filter = (
|
||
"AND doc_type = 'ДДУ' AND realestate_type_code = '002001003000'" if only_ddu else ""
|
||
)
|
||
exists_table = meta["exists_table"]
|
||
exists_col = meta["exists_col"]
|
||
rows = db.execute(
|
||
text(
|
||
f"""
|
||
SELECT DISTINCT quarter_cad_number AS cad
|
||
FROM rosreestr_deals
|
||
WHERE region_code = ANY(:rc)
|
||
{ddu_filter}
|
||
AND quarter_cad_number IS NOT NULL
|
||
AND quarter_cad_number NOT LIKE :bp
|
||
AND quarter_cad_number NOT LIKE :bs
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM {exists_table} g
|
||
WHERE g.{exists_col} = rosreestr_deals.quarter_cad_number
|
||
)
|
||
LIMIT 50000
|
||
"""
|
||
),
|
||
{"rc": region_codes, "bp": "00:00:%", "bs": "%:0000000"},
|
||
).all()
|
||
return [r[0] for r in rows]
|
||
|
||
|
||
def _collect_all_in_region(
|
||
db: Session,
|
||
region_codes: list[int],
|
||
) -> list[str]:
|
||
"""UNION всех cad-номеров из rosreestr_deals + cad_buildings + complexes.cad_quarter
|
||
с фильтром по region prefix (66xx для кодов региона 66)."""
|
||
rows = db.execute(
|
||
text("""
|
||
SELECT DISTINCT cad FROM (
|
||
SELECT quarter_cad_number AS cad
|
||
FROM rosreestr_deals
|
||
WHERE quarter_cad_number IS NOT NULL
|
||
AND region_code = ANY(:rc)
|
||
UNION
|
||
SELECT cad_num AS cad
|
||
FROM cad_buildings
|
||
WHERE cad_num IS NOT NULL
|
||
AND cad_num ~ :region_re
|
||
UNION
|
||
SELECT cad_quarter AS cad
|
||
FROM complexes
|
||
WHERE cad_quarter IS NOT NULL
|
||
AND cad_quarter ~ :region_re
|
||
) sub
|
||
WHERE cad NOT LIKE :bp
|
||
AND cad NOT LIKE :bs
|
||
LIMIT 100000
|
||
"""),
|
||
{
|
||
"rc": region_codes,
|
||
"region_re": "^(" + "|".join(str(rc) for rc in region_codes) + "):",
|
||
"bp": "00:00:%",
|
||
"bs": "%:0000000",
|
||
},
|
||
).all()
|
||
return [r[0] for r in rows]
|
||
|
||
|
||
@router.post("/geo/bulk")
|
||
def bulk_enqueue_geo(
|
||
payload: BulkGeoEnqueueRequest,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Разбить pending cad-номера на N чанков и запустить N×len(thematic_ids) geo-jobs.
|
||
|
||
source=rosreestr_pending: для каждого thematic_id выбирает cad из rosreestr_deals
|
||
которых нет в соответствующей geo-таблице.
|
||
source=all_in_region: UNION из rosreestr_deals + cad_buildings + complexes — один
|
||
набор для всех thematic_ids (каждый thematic_id обрабатывает весь список).
|
||
"""
|
||
from app.services.job_settings import get_setting_value
|
||
from app.workers.tasks.nspd_geo import enqueue_geo_job as enqueue_helper
|
||
from app.workers.tasks.nspd_geo import process_nspd_geo_job
|
||
|
||
geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
|
||
|
||
# Валидируем ВСЕ thematic_ids ДО любых сайд-эффектов (создание jobs / apply_async),
|
||
# иначе невалидный id в середине списка приводит к partial execution: для предыдущих
|
||
# валидных id строки в nspd_geo_jobs уже созданы и задачи улетели в очередь geo,
|
||
# а клиент получает 400 без идемпотентного отката (#1562).
|
||
invalid_ids = [tid for tid in payload.thematic_ids if tid not in _THEMATIC_META]
|
||
if invalid_ids:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"thematic_id={invalid_ids} не поддерживается (допустимы: 1, 2, 5)",
|
||
)
|
||
|
||
jobs_summary: list[dict[str, Any]] = []
|
||
|
||
for thematic_id in payload.thematic_ids:
|
||
meta = _THEMATIC_META[thematic_id]
|
||
|
||
# 1) Собрать cad-номера
|
||
if payload.source == "rosreestr_pending":
|
||
all_cad = _collect_pending_cad(db, thematic_id, payload.region_codes, payload.only_ddu)
|
||
else: # all_in_region
|
||
all_cad = _collect_all_in_region(db, payload.region_codes)
|
||
|
||
if not all_cad:
|
||
jobs_summary.append(
|
||
{
|
||
"thematic_id": thematic_id,
|
||
"job_ids": [],
|
||
"targets_total": 0,
|
||
"parallelism": 0,
|
||
"note": "нет cad-номеров для backfill",
|
||
}
|
||
)
|
||
continue
|
||
|
||
# 2) Разбить на чанки
|
||
pending_total = len(all_cad)
|
||
n_jobs = min(payload.parallelism, pending_total)
|
||
chunk_size, remainder = divmod(pending_total, n_jobs)
|
||
chunks: list[list[str]] = []
|
||
start = 0
|
||
for i in range(n_jobs):
|
||
end = start + chunk_size + (1 if i < remainder else 0)
|
||
chunks.append(all_cad[start:end])
|
||
start = end
|
||
|
||
# 3) Создать jobs
|
||
job_ids: list[int] = []
|
||
label = meta["label"]
|
||
for idx, chunk in enumerate(chunks):
|
||
cad_with_thematic = [(c, thematic_id) for c in chunk]
|
||
job_id = enqueue_helper(
|
||
name=f"bulk_{label}_t{thematic_id}_{idx + 1}/{n_jobs}",
|
||
job_kind=label,
|
||
source_kind=f"{payload.source}_chunk",
|
||
source_params={
|
||
"region_codes": payload.region_codes,
|
||
"thematic_id": thematic_id,
|
||
"source": payload.source,
|
||
},
|
||
cad_nums_with_thematic=cad_with_thematic,
|
||
rate_ms=payload.rate_ms,
|
||
triggered_by="bulk_admin",
|
||
)
|
||
process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
|
||
job_ids.append(job_id)
|
||
|
||
jobs_summary.append(
|
||
{
|
||
"thematic_id": thematic_id,
|
||
"job_ids": job_ids,
|
||
"targets_total": pending_total,
|
||
"parallelism": n_jobs,
|
||
}
|
||
)
|
||
|
||
if not any(j["targets_total"] > 0 for j in jobs_summary):
|
||
raise HTTPException(status_code=400, detail="Нет cad-номеров для backfill (все thematic)")
|
||
|
||
return {"jobs": jobs_summary}
|
||
|
||
|
||
@router.get("/geo/jobs")
|
||
def list_geo_jobs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
limit: int = Query(default=30, ge=0),
|
||
) -> list[dict[str, Any]]:
|
||
"""Список последних geo-jobs (для UI dashboard)."""
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT job_id, name, job_kind, source_kind, status, triggered_by,
|
||
started_at, finished_at, heartbeat_at,
|
||
targets_total, targets_done, targets_failed, targets_skipped,
|
||
requests_count, waf_blocked_count, error,
|
||
rate_ms, created_at
|
||
FROM nspd_geo_jobs
|
||
ORDER BY created_at DESC
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
{"lim": min(limit, 100)},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"job_id": r["job_id"],
|
||
"name": r["name"],
|
||
"job_kind": r["job_kind"],
|
||
"source_kind": r["source_kind"],
|
||
"status": r["status"],
|
||
"triggered_by": r["triggered_by"],
|
||
"started_at": r["started_at"].isoformat() if r["started_at"] else None,
|
||
"finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
|
||
"heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None,
|
||
"targets_total": r["targets_total"],
|
||
"targets_done": r["targets_done"],
|
||
"targets_failed": r["targets_failed"],
|
||
"targets_skipped": r["targets_skipped"],
|
||
"requests_count": r["requests_count"],
|
||
"waf_blocked_count": r["waf_blocked_count"],
|
||
"error": r["error"],
|
||
"rate_ms": r["rate_ms"],
|
||
"created_at": r["created_at"].isoformat() if r["created_at"] else None,
|
||
"progress_pct": round(
|
||
100.0 * (r["targets_done"] or 0) / max(r["targets_total"] or 1, 1), 1
|
||
),
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@router.post("/geo/jobs/{job_id}/cancel")
|
||
def cancel_geo_job(
|
||
job_id: int,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Пометить job как cancelled. Worker увидит при следующей итерации."""
|
||
db.execute(
|
||
text(
|
||
"""
|
||
UPDATE nspd_geo_jobs SET status = 'cancelled', finished_at = NOW(),
|
||
error = COALESCE(error, 'cancelled by admin')
|
||
WHERE job_id = :id AND status IN ('queued','running','paused')
|
||
"""
|
||
),
|
||
{"id": job_id},
|
||
)
|
||
db.commit()
|
||
return {"job_id": job_id, "cancelled": True}
|
||
|
||
|
||
@router.post("/geo/jobs/{job_id}/resume")
|
||
def resume_geo_job(
|
||
job_id: int,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Re-enqueue paused/failed job. Resume idempotent через pending targets."""
|
||
from app.services.job_settings import get_setting_value
|
||
from app.workers.tasks.nspd_geo import process_nspd_geo_job
|
||
|
||
db.execute(
|
||
text("UPDATE nspd_geo_jobs SET status='queued', error=NULL WHERE job_id=:id"),
|
||
{"id": job_id},
|
||
)
|
||
db.commit()
|
||
geo_queue = get_setting_value("nspd_geo", "queue_name", "geo")
|
||
process_nspd_geo_job.apply_async(args=[job_id], queue=geo_queue)
|
||
return {"job_id": job_id, "resumed": True}
|
||
|
||
|
||
# ── Newbuilding cross-load ETL (#976) ────────────────────────────────────────
|
||
|
||
|
||
@router.post("/newbuilding-crossload")
|
||
def trigger_newbuilding_crossload() -> dict[str, Any]:
|
||
"""Ручной запуск cross-load ETL tradein.houses → newbuilding_listings (#976).
|
||
|
||
Обычно запускается ночью через beat (03:30 МСК).
|
||
Если TRADEIN_DATABASE_URL не задан в settings — возвращает 503.
|
||
"""
|
||
if not settings.tradein_database_url:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail=(
|
||
"TRADEIN_DATABASE_URL не задан — cross-load ETL отключён. "
|
||
"Добавь переменную в backend/.env.runtime и перезапусти worker."
|
||
),
|
||
)
|
||
from app.workers.tasks.etl_newbuilding_crossload import etl_newbuilding_crossload
|
||
|
||
result = etl_newbuilding_crossload.apply_async()
|
||
return {"task_id": result.id, "queued_at": "now"}
|
||
|
||
|
||
# ── Unified scrape dashboard (поверх v_scrape_runs_unified / v_scrape_log_unified) ──
|
||
|
||
|
||
@router.get("/all/runs")
|
||
def list_all_runs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
scraper_type: str | None = None,
|
||
limit: int = Query(default=30, ge=0),
|
||
) -> list[dict[str, Any]]:
|
||
"""Унифицированный список прогонов (kn + nspd + objective).
|
||
|
||
Поверх v_scrape_runs_unified. Если scraper_type не задан — все 3.
|
||
Используется главной dashboard /admin/scrape/all.
|
||
"""
|
||
where = "" if not scraper_type else "WHERE scraper_type = :st"
|
||
params: dict[str, Any] = {"lim": min(limit, 200)}
|
||
if scraper_type:
|
||
params["st"] = scraper_type
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
f"""
|
||
SELECT scraper_type, run_id, started_at, finished_at, heartbeat_at,
|
||
status, error, triggered_by, scope, requests_count,
|
||
items_ok, items_failed, sub_items_ok,
|
||
progress_idx, progress_total, extra
|
||
FROM v_scrape_runs_unified
|
||
{where}
|
||
ORDER BY started_at DESC NULLS LAST
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
params,
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"scraper_type": r["scraper_type"],
|
||
"run_id": r["run_id"],
|
||
"started_at": r["started_at"].isoformat() if r["started_at"] else None,
|
||
"finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
|
||
"heartbeat_at": r["heartbeat_at"].isoformat() if r["heartbeat_at"] else None,
|
||
"status": r["status"],
|
||
"error": r["error"],
|
||
"triggered_by": r["triggered_by"],
|
||
"scope": r["scope"],
|
||
"requests_count": r["requests_count"],
|
||
"items_ok": r["items_ok"],
|
||
"items_failed": r["items_failed"],
|
||
"sub_items_ok": r["sub_items_ok"],
|
||
"progress_idx": r["progress_idx"],
|
||
"progress_total": r["progress_total"],
|
||
"extra": r["extra"],
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@router.get("/all/logs")
|
||
def list_all_logs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
scraper_type: str | None = None,
|
||
run_id: int | None = None,
|
||
limit: int = Query(default=200, ge=0),
|
||
) -> list[dict[str, Any]]:
|
||
"""Унифицированный список логов (kn + nspd). Objective пока не пишет log."""
|
||
where: list[str] = []
|
||
params: dict[str, Any] = {"lim": min(limit, 1000)}
|
||
if scraper_type:
|
||
where.append("scraper_type = :st")
|
||
params["st"] = scraper_type
|
||
if run_id is not None:
|
||
where.append("run_id = :rid")
|
||
params["rid"] = run_id
|
||
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
f"""
|
||
SELECT scraper_type, log_id, run_id, ts, level, stage, entity_id, message
|
||
FROM v_scrape_log_unified
|
||
{where_sql}
|
||
ORDER BY log_id DESC
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
params,
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"scraper_type": r["scraper_type"],
|
||
"log_id": r["log_id"],
|
||
"run_id": r["run_id"],
|
||
"ts": r["ts"].isoformat() if r["ts"] else None,
|
||
"level": r["level"],
|
||
"stage": r["stage"],
|
||
"entity_id": r["entity_id"],
|
||
"message": r["message"],
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
# ── Data-freshness monitoring (#73 B5-3) ─────────────────────────────────────
|
||
#
|
||
# GET /api/v1/admin/scrape/freshness — для каждого источника данных:
|
||
# last_success_at / last_attempt_at / last_status / objects_updated_24h|7d /
|
||
# age_days / freshness (fresh|stale|critical) / status (ok|stale|failed).
|
||
#
|
||
# Источник истины — run-таблицы (НЕ контент-таблицы): на проде проверено, что
|
||
# идемпотентный re-scrape НЕ бампит complexes.updated_at (newest 2026-05-11 при
|
||
# kn-run 3 дня назад), поэтому "objects_updated" считаем по счётчикам прогонов
|
||
# (objects_count / quarters_ok / rows_lots / targets_done) за done-прогоны в окне.
|
||
#
|
||
# Пороги (fresh/stale/critical в днях) выведены из beat-расписания каждого job'а
|
||
# (app/workers/beat_schedule.py). Sentry-алёрты по критичным job'ам шлёт отдельный
|
||
# beat-task scrape_freshness_check (tasks.scrape_freshness_check) — endpoint только
|
||
# читает, не алертит, чтобы UI-поллинг не спамил issue.
|
||
|
||
|
||
class FreshnessSource(BaseModel):
|
||
"""Описание одного источника данных в реестре freshness-мониторинга."""
|
||
|
||
source: str # стабильный ключ (kn / nspd / objective / nspd_geo / cadastre)
|
||
label: str # человекочитаемое имя для UI
|
||
table: str # run-таблица ЛИБО data-таблица (если задан timestamp_col)
|
||
# Колонка-счётчик "сколько объектов обновил done-прогон". Агрегируется как
|
||
# COALESCE(SUM(<work_col>) FILTER (...), 0) — FILTER применяется к SUM, не к COALESCE.
|
||
# В timestamp-режиме (timestamp_col задан) work_col игнорируется.
|
||
work_col: str
|
||
started_col: str = "started_at"
|
||
# Колонка-fallback для last_attempt когда started_at NULL (queued-but-not-run).
|
||
attempt_fallback_col: str | None = None
|
||
# Data-table режим: если задан, источник — это ПЛОСКАЯ data-таблица (не run-ledger,
|
||
# без status/started/finished). Свежесть = MAX(timestamp_col); upd_24h/_7d = COUNT(*)
|
||
# по строкам в окне. work_col/started_col/attempt_fallback_col при этом игнорируются.
|
||
timestamp_col: str | None = None
|
||
fresh_days: float = 7.0 # < fresh_days → fresh
|
||
stale_days: float = 30.0 # < stale_days → stale, иначе critical
|
||
critical: bool = False # критичный источник (для Sentry-алёртов в beat-таске)
|
||
# True → источник, который прогнался успешно И недавно (был бы fresh по возрасту),
|
||
# НО SUM(work_col) по done-прогонам цикла (окно fresh_days) < min_output_rows,
|
||
# ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-почти-ничего-не-извлёк"
|
||
# (напр. WAF-break / тихий частичный extract: status=done, но flats_count суммарно
|
||
# 9 при здоровых ~376k). По умолчанию OFF — включаем только там, где малый выход =
|
||
# поломка (не legitimate-low, как nspd/nspd_geo/cadastre). Применимо только к
|
||
# run-ledger (не data-table) режиму.
|
||
alert_on_zero_output: bool = False
|
||
# Пол (floor) для recent_output = SUM(work_col) по done-прогонам ЦИКЛА в окне
|
||
# fresh_days (НЕ один прогон — цикл объединяет несколько прогонов: objective по
|
||
# group_name, kn по region_codes+resume). Проверяется ТОЛЬКО при
|
||
# alert_on_zero_output=True: recent_output < min_output_rows → даунгрейд (строгое <).
|
||
# Окно = fresh_days, поэтому здоровый недельный прогон возрастом 7–8d остаётся
|
||
# внутри окна и не ложно-срабатывает (#1947 fix). Default 1 → флагует только если
|
||
# суммарный выход цикла = 0 (безопасный минимальный catch).
|
||
min_output_rows: int = 1
|
||
|
||
|
||
# Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на
|
||
# проде (information_schema) — статусы done присутствуют. nspd — data-table режим:
|
||
# смотрит nspd_quarter_dumps.fetched_at_utc (живой harvest harvest_stale_quarters +
|
||
# write-back /analyze-резолвера #1891), а НЕ defunct ledger nspd_scrape_runs (3 WAF-
|
||
# забаненных manual-прогона 2026-04-30, никогда не шедулился → ложный "no successful runs").
|
||
# Пороги:
|
||
# kn — beat ~еженедельно (scrape_kn cron, Mon) → fresh<8d, stale<14d
|
||
# kn_flats — та же run-таблица, work_col=flats_count + alert_on_zero_output +
|
||
# min_output_rows=50000 (issue #1945: kn-run status=done но flats_count
|
||
# мал/нестабилен — монитор смотрел objects_count ~1548, не flats_count;
|
||
# #1947 fix: floor сравнивается с SUM(flats_count) по done-прогонам цикла
|
||
# в окне fresh_days=8d — ловит low-output 9/2296/3670 и не ложит на 7–8d
|
||
# aging) → fresh<8d, stale<14d
|
||
# objective — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d
|
||
# nspd — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d
|
||
# nspd_geo — ручной bulk geo-job → fresh<7d, stale<30d
|
||
# cadastre — ручной cadastre-job → fresh<14d, stale<45d
|
||
_FRESHNESS_SOURCES: list[FreshnessSource] = [
|
||
FreshnessSource(
|
||
source="kn",
|
||
label="DOM.РФ kn-API (ЖК)",
|
||
table="kn_scrape_runs",
|
||
work_col="objects_count",
|
||
# kn шедулится ЕЖЕНЕДЕЛЬНО (settings.scrape_kn_cron='15 4 * * mon', Mon 04:15 МСК),
|
||
# поэтому 2/5 дней давали ложный "stale" каждую среду-воскресенье. Калибруем под
|
||
# недельную каденцию (как objective 7/14): fresh до 8 дней (полная неделя + добег
|
||
# понедельничного прогона), stale при пропуске ~двух понедельников.
|
||
fresh_days=8.0,
|
||
stale_days=14.0,
|
||
critical=True,
|
||
),
|
||
FreshnessSource(
|
||
# kn-run пишет ДВА счётчика: objects_count (ЖК, ~1548 — выше) и flats_count
|
||
# (квартиры). Источник истины для "лоадер прогнался, но почти ничего не извлёк"
|
||
# (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count мал/нестабилен,
|
||
# монитор молчал — objects_count оставался здоровым). Та же run-таблица и пороги,
|
||
# что у kn, но work_col=flats_count + alert_on_zero_output + min_output_rows.
|
||
#
|
||
# Floor сравнивается с recent_output = SUM(flats_count) по ВСЕМ done-прогонам в
|
||
# окне fresh_days=8d (kn пишет строку на region_codes + resume — это НЕ один
|
||
# прогон). Магнитуды (прод gendesign-postgres-1, #1947): healthy цикл ~376k
|
||
# квартир (snapshot 05-17=376604); сломанное состояние — суммарно ≤3670, сейчас
|
||
# SUM(flats_count,8d)=9 (latest done = resume, считает только хвост; ground truth
|
||
# snapshot 2026-06-22=9, сломан 5+ недель). Порог 50000 чисто разделяет:
|
||
# healthy ≫ 50k ≫ broken ≤3670.
|
||
#
|
||
# KNOWN LIMITATION: zombie-resume, загрузивший лишь малый хвост БЕЗ полного
|
||
# прогона в окне, может транзиентно недосчитать → максимум одно-цикловый ложный
|
||
# алёрт, самоисцеляется следующим полным прогоном. Приемлемо (а в текущем реальном
|
||
# resume-кейсе snapshot реально сломан=9 → флаг корректен).
|
||
source="kn_flats",
|
||
label="DOM.РФ kn-API (квартиры)",
|
||
table="kn_scrape_runs",
|
||
work_col="flats_count",
|
||
fresh_days=8.0,
|
||
stale_days=14.0,
|
||
critical=True,
|
||
alert_on_zero_output=True,
|
||
min_output_rows=50000,
|
||
),
|
||
FreshnessSource(
|
||
source="objective",
|
||
label="Объектив (лоты/цены)",
|
||
table="objective_scrape_runs",
|
||
work_col="rows_lots",
|
||
fresh_days=7.0,
|
||
stale_days=14.0,
|
||
critical=True,
|
||
# Объектив-sync, прогнавшийся с почти нулём лотов — тоже поломка (тихий
|
||
# частичный extract), а не легитимный low. objective пишет строку на group_name
|
||
# (4 группы: Свердловская/Тюмень/Пермь/Челябинск), поэтому floor сравнивается с
|
||
# recent_output = SUM(rows_lots) по группам в окне fresh_days=7d (прод: SUM 7d=
|
||
# 946264; даже наименьшая отдельная группа Челябинск=68387). Floor 1000 ловит
|
||
# вырожденный цикл с огромным запасом, не задевая нормальный разброс — и
|
||
# робастен к тому, КАКАЯ группа дописалась последней (single-newest был бы
|
||
# хрупок: 324570 или 68387 в зависимости от порядка финиша).
|
||
alert_on_zero_output=True,
|
||
min_output_rows=1000,
|
||
),
|
||
FreshnessSource(
|
||
source="nspd",
|
||
label="NSPD quarter dumps (harvest+resolver)",
|
||
# Data-table режим: nspd_quarter_dumps — это контент-таблица квартальных
|
||
# дампов (NO status/started/finished). Живой harvest пишет fetched_at_utc;
|
||
# defunct nspd_scrape_runs (manual WAF-ban 2026-04-30) больше НЕ источник истины.
|
||
table="nspd_quarter_dumps",
|
||
timestamp_col="fetched_at_utc",
|
||
# В timestamp-режиме не используется — оставляем валидное имя колонки.
|
||
work_col="total_features",
|
||
# Медленный кадастровый + lazy-refresh источник: дампы освежаются по мере
|
||
# устаревания кварталов / on-demand из /analyze. Текущий возраст ~4d → fresh.
|
||
fresh_days=14.0,
|
||
stale_days=30.0,
|
||
),
|
||
FreshnessSource(
|
||
source="nspd_geo",
|
||
label="NSPD geo bulk-fetcher",
|
||
table="nspd_geo_jobs",
|
||
work_col="targets_done",
|
||
attempt_fallback_col="created_at",
|
||
# On-demand источник БЕЗ cron (admin UI / CLI / lazy из analyze) — штучные
|
||
# фетчи по активности пользователя. fresh_days=7 флагал каждую неделю
|
||
# тишины ложным stale-алертом (расследование 2026-07-04); пороги — под
|
||
# реальную каденцию, critical=False и так не трогает overall.
|
||
fresh_days=30.0,
|
||
stale_days=90.0,
|
||
),
|
||
FreshnessSource(
|
||
source="cadastre",
|
||
label="Кадастровые jobs",
|
||
table="cadastre_jobs",
|
||
work_col="targets_done",
|
||
attempt_fallback_col="created_at",
|
||
fresh_days=14.0,
|
||
stale_days=45.0,
|
||
),
|
||
]
|
||
|
||
|
||
def _classify_freshness(age_days: float | None, fresh_days: float, stale_days: float) -> str:
|
||
"""fresh / stale / critical / unknown по возрасту последнего успешного прогона."""
|
||
if age_days is None:
|
||
return "unknown"
|
||
if age_days < fresh_days:
|
||
return "fresh"
|
||
if age_days < stale_days:
|
||
return "stale"
|
||
return "critical"
|
||
|
||
|
||
def compute_freshness(db: Session) -> dict[str, Any]:
|
||
"""Свежесть данных по каждому источнику (для dashboard + alerting).
|
||
|
||
Чистая функция (один источник истины для HTTP-эндпоинта и beat-таски
|
||
scrape_freshness_check). Принимает открытую сессию, НЕ коммитит/закрывает её —
|
||
жизненным циклом сессии управляет вызывающий (FastAPI-зависимость или таска).
|
||
|
||
Для run-ledger источников один проход по run-таблице агрегирует:
|
||
- last_success_at — max(COALESCE(finished_at, started_at)) WHERE status='done'
|
||
- last_attempt_at — max(started_at) (или created_at fallback) по всем прогонам
|
||
- last_status — status последнего прогона
|
||
- objects_updated_24h / _7d — сумма run-счётчика по done-прогонам в окне
|
||
- recent_output — SUM(work_col) по done-прогонам цикла в окне fresh_days (NULL
|
||
если успехов в окне нет)
|
||
- age_days — возраст last_success в днях (NULL если успехов не было)
|
||
- freshness — fresh / stale / critical / unknown (по порогам источника)
|
||
- status — ok / stale / failed (для UI-светофора)
|
||
- reason — пояснение к output-floor-даунгрейду (иначе None)
|
||
|
||
Output-floor (issue #1945, redesigned #1947): для run-ledger источников с
|
||
alert_on_zero_output=True источник, который иначе был бы fresh→ok (недавний
|
||
успешный прогон), но SUM(work_col) по ВСЕМ done-прогонам цикла в окне fresh_days
|
||
(recent_output) < min_output_rows, ДАУНГРЕЙДИТСЯ до failed — ловит
|
||
"лоадер-прогнался-но-почти-ничего-не-извлёк" (kn flats_count суммарно 9 при
|
||
здоровых ~376k). Cycle-SUM (не один прогон) робастен к multi-run циклам: objective
|
||
пишет строку на group_name, kn — на region_codes + resume. Окно fresh_days (не
|
||
хардкод 7d) → здоровый недельный прогон 7–8d остаётся в окне (нет aging-FP).
|
||
Только когда возраст НЕ покрывает поломку (status=='ok'), чтобы не двойнить.
|
||
|
||
Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps)
|
||
нет run-ledger семантики (status/started/finished отсутствуют), поэтому:
|
||
- last_success_at = last_attempt_at = MAX(timestamp_col)
|
||
- objects_updated_24h / _7d = COUNT(*) строк, обновлённых в окне
|
||
- last_status = NULL (косметика только для run-ledger'ов)
|
||
Остальной downstream (age_days / _classify_freshness / status-маппинг) — общий.
|
||
|
||
`overall_status` агрегирует худший статус по критичным источникам.
|
||
"""
|
||
sources_out: list[dict[str, Any]] = []
|
||
worst_critical = "ok"
|
||
severity = {"ok": 0, "stale": 1, "failed": 2}
|
||
|
||
for src in _FRESHNESS_SOURCES:
|
||
# table / column-имена — из доверенного серверного реестра (НЕ из
|
||
# пользовательского ввода), поэтому f-string в SQL здесь безопасен;
|
||
# все временные границы передаются параметрами (:d1/:d7).
|
||
if src.timestamp_col is not None:
|
||
# Data-table режим: плоская контент-таблица без run-ledger семантики
|
||
# (нет status/started/finished). Свежесть = MAX(timestamp_col),
|
||
# upd_24h/_7d = COUNT(*) строк, обновлённых в окне. last_status=NULL
|
||
# (косметика только для run-ledger'ов).
|
||
ts = src.timestamp_col
|
||
row = (
|
||
db.execute(
|
||
text(
|
||
f"""
|
||
SELECT
|
||
MAX({ts}) AS last_success_at,
|
||
MAX({ts}) AS last_attempt_at,
|
||
COALESCE(COUNT(*) FILTER (
|
||
WHERE {ts} > NOW() - CAST(:d1 AS interval)
|
||
), 0) AS upd_24h,
|
||
COALESCE(COUNT(*) FILTER (
|
||
WHERE {ts} > NOW() - CAST(:d7 AS interval)
|
||
), 0) AS upd_7d,
|
||
NULL AS last_status
|
||
FROM {src.table}
|
||
"""
|
||
),
|
||
{"d1": "24 hours", "d7": "7 days"},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
else:
|
||
started = src.started_col
|
||
attempt_expr = (
|
||
f"MAX(COALESCE({started}, {src.attempt_fallback_col}))"
|
||
if src.attempt_fallback_col
|
||
else f"MAX({started})"
|
||
)
|
||
# last_status — отдельным скаляром: status последнего по времени прогона.
|
||
order_col = (
|
||
f"COALESCE({started}, {src.attempt_fallback_col})"
|
||
if src.attempt_fallback_col
|
||
else started
|
||
)
|
||
# FILTER применяется к SUM(...), а COALESCE(...,0) оборачивает результат —
|
||
# COALESCE(...) FILTER(...) — синтаксическая ошибка в PostgreSQL.
|
||
row = (
|
||
db.execute(
|
||
text(
|
||
f"""
|
||
SELECT
|
||
MAX(COALESCE(finished_at, {started}))
|
||
FILTER (WHERE status = 'done') AS last_success_at,
|
||
{attempt_expr} AS last_attempt_at,
|
||
COALESCE(SUM({src.work_col}) FILTER (
|
||
WHERE status = 'done'
|
||
AND COALESCE(finished_at, {started})
|
||
> NOW() - CAST(:d1 AS interval)
|
||
), 0) AS upd_24h,
|
||
COALESCE(SUM({src.work_col}) FILTER (
|
||
WHERE status = 'done'
|
||
AND COALESCE(finished_at, {started})
|
||
> NOW() - CAST(:d7 AS interval)
|
||
), 0) AS upd_7d,
|
||
(
|
||
SELECT status FROM {src.table}
|
||
ORDER BY {order_col} DESC NULLS LAST
|
||
LIMIT 1
|
||
) AS last_status,
|
||
-- Cycle-sum: SUM(work_col) по ВСЕМ done-прогонам в окне
|
||
-- fresh_days. Логический цикл загрузки = НЕСКОЛЬКО прогонов
|
||
-- (objective — по строке на group_name, kn — по region_codes
|
||
-- + resume), поэтому суммируем, а не берём один прогон.
|
||
-- Окно = fresh_days (per-source), а НЕ хардкод 7d: здоровый
|
||
-- недельный прогон возрастом 7–8d остаётся ВНУТРИ окна →
|
||
-- нет aging-FP (#1947). Inner COALESCE(work_col,0) — NULL-
|
||
-- счётчик на done-прогоне вносит 0, не NULL. fresh_days —
|
||
-- float из доверенного реестра → f-string безопасен. Окно
|
||
-- через make_interval(secs => fresh_days*86400): secs —
|
||
-- double precision, поэтому дробный fresh_days валиден (у
|
||
-- days int-only → 7.5 упал бы). NB без двоеточийного cast.
|
||
SUM(COALESCE({src.work_col}, 0)) FILTER (
|
||
WHERE status = 'done'
|
||
AND COALESCE(finished_at, {started})
|
||
> now()
|
||
- make_interval(secs => {src.fresh_days} * 86400)
|
||
) AS recent_output
|
||
FROM {src.table}
|
||
"""
|
||
),
|
||
{"d1": "24 hours", "d7": "7 days"},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
|
||
last_success = row["last_success_at"] if row else None
|
||
last_attempt = row["last_attempt_at"] if row else None
|
||
last_status = row["last_status"] if row else None
|
||
# SUM(work_col) по done-прогонам в окне fresh_days (run-ledger режим = весь
|
||
# цикл загрузки). В data-table режиме колонки нет → None; .get() терпит
|
||
# отсутствие ключа в mapping-строке. None также когда done-прогонов в окне нет.
|
||
recent_output_raw = row.get("recent_output") if row else None
|
||
recent_output: int | None = (
|
||
int(recent_output_raw) if recent_output_raw is not None else None
|
||
)
|
||
|
||
age_days: float | None = None
|
||
if last_success is not None:
|
||
age_seconds = (_utcnow() - last_success).total_seconds()
|
||
age_days = round(age_seconds / 86400.0, 2)
|
||
|
||
freshness = _classify_freshness(age_days, src.fresh_days, src.stale_days)
|
||
|
||
upd_24h = int(row["upd_24h"] or 0) if row else 0
|
||
upd_7d = int(row["upd_7d"] or 0) if row else 0
|
||
|
||
# status (UI-светофор): нет успехов вообще → failed; иначе мапим freshness.
|
||
if last_success is None:
|
||
status = "failed"
|
||
elif freshness == "fresh":
|
||
status = "ok"
|
||
elif freshness == "stale":
|
||
status = "stale"
|
||
else: # critical
|
||
status = "failed"
|
||
|
||
# Output-floor downgrade (issue #1945, redesigned #1947): источник прогнался
|
||
# успешно И недавно (был бы fresh→ok по возрасту), НО SUM(work_col) по всем
|
||
# done-прогонам ЦИКЛА (окно fresh_days) ниже порога min_output_rows → ДАУНГРЕЙД
|
||
# до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила.
|
||
# - cycle-SUM (а не один прогон): цикл = несколько done-прогонов (objective —
|
||
# по group_name, kn — по region_codes + resume), сумма даёт магнитуду цикла
|
||
# (objective ~946k, kn_flats ~376k healthy) → робастно к multi-run.
|
||
# - окно fresh_days (а не хардкод 7d): здоровый недельный прогон 7–8d остаётся
|
||
# внутри окна → нет aging-FP первой версии #1947.
|
||
# Только run-ledger (timestamp_col is None) c alert_on_zero_output=True и ТОЛЬКО
|
||
# когда источник иначе был бы fresh/ok (по возрасту stale/failed уже покрыт — не
|
||
# двойним). status=='ok' гарантирует недавний успех (last_success is not None).
|
||
# Строгое '<': SUM ровно на пороге → ok (порог = минимально-приемлемый объём).
|
||
reason: str | None = None
|
||
if (
|
||
src.alert_on_zero_output
|
||
and src.timestamp_col is None
|
||
and status == "ok"
|
||
and recent_output is not None
|
||
and recent_output < src.min_output_rows
|
||
):
|
||
status = "failed"
|
||
reason = (
|
||
f"low-output: done-прогоны за {src.fresh_days:g}d дали "
|
||
f"SUM({src.work_col})={recent_output} < порога {src.min_output_rows} "
|
||
f"(#1945, #1947)"
|
||
)
|
||
|
||
if src.critical and severity[status] > severity[worst_critical]:
|
||
worst_critical = status
|
||
|
||
sources_out.append(
|
||
{
|
||
"source": src.source,
|
||
"label": src.label,
|
||
"critical": src.critical,
|
||
"last_success_at": last_success.isoformat() if last_success else None,
|
||
"last_attempt_at": last_attempt.isoformat() if last_attempt else None,
|
||
"last_status": last_status,
|
||
"objects_updated_24h": upd_24h,
|
||
"objects_updated_7d": upd_7d,
|
||
# SUM(work_col) по done-прогонам цикла в окне fresh_days (None для
|
||
# data-table режима / без done-прогонов в окне). Additive — база для
|
||
# output-floor-проверки, полезно и для UI-диагностики.
|
||
"recent_output": recent_output,
|
||
"age_days": age_days,
|
||
"thresholds": {"fresh_days": src.fresh_days, "stale_days": src.stale_days},
|
||
"freshness": freshness,
|
||
"status": status,
|
||
# Доп. контекст при output-floor-даунгрейде (None в обычном случае).
|
||
# Additive — не ломает существующий UI/JSON-контракт.
|
||
"reason": reason,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"generated_at": _utcnow().isoformat(),
|
||
"overall_status": worst_critical,
|
||
"sources": sources_out,
|
||
}
|
||
|
||
|
||
@router.get("/freshness")
|
||
def scrape_freshness(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""GET-обёртка над compute_freshness — только читает, не алертит.
|
||
|
||
Вся агрегация живёт в compute_freshness (один источник истины); эндпоинт
|
||
лишь прокидывает сессию и возвращает результат без изменений формы payload.
|
||
Sentry-алёрты по критичным источникам шлёт beat-таска scrape_freshness_check.
|
||
"""
|
||
return compute_freshness(db)
|
||
|
||
|
||
# ── DomRF (kn) — runs list ───────────────────────────────────────────────────
|
||
|
||
|
||
@router.get("/runs")
|
||
def list_runs(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
limit: int = Query(default=20, ge=0),
|
||
) -> list[dict[str, Any]]:
|
||
rows = (
|
||
db.execute(
|
||
text(
|
||
"""
|
||
SELECT run_id, started_at, finished_at, region_codes, developer_ids,
|
||
objects_count, flats_count, requests_count, status, error, snapshot_date
|
||
FROM kn_scrape_runs
|
||
ORDER BY started_at DESC
|
||
LIMIT :lim
|
||
"""
|
||
),
|
||
{"lim": min(limit, 100)},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return [
|
||
{
|
||
"run_id": r["run_id"],
|
||
"started_at": r["started_at"].isoformat() if r["started_at"] else None,
|
||
"finished_at": r["finished_at"].isoformat() if r["finished_at"] else None,
|
||
"region_codes": list(r["region_codes"]) if r["region_codes"] else [],
|
||
"developer_ids": list(r["developer_ids"]) if r["developer_ids"] else [],
|
||
"objects_count": r["objects_count"],
|
||
"flats_count": r["flats_count"],
|
||
"requests_count": r["requests_count"],
|
||
"status": r["status"],
|
||
"error": r["error"],
|
||
"snapshot_date": r["snapshot_date"].isoformat() if r["snapshot_date"] else None,
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
# ── Ekburg construction permits (РНС/РВЭ ЕКБ) ────────────────────────────────
|
||
|
||
|
||
class TriggerEkburgPermitsRequest(BaseModel):
|
||
year: int | None = Field(
|
||
default=None,
|
||
ge=2022,
|
||
le=2030,
|
||
description=(
|
||
"Если None — refresh_all (все 5 лет 2022-2026). "
|
||
"Иначе refresh_year(year) для конкретного года."
|
||
),
|
||
)
|
||
|
||
|
||
@router.post("/ekburg-permits")
|
||
def trigger_ekburg_permits(
|
||
payload: TriggerEkburgPermitsRequest,
|
||
) -> dict[str, Any]:
|
||
"""Manual trigger для синхронизации РНС/РВЭ ЕКБ из екатеринбург.рф (xlsx).
|
||
|
||
Обычно запускается раз в месяц через beat (1-го числа 02:00 UTC).
|
||
Этот endpoint — для ad-hoc запуска (после первого деплоя или для smoke-теста).
|
||
|
||
- year=None → refresh_all: скачивает 5 xlsx (2022-2026), ~3-8 мин.
|
||
- year=2026 → refresh_year(2026): один файл, ~1 мин.
|
||
"""
|
||
from app.workers.tasks.ekburg_permits_sync import refresh_all, refresh_year
|
||
|
||
if payload.year is None:
|
||
result = refresh_all.apply_async()
|
||
scope = "all_years_2022_2026"
|
||
else:
|
||
result = refresh_year.apply_async(args=[payload.year])
|
||
scope = f"year_{payload.year}"
|
||
|
||
return {"task_id": result.id, "scope": scope, "queued_at": "now"}
|
||
|
||
|
||
class TriggerKnCatalogObjectsRequest(BaseModel):
|
||
region_code: int = Field(default=66, ge=1, le=99)
|
||
max_objects: int | None = Field(default=None, ge=1, le=2000)
|
||
force: bool = Field(
|
||
default=False,
|
||
description=(
|
||
"True — игнорировать фильтр 'уже сегодня обновлён' и грузить ВСЕ объекты "
|
||
"последнего snapshot ('Загрузить все'). По умолчанию пропускает то, "
|
||
"что уже скраплено сегодня."
|
||
),
|
||
)
|
||
|
||
|
||
@router.post("/kn-catalog-objects")
|
||
def trigger_kn_catalog_objects(
|
||
payload: TriggerKnCatalogObjectsRequest,
|
||
) -> dict[str, Any]:
|
||
"""Manual trigger для catalog-OBJECT scraper (заполняет wall_type, energy_eff,
|
||
ceiling_height_m, parking_*, playground_*, scores из SSR __NEXT_DATA__).
|
||
|
||
Beat schedule: Tuesday 04:00 UTC, batch 300/run. Этот endpoint — для ad-hoc
|
||
запуска (smoke-тест после деплоя или повторный pass для свежесозданных
|
||
объектов до next beat fire).
|
||
|
||
- max_objects=None → дефолтный лимит таска (300).
|
||
- max_objects=3 → smoke-тест.
|
||
- force=True → "Загрузить все": игнорирует skip-today, грузит всё подряд.
|
||
"""
|
||
from app.workers.tasks.scrape_kn_catalog_objects import scrape_kn_catalog_objects
|
||
|
||
kwargs: dict[str, Any] = {
|
||
"region_code": payload.region_code,
|
||
"force": payload.force,
|
||
}
|
||
if payload.max_objects is not None:
|
||
kwargs["max_objects"] = payload.max_objects
|
||
result = scrape_kn_catalog_objects.apply_async(kwargs=kwargs)
|
||
|
||
return {
|
||
"task_id": result.id,
|
||
"region_code": payload.region_code,
|
||
"max_objects": payload.max_objects,
|
||
"force": payload.force,
|
||
"queued_at": "now",
|
||
}
|