| |
| |
| |
|
|
| import sqlite3 |
| import json |
| import datetime |
| import os |
| from config import DB_PATH, MEMORY_DIR |
|
|
| os.makedirs(MEMORY_DIR, exist_ok=True) |
|
|
|
|
| def _connect(): |
| return sqlite3.connect(DB_PATH) |
|
|
|
|
| def init_db(): |
| """Create tables if they don't exist.""" |
| with _connect() as conn: |
| conn.executescript(""" |
| CREATE TABLE IF NOT EXISTS conversations ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| role TEXT NOT NULL, -- 'user' or 'jarvis' |
| message TEXT NOT NULL, |
| timestamp TEXT NOT NULL |
| ); |
| |
| CREATE TABLE IF NOT EXISTS facts ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| key TEXT NOT NULL UNIQUE, -- e.g. 'user_name' |
| value TEXT NOT NULL, |
| updated TEXT NOT NULL |
| ); |
| |
| CREATE TABLE IF NOT EXISTS preferences ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| category TEXT NOT NULL, |
| setting TEXT NOT NULL, |
| value TEXT NOT NULL |
| ); |
| """) |
| print("[Memory] Database ready.") |
|
|
|
|
| def save_message(role: str, message: str): |
| """Save a conversation turn.""" |
| ts = datetime.datetime.now().isoformat() |
| with _connect() as conn: |
| conn.execute( |
| "INSERT INTO conversations (role, message, timestamp) VALUES (?,?,?)", |
| (role, message, ts) |
| ) |
|
|
|
|
| def get_recent_history(n: int = 10) -> list[dict]: |
| """Return the last n conversation turns.""" |
| with _connect() as conn: |
| rows = conn.execute( |
| "SELECT role, message, timestamp FROM conversations ORDER BY id DESC LIMIT ?", |
| (n,) |
| ).fetchall() |
| return [{"role": r[0], "message": r[1], "timestamp": r[2]} for r in reversed(rows)] |
|
|
|
|
| def save_fact(key: str, value: str): |
| """Save or update a fact about the user.""" |
| ts = datetime.datetime.now().isoformat() |
| with _connect() as conn: |
| conn.execute( |
| "INSERT INTO facts (key, value, updated) VALUES (?,?,?) " |
| "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated=excluded.updated", |
| (key, value, ts) |
| ) |
|
|
|
|
| def get_fact(key: str) -> str | None: |
| """Retrieve a stored fact.""" |
| with _connect() as conn: |
| row = conn.execute( |
| "SELECT value FROM facts WHERE key=?", (key,) |
| ).fetchone() |
| return row[0] if row else None |
|
|
|
|
| def get_all_facts() -> dict: |
| """Return all stored facts as a dict.""" |
| with _connect() as conn: |
| rows = conn.execute("SELECT key, value FROM facts").fetchall() |
| return {r[0]: r[1] for r in rows} |
|
|
|
|
| def search_history(query: str, limit: int = 5) -> list[dict]: |
| """Simple keyword search through conversation history.""" |
| with _connect() as conn: |
| rows = conn.execute( |
| "SELECT role, message, timestamp FROM conversations " |
| "WHERE message LIKE ? ORDER BY id DESC LIMIT ?", |
| (f"%{query}%", limit) |
| ).fetchall() |
| return [{"role": r[0], "message": r[1], "timestamp": r[2]} for r in rows] |
|
|
