112 lines
4.5 KiB
Python
112 lines
4.5 KiB
Python
"""
|
|
===============================================================================
|
|
FILE: modules/web_api/llm/db/db_snapshots.py
|
|
PROJECT: SCUD Orion AI (Unified Architecture)
|
|
MODULE: web_api / llm / db
|
|
ROLE: Выборка, фильтрация и пакетное удаление срезов логов СКУД в SQLite.
|
|
|
|
AI-CONTEXT-ANCHORS:
|
|
- ANCHOR[DB_GET_SNAPSHOTS]: Выборка снапшотов с нормализацией дат.
|
|
- ANCHOR[DB_DEL_SNAPSHOTS]: Удаление снапшотов по ID, списку ID или за дату.
|
|
===============================================================================
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
from typing import Dict, Any, Optional, List
|
|
from .connection import get_db_connection
|
|
from .db_prompts import db_set_session_state
|
|
from ..core.calendar_utils import parse_relative_date_ru
|
|
|
|
# ANCHOR[DB_GET_SNAPSHOTS]
|
|
def db_get_snapshots(session_id: str = "web_session_main", date_str: Optional[str] = None, original_user_message: str = "") -> Dict[str, Any]:
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
query = """
|
|
SELECT
|
|
snapshot_id,
|
|
log_date,
|
|
snapshot_time,
|
|
COUNT(*) as record_count
|
|
FROM scud_logs
|
|
"""
|
|
params = []
|
|
|
|
clean_date = date_str.strip() if date_str else ""
|
|
if not clean_date or not re.search(r'\d{2}\.\d{2}\.\d{4}', clean_date):
|
|
if original_user_message:
|
|
clean_date = parse_relative_date_ru(original_user_message)
|
|
|
|
if clean_date and re.search(r'\d{2}\.\d{2}\.\d{4}', clean_date):
|
|
iso_date = clean_date
|
|
compact_date = clean_date.replace(".", "")
|
|
|
|
if "." in clean_date:
|
|
parts = clean_date.split(".")
|
|
if len(parts) == 3:
|
|
iso_date = f"{parts[2]}-{parts[1]}-{parts[0]}"
|
|
compact_date = f"{parts[2]}{parts[1]}{parts[0]}"
|
|
|
|
query += """
|
|
WHERE log_date = ?
|
|
OR snapshot_time LIKE ?
|
|
OR snapshot_id LIKE ?
|
|
OR snapshot_id LIKE ?
|
|
"""
|
|
params.extend([clean_date, f"{iso_date}%", f"{compact_date}-%", f"Y{compact_date}-%"])
|
|
|
|
query += """
|
|
GROUP BY snapshot_id, log_date, snapshot_time
|
|
ORDER BY snapshot_time DESC, snapshot_id DESC
|
|
LIMIT 50
|
|
"""
|
|
|
|
cursor.execute(query, params)
|
|
rows = cursor.fetchall()
|
|
snapshots = [dict(r) for r in rows]
|
|
|
|
result_data = {
|
|
"query_date": clean_date or "все",
|
|
"snapshots_count": len(snapshots),
|
|
"snapshots": snapshots
|
|
}
|
|
db_set_session_state(session_id=session_id, state_type="SNAPSHOTS_VIEW", data=result_data)
|
|
conn.close()
|
|
return result_data
|
|
|
|
# ANCHOR[DB_DEL_SNAPSHOTS]
|
|
def db_delete_snapshots(
|
|
snapshot_id: Optional[str] = None,
|
|
snapshot_ids: Optional[List[str]] = None,
|
|
day_str: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""Удаляет один или группу дневных снапшотов с защитой итоговых Y-снапшотов."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
deleted = 0
|
|
|
|
if snapshot_ids and isinstance(snapshot_ids, list):
|
|
# Исключаем любые итоговые снапшоты, начинающиеся с Y
|
|
safe_ids = [s.strip() for s in snapshot_ids if s and not str(s).strip().startswith("Y")]
|
|
if safe_ids:
|
|
placeholders = ",".join(["?"] * len(safe_ids))
|
|
cursor.execute(f"DELETE FROM scud_logs WHERE snapshot_id IN ({placeholders})", safe_ids)
|
|
deleted = cursor.rowcount
|
|
elif snapshot_id:
|
|
clean_id = str(snapshot_id).strip()
|
|
if clean_id.startswith("Y"):
|
|
conn.close()
|
|
return {"status": "error", "message": f"Итоговый срез [{clean_id}] защищен от удаления."}
|
|
cursor.execute("DELETE FROM scud_logs WHERE snapshot_id = ?", (clean_id,))
|
|
deleted = cursor.rowcount
|
|
elif day_str:
|
|
cursor.execute("DELETE FROM scud_logs WHERE (log_date = ? OR snapshot_id LIKE ?) AND snapshot_id NOT LIKE 'Y%'", (day_str, f"%{day_str.replace('.', '')}%"))
|
|
deleted = cursor.rowcount
|
|
else:
|
|
conn.close()
|
|
return {"status": "error", "message": "Не указаны идентификаторы для удаления."}
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
return {"status": "success", "deleted_records": deleted, "message": f"Успешно удалено записей: {deleted}"} |