""" =============================================================================== FILE: modules/ai_engine/handlers/snapshot_handler.py PROJECT: SCUD Orion AI (Unified Architecture) MODULE: modules / ai_engine / handlers ROLE: Изолированная обработка запросов просмотра и удаления срезов СКУД. AI-CONTEXT-ANCHORS: - ANCHOR[SNAPSHOT_HANDLER_DISPATCH]: Обработка db_get_snapshots и db_delete_snapshots. =============================================================================== """ from typing import Dict, Any, Tuple, Optional, List from services.snapshots.service import get_snapshots_registry, delete_snapshots_safely from modules.web_api.llm.core.calendar_utils import parse_relative_date_ru from modules.web_api.llm.db.db_chat import db_save_chat_message, db_get_chat_history from modules.web_api.llm.db.db_prompts import db_set_session_state # ANCHOR[SNAPSHOT_HANDLER_DISPATCH] def handle_snapshots_call( fn_name: str, fn_args: Dict[str, Any], session_id: str, user_message: str, state_data: Dict[str, Any] ) -> Tuple[str, list, Optional[Dict[str, Any]]]: """Обрабатывает запросы реестра и безопасного удаления срезов.""" # 1. Получение срезов if fn_name == "db_get_snapshots": date_param = fn_args.get("date_str") if not date_param and user_message: date_param = parse_relative_date_ru(user_message) snapshots_res = get_snapshots_registry(date_str=date_param) query_date = snapshots_res.get("query_date", "выбранную дату") db_set_session_state(session_id, "SNAPSHOTS_VIEW", snapshots_res) reply_text = f"Реестр срезов СКУД за {query_date}:" db_save_chat_message(session_id, "assistant", reply_text, is_ephemeral=1) return reply_text, db_get_chat_history(session_id), { "type": "SNAPSHOTS_CARD", "data": snapshots_res } # 2. Удаление срезов (двухфазное подтверждение) raw_id = fn_args.get("snapshot_id") or fn_args.get("day_str") raw_ids = fn_args.get("snapshot_ids") or [] is_confirmed = fn_args.get("confirmed", False) if raw_id and not raw_ids: if isinstance(raw_id, str) and "," in raw_id: raw_ids = [s.strip() for s in raw_id.split(",")] else: raw_ids = [raw_id] safe_ids = [s.strip() for s in raw_ids if s and not str(s).strip().startswith("Y")] if not safe_ids: reply_text = "⚠️ Итоговый срез Y защищен от удаления. Выберите дневные снапшоты." db_save_chat_message(session_id, "assistant", reply_text, is_ephemeral=1) return reply_text, db_get_chat_history(session_id), None if not is_confirmed: query_date = state_data.get("query_date", "") db_set_session_state(session_id, "SNAPSHOT_DELETE_CONFIRM", { "snapshot_ids": safe_ids, "query_date": query_date, "idle_turns": 0 }) ids_str = ", ".join(safe_ids) reply_text = f"Вы действительно хотите удалить дневные снапшоты: {ids_str}?" db_save_chat_message(session_id, "assistant", reply_text, is_ephemeral=1) return reply_text, db_get_chat_history(session_id), { "type": "SNAPSHOT_DELETE_CONFIRM", "buttons": [ {"label": f"Удалить ({len(safe_ids)} шт.)", "value": f"подтверждаю удаление снапшотов {ids_str}", "style": "danger"}, {"label": "Отмена", "value": "отмена", "style": "secondary"} ] } else: delete_snapshots_safely(snapshot_ids=safe_ids) query_date = state_data.get("query_date", "") updated_data = get_snapshots_registry(date_str=query_date) reply_text = f"✅ Успешно удалено снапшотов: {len(safe_ids)} шт." db_save_chat_message(session_id, "assistant", reply_text, is_ephemeral=1) return reply_text, db_get_chat_history(session_id), { "type": "SNAPSHOTS_CARD", "data": updated_data }