scud_ai/modules/web_api/llm/db/db_prompts.py

"""
FILE: modules/web_api/llm/db/db_prompts.py
"""
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:
    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]]:
    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()

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]}