118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
import sqlite3
|
|
import time
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any, Optional
|
|
from app.config import DB_PATH, DATA_DIR
|
|
|
|
def init_db():
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
with sqlite3.connect(DB_PATH) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS hub_conversations (
|
|
universe_id TEXT NOT NULL,
|
|
session_key TEXT NOT NULL,
|
|
title TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
pinned INTEGER NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (universe_id, session_key)
|
|
)
|
|
""")
|
|
cursor.execute("""
|
|
CREATE INDEX IF NOT EXISTS idx_hub_conv_universe
|
|
ON hub_conversations(universe_id, pinned DESC, updated_at DESC)
|
|
""")
|
|
conn.commit()
|
|
|
|
def list_conversations(universe_id: str) -> List[Dict[str, Any]]:
|
|
init_db()
|
|
with sqlite3.connect(DB_PATH) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT universe_id, session_key, title, created_at, updated_at, pinned
|
|
FROM hub_conversations
|
|
WHERE universe_id = ?
|
|
ORDER BY pinned DESC, updated_at DESC
|
|
""", (universe_id,))
|
|
rows = cursor.fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def get_conversation(universe_id: str, session_key: str) -> Optional[Dict[str, Any]]:
|
|
init_db()
|
|
with sqlite3.connect(DB_PATH) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT universe_id, session_key, title, created_at, updated_at, pinned
|
|
FROM hub_conversations
|
|
WHERE universe_id = ? AND session_key = ?
|
|
""", (universe_id, session_key))
|
|
row = cursor.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def save_conversation(
|
|
universe_id: str,
|
|
session_key: str,
|
|
title: Optional[str] = None,
|
|
pinned: Optional[int] = None,
|
|
created_at: Optional[int] = None,
|
|
updated_at: Optional[int] = None
|
|
) -> Dict[str, Any]:
|
|
init_db()
|
|
now = int(time.time() * 1000)
|
|
c_at = created_at or now
|
|
u_at = updated_at or now
|
|
p_val = 1 if pinned else 0
|
|
t_val = title or "Nouvelle conversation"
|
|
|
|
with sqlite3.connect(DB_PATH) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO hub_conversations (universe_id, session_key, title, created_at, updated_at, pinned)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(universe_id, session_key) DO UPDATE SET
|
|
title = COALESCE(excluded.title, hub_conversations.title),
|
|
pinned = COALESCE(excluded.pinned, hub_conversations.pinned),
|
|
updated_at = excluded.updated_at
|
|
""", (universe_id, session_key, t_val, c_at, u_at, p_val))
|
|
conn.commit()
|
|
return get_conversation(universe_id, session_key)
|
|
|
|
def update_conversation(
|
|
universe_id: str,
|
|
session_key: str,
|
|
title: Optional[str] = None,
|
|
pinned: Optional[bool] = None
|
|
) -> Optional[Dict[str, Any]]:
|
|
init_db()
|
|
existing = get_conversation(universe_id, session_key)
|
|
if not existing:
|
|
return None
|
|
|
|
now = int(time.time() * 1000)
|
|
new_title = title if title is not None else existing["title"]
|
|
new_pinned = (1 if pinned else 0) if pinned is not None else existing["pinned"]
|
|
|
|
with sqlite3.connect(DB_PATH) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
UPDATE hub_conversations
|
|
SET title = ?, pinned = ?, updated_at = ?
|
|
WHERE universe_id = ? AND session_key = ?
|
|
""", (new_title, new_pinned, now, universe_id, session_key))
|
|
conn.commit()
|
|
return get_conversation(universe_id, session_key)
|
|
|
|
def delete_conversation_record(universe_id: str, session_key: str) -> bool:
|
|
init_db()
|
|
with sqlite3.connect(DB_PATH) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
DELETE FROM hub_conversations
|
|
WHERE universe_id = ? AND session_key = ?
|
|
""", (universe_id, session_key))
|
|
conn.commit()
|
|
return cursor.rowcount > 0
|