| import os |
| import sqlite3 |
| import logging |
| from pathlib import Path |
|
|
| log = logging.getLogger("db") |
|
|
| DB_PATH = os.environ.get("DB_PATH", "/data/bot.db") |
|
|
|
|
| def _ensure_dir(): |
| p = Path(DB_PATH).parent |
| p.mkdir(parents=True, exist_ok=True) |
|
|
|
|
| def get_conn(): |
| _ensure_dir() |
| conn = sqlite3.connect(DB_PATH, timeout=10, check_same_thread=False) |
| conn.row_factory = sqlite3.Row |
| conn.execute("PRAGMA journal_mode=WAL;") |
| conn.execute("PRAGMA foreign_keys=ON;") |
| return conn |
|
|
|
|
| def init_db(): |
| _ensure_dir() |
| schema_path = Path(__file__).parent / "schema.sql" |
| with open(schema_path, "r", encoding="utf-8") as f: |
| schema = f.read() |
| with get_conn() as conn: |
| conn.executescript(schema) |
| |
| with get_conn() as conn: |
| conn.execute( |
| "INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", |
| ("conversation_window", "6"), |
| ) |
| conn.execute( |
| "INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", |
| ("max_new_tokens", "1024"), |
| ) |
| conn.execute( |
| "INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", |
| ("temperature", "0.7"), |
| ) |
| conn.execute( |
| "INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", |
| ("top_p", "0.9"), |
| ) |
| conn.execute( |
| "INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", |
| ("system_prompt", "You are a helpful assistant."), |
| ) |
| conn.execute( |
| "INSERT OR IGNORE INTO settings(key, value) VALUES (?, ?)", |
| ("daily_turn_limit", "100"), |
| ) |
| log.info(f"DB initialised at {DB_PATH}") |
