File size: 3,966 Bytes
0bc4913 | 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 | """tutor/progress_store.py — SQLite-backed learner progress store."""
from __future__ import annotations
import json, sqlite3, time, uuid
from pathlib import Path
from typing import Any, Dict, Optional
_SCHEMA = """
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS learners (
learner_id TEXT PRIMARY KEY,
display_name TEXT,
created_at REAL
);
CREATE TABLE IF NOT EXISTS sessions (
session_id TEXT PRIMARY KEY,
learner_id TEXT,
started_at REAL,
ended_at REAL,
state_json TEXT,
lang TEXT
);
CREATE TABLE IF NOT EXISTS responses (
response_id TEXT PRIMARY KEY,
learner_id TEXT,
session_id TEXT,
item_id TEXT,
skill TEXT,
difficulty REAL,
is_correct INTEGER,
latency_ms INTEGER,
ts REAL
);
"""
class ProgressStore:
def __init__(self, db_path=":memory:"):
self._path = str(db_path)
self._conn = sqlite3.connect(self._path, check_same_thread=False)
self._conn.executescript(_SCHEMA)
self._conn.commit()
# ── Learners ──────────────────────────────────────────────────────────
def add_learner(self, learner_id: str, display_name: str = "") -> None:
self._conn.execute(
"INSERT OR IGNORE INTO learners VALUES (?,?,?)",
(learner_id, display_name or learner_id, time.time()))
self._conn.commit()
# ── Sessions ──────────────────────────────────────────────────────────
def start_session(self, learner_id: str, state: Dict, lang: str = "en") -> str:
sid = str(uuid.uuid4())
self._conn.execute(
"INSERT INTO sessions VALUES (?,?,?,?,?,?)",
(sid, learner_id, time.time(), None, json.dumps(state), lang))
self._conn.commit()
return sid
def end_session(self, session_id: str, state: Dict) -> None:
self._conn.execute(
"UPDATE sessions SET ended_at=?, state_json=? WHERE session_id=?",
(time.time(), json.dumps(state), session_id))
self._conn.commit()
def load_latest_state(self, learner_id: str) -> Optional[Dict]:
row = self._conn.execute(
"SELECT state_json FROM sessions WHERE learner_id=? "
"ORDER BY started_at DESC LIMIT 1", (learner_id,)).fetchone()
if row and row[0]:
try:
return json.loads(row[0])
except Exception:
pass
return None
# ── Responses ─────────────────────────────────────────────────────────
def log_response(self, learner_id: str, session_id: str, item_id: str,
skill: str, difficulty: float, is_correct: bool,
latency_ms: int) -> None:
self._conn.execute(
"INSERT INTO responses VALUES (?,?,?,?,?,?,?,?,?)",
(str(uuid.uuid4()), learner_id, session_id, item_id,
skill, difficulty, int(is_correct), latency_ms, time.time()))
self._conn.commit()
# ── Reports ───────────────────────────────────────────────────────────
def skill_summary(self, learner_id: str) -> Dict[str, Any]:
rows = self._conn.execute(
"SELECT skill, COUNT(*) as n, SUM(is_correct) as c "
"FROM responses WHERE learner_id=? GROUP BY skill",
(learner_id,)).fetchall()
return {r[0]: {"attempts": r[1], "correct": r[2],
"accuracy": round(r[2]/r[1], 3) if r[1] else 0}
for r in rows}
|