236 lines
10 KiB
Python
236 lines
10 KiB
Python
"""Admin endpoints для ETL operations (#203, #44).
|
||
|
||
POST /api/v1/admin/etl/objective-backfill
|
||
Запустить fuzzy-match backfill objective_complex_mapping.
|
||
Поддерживает два режима:
|
||
- v1 (default): threshold=0.85, match_method='fuzzy_trgm'
|
||
- v2: threshold=0.80, match_method='fuzzy_v2' — для coverage 8.5%→17%
|
||
|
||
POST /api/v1/admin/etl/nspd-denorm-backfill
|
||
Запустить backfill nspd_parcels/nspd_buildings из всех nspd_quarter_dumps.
|
||
|
||
GET /api/v1/admin/etl/mapping-review
|
||
Список авто-маппингов objective_complex_mapping + обогащение для ревью.
|
||
POST /api/v1/admin/etl/mapping-review/{id}/approve
|
||
Подтвердить маппинг (is_reviewed=true + note append).
|
||
POST /api/v1/admin/etl/mapping-review/{id}/reject
|
||
Отклонить маппинг — DELETE строки (иначе травит mv_layout_velocity §4.2).
|
||
|
||
Auth: gendsgn.ru-wide Caddy basic_auth gate (PR #426). App-level X-Admin-Token
|
||
header removed 2026-05-23 — двойная auth избыточна для pilot.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Annotated, Any
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.db import get_db
|
||
from app.services.etl.mapping_review import (
|
||
approve_mapping,
|
||
list_mapping_review,
|
||
reject_mapping,
|
||
)
|
||
from app.services.etl.objective_backfill import (
|
||
AUTO_ACCEPT_THRESHOLD,
|
||
AUTO_ACCEPT_THRESHOLD_V2,
|
||
REVIEW_THRESHOLD,
|
||
apply_core_matches,
|
||
auto_apply_matches,
|
||
find_core_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)],
|
||
dry_run: Annotated[bool, Query(description="Preview без insertions")] = False,
|
||
refresh_mv: Annotated[
|
||
bool, Query(description="REFRESH mv_layout_velocity после backfill")
|
||
] = True,
|
||
v2: Annotated[
|
||
bool,
|
||
Query(
|
||
description=(
|
||
"v2 mode: threshold=0.80, match_method='fuzzy_v2'. "
|
||
"Запускать после v1 run — только для unmapped объектов."
|
||
)
|
||
),
|
||
] = False,
|
||
) -> dict[str, object]:
|
||
"""Запустить backfill objective_complex_mapping + опционально REFRESH MV.
|
||
|
||
Ищет DOM.РФ комплексы (is_ekb=true) без mapping и применяет fuzzy match
|
||
к project_name из objective_corpus_room_month через pg_trgm similarity.
|
||
|
||
Режим v1 (default, ?v2=false):
|
||
- score >= 0.85 (AUTO_ACCEPT_THRESHOLD): auto-insert, match_method='fuzzy_trgm'
|
||
- score >= 0.6 (REVIEW_THRESHOLD) и < 0.85: только счётчик review_queue
|
||
|
||
Режим v2 (?v2=true, task #44 coverage expansion):
|
||
- score >= 0.80 (AUTO_ACCEPT_THRESHOLD_V2): auto-insert, match_method='fuzzy_v2'
|
||
- is_reviewed=false — требует ручной проверки (false positives вероятны)
|
||
- Целевой прирост: +~47-80 строк, coverage 8.5% → ~17%
|
||
|
||
Returns dict:
|
||
auto_accepted: сколько строк вставлено
|
||
review_queue: сколько кандидатов ниже порога auto-accept
|
||
skipped: ON CONFLICT + ошибки
|
||
mv_rows_after_refresh: строк в MV после REFRESH (0 если refresh_mv=False)
|
||
threshold_used: фактический порог (float)
|
||
match_method_used: match_method в БД (str)
|
||
"""
|
||
threshold = AUTO_ACCEPT_THRESHOLD_V2 if v2 else AUTO_ACCEPT_THRESHOLD
|
||
method = "fuzzy_v2" if v2 else "fuzzy_trgm"
|
||
# v2 ищет кандидатов начиная с threshold (не от REVIEW_THRESHOLD)
|
||
search_min = threshold if v2 else REVIEW_THRESHOLD
|
||
|
||
candidates = find_match_candidates(db, only_unmapped=True, min_threshold=search_min)
|
||
logger.info(
|
||
"Backfill candidates found: %d (score >= %.2f, method=%s)",
|
||
len(candidates),
|
||
search_min,
|
||
method,
|
||
)
|
||
|
||
result: dict[str, object] = dict(
|
||
auto_apply_matches(
|
||
db, candidates, threshold=threshold, match_method=method, dry_run=dry_run
|
||
)
|
||
)
|
||
|
||
mv_rows = 0
|
||
if refresh_mv and not dry_run and result.get("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
|
||
result["threshold_used"] = threshold
|
||
result["match_method_used"] = method
|
||
return result
|
||
|
||
|
||
@router.post("/objective-core-backfill")
|
||
def run_objective_core_backfill(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
dry_run: Annotated[bool, Query(description="Preview без insertions (default True)")] = True,
|
||
refresh_mv: Annotated[bool, Query(description="REFRESH mv_layout_velocity после apply")] = True,
|
||
) -> dict[str, object]:
|
||
"""Core-name + dev-confirm проход (#2177 шаг 2, match_method='auto_core_dev_v5').
|
||
|
||
Нормализует ЯДРО имени обеих сторон (normalize_complex_name, #2198) и мапит
|
||
несопоставленные domrf ЕКБ-объекты на objective-проекты. Пишутся ТОЛЬКО tier_a
|
||
(ровно 1 кандидат + подтверждение застройщика). tier_b/ambiguous — в counts.
|
||
|
||
Дефолт dry_run=True (безопасно). Реальный insert — ?dry_run=false.
|
||
|
||
Returns dict:
|
||
inserted / conflict_skipped / error_skipped / tier_a_total — из apply;
|
||
tier_b / ambiguous / skipped_taken — счётчики отчёта;
|
||
mv_rows_after_refresh — строк в MV после REFRESH (0 если dry-run / нет вставок).
|
||
"""
|
||
report = find_core_matches(db)
|
||
result: dict[str, object] = dict(apply_core_matches(db, report, dry_run=dry_run))
|
||
result.update(report.counts())
|
||
|
||
mv_rows = 0
|
||
if refresh_mv and not dry_run and result.get("inserted", 0):
|
||
mv_rows = trigger_mv_refresh(db)
|
||
logger.info("mv_layout_velocity refreshed after core-backfill: %d rows", mv_rows)
|
||
result["mv_rows_after_refresh"] = mv_rows
|
||
return result
|
||
|
||
|
||
@router.post("/nspd-denorm-backfill")
|
||
def run_nspd_denorm_backfill(
|
||
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}
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# Mapping review — глазами подтвердить/отклонить авто-маппинги (feat/…-review)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
@router.get("/mapping-review")
|
||
def get_mapping_review(
|
||
db: Annotated[Session, Depends(get_db)],
|
||
only_unreviewed: Annotated[
|
||
bool, Query(description="Только is_reviewed=false (авто-матчи без ревью)")
|
||
] = True,
|
||
limit: Annotated[int, Query(ge=1, le=500, description="Строк на странице")] = 100,
|
||
offset: Annotated[int, Query(ge=0, description="Смещение страницы")] = 0,
|
||
) -> dict[str, Any]:
|
||
"""Список строк objective_complex_mapping + обогащение для ревьюера.
|
||
|
||
Обогащение каждой строки:
|
||
- domrf_comm_name / domrf_dev_name — latest snapshot по domrf_obj_id;
|
||
- objective_developers — застройщики objective-проекта (агрегат
|
||
objective_lots.developer, scoped по project_name).
|
||
|
||
Сорт: is_reviewed asc, match_score asc NULLS FIRST (сомнительные сверху).
|
||
|
||
Returns dict:
|
||
rows: list обогащённых строк (id, objective_complex_name,
|
||
objective_project_id, objective_group, domrf_obj_id, match_method,
|
||
match_score, is_reviewed, note, created_at, domrf_comm_name,
|
||
domrf_dev_name, objective_developers);
|
||
total: всего строк (с учётом only_unreviewed, БЕЗ limit/offset);
|
||
limit / offset / only_unreviewed: эхо параметров.
|
||
"""
|
||
return list_mapping_review(db, only_unreviewed=only_unreviewed, limit=limit, offset=offset)
|
||
|
||
|
||
@router.post("/mapping-review/{mapping_id}/approve")
|
||
def approve_mapping_review(
|
||
mapping_id: int,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Подтвердить маппинг: is_reviewed=true + append «approved <date>» в note.
|
||
|
||
404 если строки с таким id нет.
|
||
"""
|
||
result = approve_mapping(db, mapping_id)
|
||
if result is None:
|
||
raise HTTPException(status_code=404, detail=f"mapping {mapping_id} не найден")
|
||
return result
|
||
|
||
|
||
@router.post("/mapping-review/{mapping_id}/reject")
|
||
def reject_mapping_review(
|
||
mapping_id: int,
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> dict[str, Any]:
|
||
"""Отклонить маппинг — DELETE строки (иначе травит mv_layout_velocity §4.2).
|
||
|
||
Удаляемая строка целиком логируется (logger.info) перед commit — в note
|
||
писать некуда, строки не будет. 404 если id не найден.
|
||
"""
|
||
deleted = reject_mapping(db, mapping_id)
|
||
if deleted is None:
|
||
raise HTTPException(status_code=404, detail=f"mapping {mapping_id} не найден")
|
||
return {"status": "deleted", "deleted": deleted}
|