chore: save working baseline before v3.0 architecture refactoring

modules/web_api/llm/db/db_snapshots.py

@@ -1,49 +1,112 @@
 """
+===============================================================================
 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
-from typing import Dict, Any, Optional
+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 "
+    
+    query = """
+        SELECT 
+            snapshot_id, 
+            log_date, 
+            snapshot_time, 
+            COUNT(*) as record_count 
+        FROM scud_logs 
+    """
     params = []
-    if date_str:
-        iso_date = date_str
-        if "." in date_str:
-            parts = date_str.split(".")
+    
+    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]}"
-        query += " WHERE log_date = ? OR log_date = ? OR snapshot_time LIKE ? "
-        params.extend([date_str, iso_date, f"{iso_date}%"])
+                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 ORDER BY id DESC LIMIT 50"
+    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": date_str or "все",
+        "query_date": clean_date or "все",
         "snapshots_count": len(snapshots),
         "snapshots": snapshots
     }
-    db_set_session_state(session_id=session_id, state_type="SNAPSHOTS_VIEW", data=json.dumps(result_data, ensure_ascii=False))
+    db_set_session_state(session_id=session_id, state_type="SNAPSHOTS_VIEW", data=result_data)
     conn.close()
     return result_data
 
-def db_delete_snapshots(snapshot_id: Optional[str] = None, day_str: Optional[str] = None) -> Dict[str, Any]:
-    if not snapshot_id and not day_str:
-        return {"status": "error", "message": "Необходимо указать snapshot_id или day_str (ДД.ММ.ГГГГ)."}
+# 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()
-    if snapshot_id:
-        cursor.execute("DELETE FROM scud_logs WHERE snapshot_id = ?", (snapshot_id,))
+    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:
-        cursor.execute("DELETE FROM scud_logs WHERE log_date = ? OR snapshot_id LIKE ?", (day_str, f"%{day_str.replace('.', '')}%"))
-    deleted = cursor.rowcount
+        conn.close()
+        return {"status": "error", "message": "Не указаны идентификаторы для удаления."}
+
     conn.commit()
     conn.close()
-    return {"status": "success", "message": f"Успешно удалено записей: {deleted}"}
+    return {"status": "success", "deleted_records": deleted, "message": f"Успешно удалено записей: {deleted}"}