File size: 8,444 Bytes
f48b219 | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 | """
src/database/db.py
Fix #9 — Persistent query history per user
Fix #3 — URL page cache with TTL
Schema:
users → id, username, email, password_hash, created_at
queries → id, user_id, session_id, question, answer, context_preview,
confidence, intent, timestamp
url_cache → url_hash, url, text, cached_at
comparisons → id, user_id, urls_json, question, results_json, timestamp
"""
import sqlite3
import time
import logging
from pathlib import Path
from contextlib import contextmanager
logger = logging.getLogger(__name__)
DB_PATH = Path("data") / "ecom_qa.db"
CACHE_TTL_SECONDS = 60 * 60 * 6 # 6 hours
import hashlib
def _url_hash(url: str) -> str:
return hashlib.sha256(url.encode()).hexdigest()[:32]
class Database:
def __init__(self, path: Path = DB_PATH):
self.path = path
self.path.parent.mkdir(parents=True, exist_ok=True)
@contextmanager
def _conn(self):
conn = sqlite3.connect(str(self.path))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
# ── Schema ────────────────────────────────────────────────────────────────
def init_tables(self):
with self._conn() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT DEFAULT '',
password_hash TEXT NOT NULL,
created_at REAL DEFAULT (strftime('%s','now'))
);
CREATE TABLE IF NOT EXISTS queries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
session_id TEXT,
question TEXT NOT NULL,
answer TEXT DEFAULT '',
context_preview TEXT DEFAULT '',
confidence REAL,
intent TEXT DEFAULT 'factual',
timestamp REAL DEFAULT (strftime('%s','now'))
);
CREATE TABLE IF NOT EXISTS url_cache (
url_hash TEXT PRIMARY KEY,
url TEXT NOT NULL,
text TEXT NOT NULL,
cached_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS comparisons (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id) ON DELETE CASCADE,
urls_json TEXT NOT NULL,
question TEXT NOT NULL,
results_json TEXT,
timestamp REAL DEFAULT (strftime('%s','now'))
);
CREATE INDEX IF NOT EXISTS idx_queries_user ON queries(user_id);
CREATE INDEX IF NOT EXISTS idx_queries_session ON queries(session_id);
CREATE INDEX IF NOT EXISTS idx_cache_hash ON url_cache(url_hash);
""")
logger.info("Database initialised at %s", self.path)
# ── User CRUD ─────────────────────────────────────────────────────────────
def create_user(self, username: str, email: str, password_hash: str) -> int:
with self._conn() as conn:
cur = conn.execute(
"INSERT INTO users (username, email, password_hash) VALUES (?,?,?)",
(username, email, password_hash),
)
return cur.lastrowid
def get_user_by_username(self, username: str) -> dict | None:
with self._conn() as conn:
row = conn.execute(
"SELECT * FROM users WHERE username = ?", (username,)
).fetchone()
return dict(row) if row else None
def get_user_by_id(self, user_id: int) -> dict | None:
with self._conn() as conn:
row = conn.execute(
"SELECT * FROM users WHERE id = ?", (user_id,)
).fetchone()
return dict(row) if row else None
# ── Query history ─────────────────────────────────────────────────────────
def save_query(self, user_id, session_id, question, answer,
context_preview="", confidence=None, intent="factual"):
with self._conn() as conn:
conn.execute(
"""INSERT INTO queries
(user_id, session_id, question, answer, context_preview, confidence, intent)
VALUES (?,?,?,?,?,?,?)""",
(user_id, session_id, question, answer, context_preview, confidence, intent),
)
def get_user_history(self, user_id: int, limit: int = 200) -> list[dict]:
with self._conn() as conn:
rows = conn.execute(
"""SELECT id, question, answer, confidence, intent,
datetime(timestamp, 'unixepoch', 'localtime') AS timestamp
FROM queries WHERE user_id = ?
ORDER BY timestamp DESC LIMIT ?""",
(user_id, limit),
).fetchall()
return [dict(r) for r in rows]
def get_session_history(self, session_id: str, limit: int = 50) -> list[dict]:
with self._conn() as conn:
rows = conn.execute(
"""SELECT question, answer, confidence, intent,
datetime(timestamp, 'unixepoch', 'localtime') AS timestamp
FROM queries WHERE session_id = ?
ORDER BY timestamp DESC LIMIT ?""",
(session_id, limit),
).fetchall()
return [dict(r) for r in rows]
def delete_query(self, query_id: int, user_id: int):
with self._conn() as conn:
conn.execute(
"DELETE FROM queries WHERE id=? AND user_id=?", (query_id, user_id)
)
# ── URL cache ─────────────────────────────────────────────────────────────
def get_cached_page(self, url: str) -> str | None:
key = _url_hash(url)
now = time.time()
with self._conn() as conn:
row = conn.execute(
"SELECT text, cached_at FROM url_cache WHERE url_hash=?", (key,)
).fetchone()
if row and (now - row["cached_at"]) < CACHE_TTL_SECONDS:
return row["text"]
if row:
conn.execute("DELETE FROM url_cache WHERE url_hash=?", (key,))
return None
def cache_page(self, url: str, text: str):
key = _url_hash(url)
with self._conn() as conn:
conn.execute(
"""INSERT OR REPLACE INTO url_cache (url_hash, url, text, cached_at)
VALUES (?,?,?,?)""",
(key, url, text, time.time()),
)
def clear_expired_cache(self):
cutoff = time.time() - CACHE_TTL_SECONDS
with self._conn() as conn:
conn.execute("DELETE FROM url_cache WHERE cached_at < ?", (cutoff,))
# ── Comparisons ───────────────────────────────────────────────────────────
def save_comparison(self, user_id, urls_json, question, results_json):
with self._conn() as conn:
conn.execute(
"""INSERT INTO comparisons (user_id, urls_json, question, results_json)
VALUES (?,?,?,?)""",
(user_id, urls_json, question, results_json),
)
|