137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
import sqlite3
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
DB_PATH = "/home/puh/scud_context_api/scud_orion_ai.db"
|
|
|
|
def get_db_connection():
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
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]:
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Проверяем наличие промпта с таким именем, чтобы не плодить мусор
|
|
cursor.execute("SELECT id FROM system_prompts WHERE name = ?", (name,))
|
|
existing = cursor.fetchone()
|
|
|
|
if existing:
|
|
cursor.execute("""
|
|
UPDATE system_prompts
|
|
SET prompt_text = ?, updated_at = CURRENT_TIMESTAMP, is_active = 1
|
|
WHERE name = ?
|
|
""", (prompt_text, name))
|
|
else:
|
|
cursor.execute("UPDATE system_prompts SET is_active = 0")
|
|
cursor.execute("INSERT INTO system_prompts (name, prompt_text, is_active) VALUES (?, ?, 1)", (name, prompt_text))
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
return {"status": "success", "message": "Системный промпт успешно обновлен"}
|
|
|
|
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] |