""" =============================================================================== FILE: modules/web_api/llm/db_tools.py PROJECT: SCUD Orion AI (Unified Architecture) MODULE: web_api / llm ROLE: Фасадная точка доступа к доменным сервисам (Domain Facade). AI-CONTEXT-ANCHORS: - ANCHOR[FACADE_EXPORTS]: Экспорт методов предметных сервисов для LLM и API. =============================================================================== """ # ANCHOR[FACADE_EXPORTS] from typing import Optional, List, Dict, Any from core.connection import DB_PATH, get_connection as get_db_connection # Домен: Задачи from services.tasks.service import ( get_tasks as db_get_tasks, add_task as db_add_task, update_task_details as db_update_task_details, delete_task as db_delete_task, execute_task_action as db_tasks_edit ) # Защитный алиас для обратной совместимости db_update_task_status = db_update_task_details from services.tasks.exporter import export_tasks_to_markdown as db_export_tasks_markdown from services.tasks.repository import normalize_task_id # Домен: Системный промпт from services.prompts.service import ( get_active_system_prompt as db_get_active_system_prompt, apply_prompt_action as db_apply_prompt_node_action, save_full_prompt_draft, create_prompt_preview ) # Домен: База знаний from services.knowledge.service import ( get_rules as db_get_rules, add_rule as db_add_rule ) # Чат, сессии, статистика from .db.db_chat import ( db_save_chat_message, db_get_chat_history, db_purge_ephemeral_messages, db_clear_chat_history ) from .db.db_prompts import ( db_get_tool_action, db_set_session_state, db_get_session_state, db_clear_session_state, db_get_stats, db_get_anomalies, db_get_reference ) from .core.calendar_utils import get_dynamic_calendar_context as db_get_current_server_time def db_add_system_prompt(name_or_text: str, draft_text: str = None) -> None: """Совместимая обертка для сохранения системного промпта.""" if draft_text is not None: save_full_prompt_draft(draft_text, prompt_name=name_or_text) else: save_full_prompt_draft(name_or_text, prompt_name="main_agent") def db_get_snapshots(session_id: str = "web_session_main", date_str: str = None, original_user_message: str = "") -> Dict[str, Any]: """Совместимый фасад выборки снапшотов с сохранением стейта сессии.""" from services.snapshots.service import get_snapshots_registry from .core.calendar_utils import parse_relative_date_ru clean_date = (date_str or "").strip() if not clean_date and original_user_message: clean_date = parse_relative_date_ru(original_user_message) res = get_snapshots_registry(date_str=clean_date if clean_date else None) db_set_session_state(session_id=session_id, state_type="SNAPSHOTS_VIEW", data=res) return res def db_delete_snapshots(snapshot_id: str = None, snapshot_ids: List[str] = None, day_str: str = None) -> Dict[str, Any]: """Совместимый фасад безопасного удаления снапшотов.""" from services.snapshots.service import delete_snapshots_safely target_ids = [] if snapshot_ids: target_ids.extend(snapshot_ids) if snapshot_id: if isinstance(snapshot_id, str) and "," in snapshot_id: target_ids.extend([s.strip() for s in snapshot_id.split(",")]) else: target_ids.append(snapshot_id) return delete_snapshots_safely(snapshot_ids=target_ids)