239 lines
9.2 KiB
Python
239 lines
9.2 KiB
Python
"""
|
|
FILE: modules/web_api/llm/db/db_prompts.py
|
|
PROJECT: SCUD Orion AI
|
|
MODULE: web_api / llm / db
|
|
ROLE: Управление системными промптами, правилами базы знаний,
|
|
декларативным реестром действий инструментов (tool_action_registry)
|
|
и расширенным сессионным состоянием (session_states + idle_turns).
|
|
"""
|
|
import json
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from .connection import get_db_connection
|
|
|
|
logger = logging.getLogger("DB_PROMPTS")
|
|
|
|
|
|
def db_get_active_system_prompt() -> str:
|
|
"""Извлекает актуальный активный системный промпт из SQLite."""
|
|
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_get_tool_action(tool_name: str) -> Optional[Dict[str, Any]]:
|
|
"""Получение шаблона, кнопок и настроек эфемерности инструмента из SQLite."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT tool_name, category, bypass_llm, success_template,
|
|
follow_up_question, action_type, buttons_json, is_ephemeral
|
|
FROM tool_action_registry
|
|
WHERE tool_name = ? AND is_active = 1
|
|
""", (tool_name,))
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if row:
|
|
res = dict(row)
|
|
res["buttons"] = json.loads(res["buttons_json"]) if res.get("buttons_json") else []
|
|
return res
|
|
return None
|
|
|
|
|
|
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()
|
|
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]]:
|
|
"""Извлекает список глобальных правил компании из ai_knowledge_base."""
|
|
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]
|
|
|
|
|
|
# =============================================================================
|
|
# РАБОТА С СЕССИОННЫМИ СОСТОЯНИЯМИ, ЧЕРНОВИКАМИ И СЧЕТЧИКОМ ОТВЛЕЧЕНИЙ (IDLE_TURNS)
|
|
# =============================================================================
|
|
|
|
def db_set_session_state(session_id: str, state_type: str, data: Any) -> None:
|
|
"""
|
|
Сохраняет состояние сессии в SQLite.
|
|
Если передан dict/list — автоматически сериализует его в JSON-строку.
|
|
"""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
if isinstance(data, (dict, list)):
|
|
payload_str = json.dumps(data, ensure_ascii=False)
|
|
else:
|
|
payload_str = str(data) if data is not None else ""
|
|
|
|
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, payload_str))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def db_get_session_state(session_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Извлекает состояние сессии из SQLite.
|
|
Если в pending_data лежит валидный JSON-объект — парсит его в data_json.
|
|
"""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT session_id, state_type, pending_data, updated_at FROM session_states WHERE session_id = ?",
|
|
(session_id,)
|
|
)
|
|
row = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
res = dict(row)
|
|
raw_data = res.get("pending_data") or ""
|
|
|
|
try:
|
|
if raw_data.strip().startswith("{") or raw_data.strip().startswith("["):
|
|
res["data_json"] = json.loads(raw_data)
|
|
else:
|
|
res["data_json"] = None
|
|
except Exception:
|
|
res["data_json"] = None
|
|
|
|
return res
|
|
|
|
|
|
def db_increment_session_idle(session_id: str) -> int:
|
|
"""
|
|
Инкрементирует счетчик idle_turns для активного черновика сессии.
|
|
Возвращает обновленное значение счетчика отвлечений.
|
|
"""
|
|
state = db_get_session_state(session_id)
|
|
if not state:
|
|
return 0
|
|
|
|
data_meta = state.get("data_json")
|
|
if isinstance(data_meta, dict):
|
|
current_turns = data_meta.get("idle_turns", 0) + 1
|
|
data_meta["idle_turns"] = current_turns
|
|
db_set_session_state(session_id, state["state_type"], data_meta)
|
|
return current_turns
|
|
else:
|
|
# Если ранее данные были сохранены обычной строкой
|
|
new_meta = {
|
|
"draft_text": state.get("pending_data", ""),
|
|
"idle_turns": 1
|
|
}
|
|
db_set_session_state(session_id, state["state_type"], new_meta)
|
|
return 1
|
|
|
|
|
|
def db_clear_session_state(session_id: str) -> None:
|
|
"""Сбрасывает и удаляет активное состояние сессии."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM session_states WHERE session_id = ?", (session_id,))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def db_get_session_states() -> Dict[str, Any]:
|
|
"""Возвращает реестр всех активных сессий."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT session_id, state_type, updated_at FROM session_states")
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
return {"status": "success", "active_sessions": [dict(r) for r in rows]}
|
|
|
|
|
|
# =============================================================================
|
|
# СЛУЖЕБНАЯ ДИАГНОСТИКА И СТАТИСТИКА
|
|
# =============================================================================
|
|
|
|
def db_get_stats() -> Dict[str, Any]:
|
|
"""Возвращает статистику по количеству записей в таблицах."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
tables = [
|
|
'scud_logs', 'zup_staff', 'zup_absences', 'anomalies_history',
|
|
'ai_knowledge_base', 'system_prompts', 'session_states', 'tasks'
|
|
]
|
|
stats = {}
|
|
for t in tables:
|
|
try:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {t}")
|
|
stats[t] = cursor.fetchone()[0]
|
|
except Exception:
|
|
stats[t] = 0
|
|
conn.close()
|
|
return {"status": "success", "tables_stats": stats}
|
|
|
|
|
|
def db_get_anomalies(limit: int = 100, date_str: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Извлекает журнал зафиксированных аномалий."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
query = "SELECT anomaly_date, fio, anomaly_type, details FROM anomalies_history"
|
|
params = []
|
|
if date_str:
|
|
query += " WHERE anomaly_date = ?"
|
|
params.append(date_str)
|
|
query += " ORDER BY id DESC LIMIT ?"
|
|
params.append(limit)
|
|
cursor.execute(query, params)
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
return {"status": "success", "count": len(rows), "anomalies": [dict(r) for r in rows]}
|
|
|
|
|
|
def db_get_reference(category: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Извлекает справочные команды и примеры подсказок."""
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
query = "SELECT category, title, example_prompt, description FROM system_reference"
|
|
params = []
|
|
if category:
|
|
query += " WHERE category = ?"
|
|
params.append(category)
|
|
query += " ORDER BY id ASC"
|
|
cursor.execute(query, params)
|
|
rows = cursor.fetchall()
|
|
conn.close()
|
|
return {"status": "success", "count": len(rows), "reference_items": [dict(r) for r in rows]} |