91 lines
4.2 KiB
Python
91 lines
4.2 KiB
Python
"""
|
|
===============================================================================
|
|
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
|
|
} |