import sqlite3 import logging from typing import List, Dict, Any, Optional # Настройка логирования logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger("DB_TOOLS") DB_PATH = "/home/puh/scud_orion_ai_v2/data/scud_orion_ai.db" def get_db_connection(): # Увеличиваем таймаут до 30 секунд, чтобы соединения ожидали завершения соседних транзакций, # а также включаем WAL-режим для безопасного параллельного чтения и записи. conn = sqlite3.connect(DB_PATH, timeout=30.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode = WAL;") conn.execute("PRAGMA synchronous = NORMAL;") return conn def normalize_task_id(task_id_input: str) -> str: """Преобразует 6, '6', 'task-6' в красивый формат TASK-06 или TASK-006""" if not task_id_input: return "" clean_id = str(task_id_input).strip().upper().replace("TASK-", "").replace("TASK", "") if clean_id.isdigit(): num = int(clean_id) return f"TASK-{num:02d}" if num < 100 else f"TASK-{num:03d}" return f"TASK-{clean_id}" # === ЗАДАЧИ С ФИЛЬТРАЦИЕЙ ПО USER_ID === def db_get_tasks(user_id: int) -> List[Dict[str, Any]]: conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" SELECT id, task_id, module, title, priority, status, due_date, created_at FROM tasks WHERE user_id = ? ORDER BY id DESC """, (user_id,)) rows = cursor.fetchall() conn.close() return [dict(r) for r in rows] def db_add_task(user_id: int, module: str, title: str, priority: str = "MEDIUM", due_date: Optional[str] = None) -> Dict[str, Any]: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT MAX(id) FROM tasks") max_id = cursor.fetchone()[0] or 0 new_task_id = f"TASK-{(max_id + 1):02d}" cursor.execute(""" INSERT INTO tasks (task_id, module, title, priority, status, due_date, user_id) VALUES (?, ?, ?, ?, 'BACKLOG', ?, ?) """, (new_task_id, module, title, priority.upper(), due_date, user_id)) conn.commit() conn.close() return {"status": "success", "task_id": new_task_id, "message": f"Задача {new_task_id} создана"} def db_update_task_status(user_id: int, task_id: str, status: str = "COMPLETED", due_date: Optional[str] = None) -> Dict[str, Any]: conn = get_db_connection() cursor = conn.cursor() formatted_id = normalize_task_id(task_id) if due_date: cursor.execute(""" UPDATE tasks SET status = ?, due_date = ? WHERE (UPPER(task_id) = ? OR task_id LIKE ?) AND user_id = ? """, (status.upper(), due_date, formatted_id, f"%{task_id.strip()}", user_id)) else: cursor.execute(""" UPDATE tasks SET status = ? WHERE (UPPER(task_id) = ? OR task_id LIKE ?) AND user_id = ? """, (status.upper(), formatted_id, f"%{task_id.strip()}", user_id)) if cursor.rowcount == 0: conn.close() return {"error": f"Задача {task_id} не найдена или принадлежит другому пользователю"} conn.commit() conn.close() return {"status": "success", "message": f"Статус задачи {formatted_id} обновлен на {status.upper()}"} def db_delete_task(user_id: int, task_id: str) -> Dict[str, Any]: conn = get_db_connection() cursor = conn.cursor() formatted_id = normalize_task_id(task_id) cursor.execute(""" DELETE FROM tasks WHERE (UPPER(task_id) = ? OR task_id LIKE ?) AND user_id = ? """, (formatted_id, f"%{task_id.strip()}", user_id)) if cursor.rowcount == 0: conn.close() return {"error": f"Задача {task_id} не найдена"} conn.commit() conn.close() return {"status": "success", "message": f"Задача {formatted_id} удалена"} # === ОБЩИЕ СИСТЕМНЫЕ РЕСУРСЫ === def db_get_active_system_prompt() -> str: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT prompt_text FROM system_prompts WHERE is_active = 1 ORDER BY id DESC LIMIT 1") row = cursor.fetchone() conn.close() return row["prompt_text"] if row else "Ты — ИИ-ассистент SCUD Orion AI." def db_add_system_prompt(name: str, prompt_text: str) -> Dict[str, Any]: try: with get_db_connection() as conn: cursor = conn.cursor() cursor.execute("BEGIN IMMEDIATE;") # Проверяем, существует ли уже промпт с таким именем cursor.execute("SELECT id FROM system_prompts WHERE name = ?", (name,)) existing = cursor.fetchone() if existing: # Обновляем существующий активный промпт cursor.execute( "UPDATE system_prompts SET prompt_text = ?, is_active = 1, updated_at = CURRENT_TIMESTAMP WHERE name = ?", (prompt_text, name) ) else: # Вставляем новый, если его не было cursor.execute( "INSERT INTO system_prompts (name, prompt_text, is_active) VALUES (?, ?, 1)", (name, prompt_text) ) conn.commit() logger.info("Системный промпт успешно сохранен и применен в базе данных.") return {"status": "success", "message": "Системный промпт успешно обновлен"} except Exception as e: logger.error(f"Ошибка при сохранении промпта в БД: {e}") return {"status": "error", "error": str(e)} def db_get_rules() -> List[Dict[str, Any]]: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT id, rule_text FROM ai_knowledge_base ORDER BY id ASC") rows = cursor.fetchall() conn.close() return [dict(r) for r in rows] def db_set_session_state(session_id: str, state_type: str, data: str): conn = get_db_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO session_states (session_id, state_type, pending_data, updated_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(session_id) DO UPDATE SET state_type = excluded.state_type, pending_data = excluded.pending_data, updated_at = CURRENT_TIMESTAMP """, (session_id, state_type, data)) conn.commit() conn.close() def db_get_session_state(session_id: str) -> Optional[Dict[str, Any]]: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT state_type, pending_data FROM session_states WHERE session_id = ?", (session_id,)) row = cursor.fetchone() conn.close() return dict(row) if row else None def db_clear_session_state(session_id: str): conn = get_db_connection() cursor = conn.cursor() cursor.execute("DELETE FROM session_states WHERE session_id = ?", (session_id,)) conn.commit() conn.close()