| """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() |
|
|
| |
|
|
| 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() |
|
|
| |
|
|
| 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 |
|
|
| |
|
|
| 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() |
|
|
| |
|
|
| 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} |
|
|
