db1ff5b1fc
- data/sql/99_nspd_entities_denorm.sql: 2 tables (cad_num PK) + 6 indexes (quarter_cad, GIST geom, partial WHERE purpose ILIKE '%многокв%') - nspd_denorm.py: denorm_parcel_feature / denorm_building_feature / denorm_dump - SAVEPOINT per row via with db.begin_nested() - ST_Transform(ST_SetSRID(ST_GeomFromGeoJSON(:geom), 3857), 4326) - json.dumps(geom, ensure_ascii=False) for psycopg bind - All CAST(:x AS type) — no ❌:type - nspd_sync.py: inline denorm after _upsert_dump in harvest_quarter (non-fatal) - nspd_denorm_backfill.py: Celery one-shot task (per-quarter commit) - admin_etl.py: POST /api/v1/admin/etl/nspd-denorm-backfill (AdminTokenAuth) - 19 tests: coerce helpers, denorm parcel/building, aggregate counts Foundation for downstream features (МКД neighbors lookup, parcel attrs). Part of #94
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
"""Admin endpoints для ETL operations (#203).
|
||
|
||
POST /api/v1/admin/etl/objective-backfill
|
||
Запустить fuzzy-match backfill objective_complex_mapping.
|
||
Опциональный REFRESH mv_layout_velocity после успешного backfill.
|
||
|
||
POST /api/v1/admin/etl/nspd-denorm-backfill
|
||
Запустить backfill nspd_parcels/nspd_buildings из всех nspd_quarter_dumps.
|
||
|
||
Header: X-Admin-Token: <SCRAPE_ADMIN_TOKEN>
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Annotated
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import get_db
|
||
from app.core.deps import AdminTokenAuth
|
||
from app.services.etl.objective_backfill import (
|
||
REVIEW_THRESHOLD,
|
||
auto_apply_matches,
|
||
find_match_candidates,
|
||
trigger_mv_refresh,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
router = APIRouter()
|
||
|
||
|
||
@router.post("/objective-backfill")
|
||
def run_objective_backfill(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
_: AdminTokenAuth,
|
||
dry_run: Annotated[bool, Query(description="Preview без insertions")] = False,
|
||
refresh_mv: Annotated[
|
||
bool, Query(description="REFRESH mv_layout_velocity после backfill")
|
||
] = True,
|
||
) -> dict[str, int]:
|
||
"""Запустить backfill objective_complex_mapping + опционально REFRESH MV.
|
||
|
||
Ищет DOM.РФ комплексы (is_ekb=true) без mapping и применяет fuzzy match
|
||
к project_name из objective_corpus_room_month через pg_trgm similarity.
|
||
|
||
- score >= 0.85 (AUTO_ACCEPT_THRESHOLD): auto-insert с match_method='fuzzy_trgm'
|
||
- score >= 0.6 (REVIEW_THRESHOLD) и < 0.85: только в счётчик review_queue
|
||
(Phase 2 — UI для ручного review)
|
||
|
||
Returns dict:
|
||
auto_accepted: сколько строк вставлено
|
||
review_queue: сколько кандидатов ниже порога auto-accept
|
||
skipped: ON CONFLICT + ошибки
|
||
mv_rows_after_refresh: строк в MV после REFRESH (0 если refresh_mv=False)
|
||
"""
|
||
candidates = find_match_candidates(db, only_unmapped=True)
|
||
logger.info(
|
||
"Backfill candidates found: %d (score >= %.2f)",
|
||
len(candidates),
|
||
REVIEW_THRESHOLD,
|
||
)
|
||
|
||
result = auto_apply_matches(db, candidates, dry_run=dry_run)
|
||
|
||
mv_rows = 0
|
||
if refresh_mv and not dry_run and result["auto_accepted"] > 0:
|
||
mv_rows = trigger_mv_refresh(db)
|
||
logger.info("mv_layout_velocity refreshed after backfill: %d rows", mv_rows)
|
||
|
||
result["mv_rows_after_refresh"] = mv_rows
|
||
return result
|
||
|
||
|
||
@router.post("/nspd-denorm-backfill")
|
||
def run_nspd_denorm_backfill(
|
||
_: AdminTokenAuth,
|
||
limit: Annotated[
|
||
int | None,
|
||
Query(description="Максимум кварталов для обработки (None = все)"),
|
||
] = None,
|
||
) -> dict[str, object]:
|
||
"""Запустить Celery backfill: denormalize nspd_quarter_dumps → nspd_parcels/nspd_buildings.
|
||
|
||
Задача идемпотентна (ON CONFLICT DO UPDATE). Безопасно запускать повторно.
|
||
Возвращает Celery task_id — статус через /flower или celery inspect.
|
||
|
||
Args:
|
||
limit: если задан — обработать только первые N кварталов ORDER BY quarter_cad.
|
||
"""
|
||
from app.workers.tasks.nspd_denorm_backfill import backfill_all_dumps
|
||
|
||
task = backfill_all_dumps.apply_async(kwargs={"limit": limit})
|
||
logger.info("nspd-denorm-backfill enqueued: task_id=%s limit=%s", task.id, limit)
|
||
return {"task_id": task.id, "status": "enqueued", "limit": limit}
|