69 lines
2.8 KiB
Python
69 lines
2.8 KiB
Python
"""
|
|
===============================================================================
|
|
FILE: services/snapshots/service.py
|
|
PROJECT: SCUD Orion AI (Unified Architecture)
|
|
MODULE: services / snapshots
|
|
ROLE: Бизнес-логика срезов СКУД (выборка, валидация Y-срезов, удаление).
|
|
|
|
AI-CONTEXT-ANCHORS:
|
|
- ANCHOR[SNAPSHOT_GET_REGISTRY]: Выборка срезов с разметкой защищенных Y-снапшотов.
|
|
- ANCHOR[SNAPSHOT_DELETE_SAFE]: Безопасное удаление дневных срезов с защитой итоговых.
|
|
===============================================================================
|
|
"""
|
|
|
|
from typing import Dict, Any, Optional, List
|
|
from core.connection import get_connection
|
|
from core.repositories.scud_repo import get_available_snapshots, delete_snapshot_by_id
|
|
|
|
|
|
# ANCHOR[SNAPSHOT_GET_REGISTRY]
|
|
def get_snapshots_registry(date_str: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Возвращает реестр снапшотов за дату или за все доступные дни."""
|
|
clean_date = date_str.strip() if date_str else ""
|
|
rows = get_available_snapshots(date_str=clean_date if clean_date else None)
|
|
|
|
snapshots = [
|
|
{
|
|
"snapshot_id": r[0],
|
|
"log_date": r[1],
|
|
"snapshot_time": r[2],
|
|
"record_count": r[3],
|
|
"is_final": str(r[0]).startswith("Y")
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
return {
|
|
"query_date": clean_date or "все",
|
|
"snapshots_count": len(snapshots),
|
|
"snapshots": snapshots
|
|
}
|
|
|
|
|
|
# ANCHOR[SNAPSHOT_DELETE_SAFE]
|
|
def delete_snapshots_safely(snapshot_ids: List[str]) -> Dict[str, Any]:
|
|
"""
|
|
Удаляет выбранные дневные снапшоты.
|
|
Итоговые вечерние срезы с префиксом 'Y' гарантированно защищены от удаления.
|
|
"""
|
|
if not snapshot_ids:
|
|
return {"status": "error", "message": "Не указаны ID снапшотов для удаления."}
|
|
|
|
safe_ids = [str(s).strip() for s in snapshot_ids if s and not str(s).strip().startswith("Y")]
|
|
|
|
if not safe_ids:
|
|
return {"status": "error", "message": "⚠️ Итоговый срез Y защищен от удаления. Выберите дневные снапшоты."}
|
|
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
placeholders = ",".join(["?"] * len(safe_ids))
|
|
cursor.execute(f"DELETE FROM scud_logs WHERE snapshot_id IN ({placeholders})", safe_ids)
|
|
deleted_count = cursor.rowcount
|
|
conn.commit()
|
|
|
|
return {
|
|
"status": "success",
|
|
"deleted_count": deleted_count,
|
|
"deleted_ids": safe_ids,
|
|
"message": f"Успешно удалено снапшотов: {len(safe_ids)} шт."
|
|
} |