Repository navigation
Expand file tree
/
Copy pathdatabase.py
More file actions
70 lines (56 loc) · 1.8 KB
/
Copy pathdatabase.py
File metadata and controls
70 lines (56 loc) · 1.8 KB
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
import sqlite3
import json
from datetime import datetime
def init_db():
connection = sqlite3.connect("chat_history.db")
cursor = connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT,
role TEXT,
content TEXT,
timestamp TEXT
)
""")
connection.commit()
connection.close()
def save_message(user_id, role, content):
connection = sqlite3.connect("chat_history.db")
cursor = connection.cursor()
try:
content_json = json.dumps(content)
timestamp = datetime.now().isoformat()
cursor.execute(
"INSERT INTO history (user_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
(user_id, role, content_json, timestamp)
)
connection.commit()
return True, None
except (sqlite3.Error, TypeError, ValueError) as e:
return False, f"Failed to save the message for {user_id}: {e}"
finally:
connection.close()
def load_history(user_id):
connection = sqlite3.connect("chat_history.db")
cursor = connection.cursor()
try:
cursor.execute(
"SELECT * FROM history WHERE user_id = ? ORDER BY timestamp",
(user_id,)
)
rows = cursor.fetchall()
reconstructed = []
for row in rows:
temp_dict = {}
temp_dict['role'] = row[2]
content = json.loads(row[3])
temp_dict['parts'] = content
reconstructed.append(temp_dict)
return reconstructed
except sqlite3.Error as e:
print(f"Database error occurred: {e}")
return []
finally:
# No need to commit anything as no change has been made
connection.close()