Spaces:
Sleeping
Sleeping
File size: 2,719 Bytes
386ee45 65cfd5e 386ee45 65cfd5e 386ee45 65cfd5e 386ee45 65cfd5e 968f79d 65cfd5e 968f79d 65cfd5e 386ee45 968f79d 386ee45 968f79d 65cfd5e 968f79d 65cfd5e 968f79d 65cfd5e 386ee45 968f79d 65cfd5e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
import os
import sqlite3
import logging
from typing import List, Tuple, Optional
logger = logging.getLogger(__name__)
DB_FOLDER = "db"
os.makedirs(DB_FOLDER, exist_ok=True)
DB_PATH = os.path.join(DB_FOLDER, "data.db")
_conn: Optional[sqlite3.Connection] = None
def get_db_connection() -> Optional[sqlite3.Connection]:
global _conn
if _conn:
return _conn
try:
conn = sqlite3.connect(DB_PATH, check_same_thread=False, timeout=20)
conn.row_factory = sqlite3.Row
_conn = conn
return conn
except sqlite3.Error as e:
logger.error("Database connection error: %s", e)
return None
def init_db() -> None:
conn = get_db_connection()
if not conn:
raise RuntimeError("Could not obtain database connection")
try:
cur = conn.cursor()
cur.execute("PRAGMA journal_mode=WAL")
cur.execute("""
CREATE TABLE IF NOT EXISTS history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
text TEXT NOT NULL,
sentiment TEXT NOT NULL,
confidence REAL NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
cur.execute("CREATE INDEX IF NOT EXISTS idx_user_id ON history(user_id)")
cur.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON history(timestamp)")
conn.commit()
logger.info("Database initialized")
except sqlite3.Error as e:
logger.exception("Database initialization error: %s", e)
raise
def save_message(user_id: str, text: str, sentiment: str, confidence: float) -> bool:
conn = get_db_connection()
if not conn:
return False
try:
cur = conn.cursor()
cur.execute(
"INSERT INTO history (user_id, text, sentiment, confidence) VALUES (?, ?, ?, ?)",
(user_id, text, sentiment, confidence),
)
conn.commit()
return True
except sqlite3.Error as e:
logger.exception("Error saving message: %s", e)
try:
conn.rollback()
except Exception:
pass
return False
def get_recent(user_id: str, limit: int = 10) -> List[Tuple]:
conn = get_db_connection()
if not conn:
return []
try:
cur = conn.cursor()
cur.execute("""
SELECT text, sentiment, confidence, timestamp
FROM history
WHERE user_id = ?
ORDER BY timestamp DESC
LIMIT ?
""", (user_id, limit))
rows = cur.fetchall()
return [tuple(r) for r in rows]
except sqlite3.Error as e:
logger.exception("Error fetching recent messages: %s", e)
return []
|