Spaces:
Running
Running
File size: 2,686 Bytes
0984c49 f70ee56 ee25239 f70ee56 0984c49 f70ee56 0984c49 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 ee25239 f70ee56 |
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 95 96 97 |
"""Database for conversations"""
import sqlite3
from typing import List, Dict
from config import DATABASE_PATH
class VedaDatabase:
"""Database handler"""
def __init__(self):
self._init_db()
def _get_conn(self):
conn = sqlite3.connect(DATABASE_PATH)
conn.row_factory = sqlite3.Row
return conn
def _init_db(self):
conn = self._get_conn()
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS conversations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
user_input TEXT NOT NULL,
assistant_response TEXT NOT NULL,
feedback INTEGER DEFAULT 0
)
''')
conn.commit()
conn.close()
def save_conversation(self, user_input: str, response: str) -> int:
conn = self._get_conn()
cursor = conn.cursor()
cursor.execute('''
INSERT INTO conversations (user_input, assistant_response)
VALUES (?, ?)
''', (user_input, response))
conv_id = cursor.lastrowid
conn.commit()
conn.close()
return conv_id
def update_feedback(self, conv_id: int, feedback: int):
conn = self._get_conn()
cursor = conn.cursor()
cursor.execute('''
UPDATE conversations SET feedback = ? WHERE id = ?
''', (feedback, conv_id))
conn.commit()
conn.close()
def get_good_conversations(self, limit: int = 100) -> List[Dict]:
conn = self._get_conn()
cursor = conn.cursor()
cursor.execute('''
SELECT user_input, assistant_response
FROM conversations
WHERE feedback > 0
ORDER BY timestamp DESC
LIMIT ?
''', (limit,))
rows = cursor.fetchall()
conn.close()
return [dict(row) for row in rows]
def get_stats(self) -> Dict:
conn = self._get_conn()
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM conversations')
total = cursor.fetchone()[0]
cursor.execute('SELECT COUNT(*) FROM conversations WHERE feedback > 0')
positive = cursor.fetchone()[0]
cursor.execute('SELECT COUNT(*) FROM conversations WHERE feedback < 0')
negative = cursor.fetchone()[0]
conn.close()
return {'total': total, 'positive': positive, 'negative': negative}
db = VedaDatabase() |