88 lines
3.6 KiB
Python
88 lines
3.6 KiB
Python
"""
|
||
===============================================================================
|
||
FILE: services/snapshots/service.py
|
||
PROJECT: SCUD Orion AI (Unified Architecture)
|
||
MODULE: services / snapshots
|
||
ROLE: Бизнес-логика срезов СКУД (выборка, отображение времени, удаление).
|
||
|
||
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]:
|
||
"""
|
||
Возвращает реестр снапшотов.
|
||
В поле snapshot_time объединяет время среза и фактическое время создания снапшота.
|
||
"""
|
||
clean_date = date_str.strip() if date_str else ""
|
||
rows = get_available_snapshots(date_str=clean_date if clean_date else None)
|
||
|
||
snapshots = []
|
||
for r in rows:
|
||
snap_id = r[0]
|
||
log_date = r[1]
|
||
snap_time = r[2]
|
||
rec_count = r[3]
|
||
created_at = r[4] if len(r) > 4 else None
|
||
|
||
slice_time_str = snap_time.split()[1] if snap_time and " " in snap_time else snap_time
|
||
|
||
created_str = ""
|
||
if created_at and " " in str(created_at):
|
||
c_date, c_time = str(created_at).split()[:2]
|
||
c_parts = c_date.split("-")
|
||
c_date_fmt = f"{c_parts[2]}.{c_parts[1]}" if len(c_parts) == 3 else c_date
|
||
created_str = f" · создан {c_date_fmt} {c_time[:5]}"
|
||
|
||
display_label = f"Срез {slice_time_str}{created_str}"
|
||
|
||
snapshots.append({
|
||
"snapshot_id": snap_id,
|
||
"log_date": log_date,
|
||
"snapshot_time": display_label,
|
||
"record_count": rec_count,
|
||
"is_final": str(snap_id).startswith("Y") or "_FINAL" in str(snap_id)
|
||
})
|
||
|
||
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)} шт."
|
||
} |