database.py (29285B)
1 """Database module for multi-user stenography practice app.""" 2 3 import sqlite3 4 from contextlib import contextmanager 5 from datetime import datetime, timedelta 6 7 from spaced_repetition import calculate_review 8 9 DB_PATH = "/mnt/M2_1/dev/plover-practice/plover_practice.db" 10 11 12 class Database: 13 def __init__(self, db_path=None): 14 self.db_path = db_path or DB_PATH 15 self._init_db() 16 17 @contextmanager 18 def _get_conn(self): 19 conn = sqlite3.connect(self.db_path, check_same_thread=False) 20 conn.row_factory = sqlite3.Row 21 conn.execute("PRAGMA journal_mode=WAL") 22 conn.execute("PRAGMA foreign_keys=ON") 23 try: 24 yield conn 25 conn.commit() 26 except Exception: 27 conn.rollback() 28 raise 29 finally: 30 conn.close() 31 32 def _init_db(self): 33 with self._get_conn() as conn: 34 tables = {r[0] for r in conn.execute( 35 "SELECT name FROM sqlite_master WHERE type='table'" 36 ).fetchall()} 37 38 if "sessions" not in tables: 39 self._create_tables(conn) 40 return 41 42 session_cols = {r[1] for r in conn.execute("PRAGMA table_info(sessions)").fetchall()} 43 if "user_id" not in session_cols: 44 self._migrate_to_v2(conn) 45 else: 46 self._create_tables(conn) 47 48 def _create_tables(self, conn): 49 conn.executescript(""" 50 CREATE TABLE IF NOT EXISTS users ( 51 id INTEGER PRIMARY KEY AUTOINCREMENT, 52 email TEXT UNIQUE NOT NULL, 53 name TEXT, 54 avatar_url TEXT, 55 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 56 ); 57 58 CREATE TABLE IF NOT EXISTS sessions ( 59 id INTEGER PRIMARY KEY AUTOINCREMENT, 60 user_id INTEGER NOT NULL REFERENCES users(id), 61 mode TEXT NOT NULL, 62 started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 63 ended_at TIMESTAMP, 64 words_attempted INTEGER DEFAULT 0, 65 words_correct INTEGER DEFAULT 0, 66 total_time_ms INTEGER DEFAULT 0 67 ); 68 69 CREATE TABLE IF NOT EXISTS attempts ( 70 id INTEGER PRIMARY KEY AUTOINCREMENT, 71 session_id INTEGER REFERENCES sessions(id), 72 word TEXT NOT NULL, 73 strokes TEXT, 74 typed_text TEXT, 75 correct BOOLEAN NOT NULL, 76 time_ms INTEGER NOT NULL, 77 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 78 ); 79 80 CREATE TABLE IF NOT EXISTS word_stats ( 81 user_id INTEGER NOT NULL REFERENCES users(id), 82 word TEXT NOT NULL, 83 strokes TEXT, 84 attempts INTEGER DEFAULT 0, 85 correct INTEGER DEFAULT 0, 86 avg_time_ms REAL DEFAULT 0, 87 last_seen TIMESTAMP, 88 next_review TIMESTAMP, 89 ease_factor REAL DEFAULT 2.5, 90 interval_days REAL DEFAULT 0, 91 repetitions INTEGER DEFAULT 0, 92 PRIMARY KEY (user_id, word) 93 ); 94 95 CREATE TABLE IF NOT EXISTS daily_stats ( 96 user_id INTEGER NOT NULL REFERENCES users(id), 97 date TEXT NOT NULL, 98 words_practiced INTEGER DEFAULT 0, 99 words_correct INTEGER DEFAULT 0, 100 total_time_ms INTEGER DEFAULT 0, 101 sessions INTEGER DEFAULT 0, 102 PRIMARY KEY (user_id, date) 103 ); 104 105 CREATE TABLE IF NOT EXISTS user_dictionaries ( 106 id INTEGER PRIMARY KEY AUTOINCREMENT, 107 user_id INTEGER NOT NULL REFERENCES users(id), 108 filename TEXT NOT NULL, 109 display_name TEXT, 110 dict_order INTEGER NOT NULL DEFAULT 0, 111 enabled BOOLEAN DEFAULT 1, 112 entry_count INTEGER DEFAULT 0, 113 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 114 ); 115 116 CREATE TABLE IF NOT EXISTS user_settings ( 117 user_id INTEGER PRIMARY KEY REFERENCES users(id), 118 words_per_session INTEGER DEFAULT 50 119 ); 120 121 CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id); 122 CREATE INDEX IF NOT EXISTS idx_attempts_session ON attempts(session_id); 123 CREATE INDEX IF NOT EXISTS idx_attempts_word ON attempts(word); 124 CREATE INDEX IF NOT EXISTS idx_word_stats_review ON word_stats(user_id, next_review); 125 CREATE INDEX IF NOT EXISTS idx_daily_stats_user ON daily_stats(user_id, date); 126 CREATE INDEX IF NOT EXISTS idx_user_dicts ON user_dictionaries(user_id, dict_order); 127 """) 128 129 def _migrate_to_v2(self, conn): 130 print("Migrating database to multi-user schema...") 131 132 conn.execute(""" 133 CREATE TABLE IF NOT EXISTS users ( 134 id INTEGER PRIMARY KEY AUTOINCREMENT, 135 email TEXT UNIQUE NOT NULL, 136 name TEXT, 137 avatar_url TEXT, 138 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 139 ) 140 """) 141 conn.execute("INSERT OR IGNORE INTO users (id, email, name) VALUES (1, 'local@stenodojo', 'Local User')") 142 143 session_cols = {r[1] for r in conn.execute("PRAGMA table_info(sessions)").fetchall()} 144 if "user_id" not in session_cols: 145 conn.execute("ALTER TABLE sessions ADD COLUMN user_id INTEGER DEFAULT 1") 146 147 ws_cols = {r[1] for r in conn.execute("PRAGMA table_info(word_stats)").fetchall()} 148 if "user_id" not in ws_cols: 149 conn.execute(""" 150 CREATE TABLE word_stats_v2 ( 151 user_id INTEGER NOT NULL DEFAULT 1 REFERENCES users(id), 152 word TEXT NOT NULL, 153 strokes TEXT, 154 attempts INTEGER DEFAULT 0, 155 correct INTEGER DEFAULT 0, 156 avg_time_ms REAL DEFAULT 0, 157 last_seen TIMESTAMP, 158 next_review TIMESTAMP, 159 ease_factor REAL DEFAULT 2.5, 160 interval_days REAL DEFAULT 0, 161 repetitions INTEGER DEFAULT 0, 162 PRIMARY KEY (user_id, word) 163 ) 164 """) 165 conn.execute(""" 166 INSERT INTO word_stats_v2 (user_id, word, strokes, attempts, correct, avg_time_ms, 167 last_seen, next_review, ease_factor, interval_days, repetitions) 168 SELECT 1, word, strokes, attempts, correct, avg_time_ms, 169 last_seen, next_review, ease_factor, interval_days, repetitions 170 FROM word_stats 171 """) 172 conn.execute("DROP TABLE word_stats") 173 conn.execute("ALTER TABLE word_stats_v2 RENAME TO word_stats") 174 175 ds_cols = {r[1] for r in conn.execute("PRAGMA table_info(daily_stats)").fetchall()} 176 if "user_id" not in ds_cols: 177 conn.execute(""" 178 CREATE TABLE daily_stats_v2 ( 179 user_id INTEGER NOT NULL DEFAULT 1 REFERENCES users(id), 180 date TEXT NOT NULL, 181 words_practiced INTEGER DEFAULT 0, 182 words_correct INTEGER DEFAULT 0, 183 total_time_ms INTEGER DEFAULT 0, 184 sessions INTEGER DEFAULT 0, 185 PRIMARY KEY (user_id, date) 186 ) 187 """) 188 conn.execute(""" 189 INSERT INTO daily_stats_v2 (user_id, date, words_practiced, words_correct, total_time_ms, sessions) 190 SELECT 1, date, words_practiced, words_correct, total_time_ms, sessions 191 FROM daily_stats 192 """) 193 conn.execute("DROP TABLE daily_stats") 194 conn.execute("ALTER TABLE daily_stats_v2 RENAME TO daily_stats") 195 196 conn.executescript(""" 197 CREATE TABLE IF NOT EXISTS user_dictionaries ( 198 id INTEGER PRIMARY KEY AUTOINCREMENT, 199 user_id INTEGER NOT NULL REFERENCES users(id), 200 filename TEXT NOT NULL, 201 display_name TEXT, 202 dict_order INTEGER NOT NULL DEFAULT 0, 203 enabled BOOLEAN DEFAULT 1, 204 entry_count INTEGER DEFAULT 0, 205 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP 206 ); 207 CREATE TABLE IF NOT EXISTS user_settings ( 208 user_id INTEGER PRIMARY KEY REFERENCES users(id), 209 words_per_session INTEGER DEFAULT 50 210 ); 211 CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id); 212 CREATE INDEX IF NOT EXISTS idx_word_stats_review ON word_stats(user_id, next_review); 213 CREATE INDEX IF NOT EXISTS idx_daily_stats_user ON daily_stats(user_id, date); 214 CREATE INDEX IF NOT EXISTS idx_user_dicts ON user_dictionaries(user_id, dict_order); 215 """) 216 217 print("Migration complete.") 218 219 # ── User methods ── 220 221 def get_or_create_user(self, email, name=None, avatar_url=None): 222 with self._get_conn() as conn: 223 row = conn.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone() 224 if row: 225 if name or avatar_url: 226 conn.execute( 227 "UPDATE users SET name = COALESCE(?, name), avatar_url = COALESCE(?, avatar_url) WHERE id = ?", 228 (name, avatar_url, row["id"]), 229 ) 230 return dict(conn.execute("SELECT * FROM users WHERE id = ?", (row["id"],)).fetchone()) 231 cursor = conn.execute( 232 "INSERT INTO users (email, name, avatar_url) VALUES (?, ?, ?)", 233 (email, name, avatar_url), 234 ) 235 return {"id": cursor.lastrowid, "email": email, "name": name, "avatar_url": avatar_url} 236 237 def get_user(self, user_id): 238 with self._get_conn() as conn: 239 row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() 240 return dict(row) if row else None 241 242 def ensure_dev_user(self): 243 return self.get_or_create_user("dev@stenodojo.local", name="Dev User") 244 245 # ── User Settings ── 246 247 def get_user_settings(self, user_id): 248 with self._get_conn() as conn: 249 row = conn.execute("SELECT * FROM user_settings WHERE user_id = ?", (user_id,)).fetchone() 250 if row: 251 return dict(row) 252 return {"user_id": user_id, "words_per_session": 50} 253 254 def update_user_settings(self, user_id, **kwargs): 255 allowed = {"words_per_session"} 256 kwargs = {k: v for k, v in kwargs.items() if k in allowed} 257 if not kwargs: 258 return 259 with self._get_conn() as conn: 260 existing = conn.execute("SELECT 1 FROM user_settings WHERE user_id = ?", (user_id,)).fetchone() 261 if existing: 262 sets = ", ".join(f"{k} = ?" for k in kwargs) 263 conn.execute(f"UPDATE user_settings SET {sets} WHERE user_id = ?", (*kwargs.values(), user_id)) 264 else: 265 kwargs["user_id"] = user_id 266 cols = ", ".join(kwargs.keys()) 267 placeholders = ", ".join("?" * len(kwargs)) 268 conn.execute(f"INSERT INTO user_settings ({cols}) VALUES ({placeholders})", tuple(kwargs.values())) 269 270 # ── Dictionary management ── 271 272 def get_user_dictionaries(self, user_id): 273 with self._get_conn() as conn: 274 rows = conn.execute( 275 "SELECT * FROM user_dictionaries WHERE user_id = ? ORDER BY dict_order", 276 (user_id,), 277 ).fetchall() 278 return [dict(r) for r in rows] 279 280 def add_user_dictionary(self, user_id, filename, display_name, dict_order, entry_count=0): 281 with self._get_conn() as conn: 282 cursor = conn.execute( 283 """INSERT INTO user_dictionaries (user_id, filename, display_name, dict_order, entry_count) 284 VALUES (?, ?, ?, ?, ?)""", 285 (user_id, filename, display_name, dict_order, entry_count), 286 ) 287 return cursor.lastrowid 288 289 def clear_user_dictionaries(self, user_id): 290 with self._get_conn() as conn: 291 conn.execute("DELETE FROM user_dictionaries WHERE user_id = ?", (user_id,)) 292 293 def toggle_dictionary(self, dict_id, user_id, enabled): 294 with self._get_conn() as conn: 295 conn.execute( 296 "UPDATE user_dictionaries SET enabled = ? WHERE id = ? AND user_id = ?", 297 (enabled, dict_id, user_id), 298 ) 299 300 def delete_dictionary(self, dict_id, user_id): 301 with self._get_conn() as conn: 302 row = conn.execute( 303 "SELECT filename FROM user_dictionaries WHERE id = ? AND user_id = ?", 304 (dict_id, user_id), 305 ).fetchone() 306 if row: 307 conn.execute("DELETE FROM user_dictionaries WHERE id = ? AND user_id = ?", (dict_id, user_id)) 308 return row["filename"] 309 return None 310 311 def update_dictionary_entry_count(self, dict_id, user_id, count): 312 with self._get_conn() as conn: 313 conn.execute( 314 "UPDATE user_dictionaries SET entry_count = ? WHERE id = ? AND user_id = ?", 315 (count, dict_id, user_id), 316 ) 317 318 # ── Session methods ── 319 320 def create_session(self, user_id, mode): 321 with self._get_conn() as conn: 322 cursor = conn.execute( 323 "INSERT INTO sessions (user_id, mode) VALUES (?, ?)", 324 (user_id, mode), 325 ) 326 return cursor.lastrowid 327 328 def end_session(self, session_id): 329 with self._get_conn() as conn: 330 conn.execute( 331 "UPDATE sessions SET ended_at = CURRENT_TIMESTAMP WHERE id = ?", 332 (session_id,), 333 ) 334 335 def record_attempt(self, user_id, session_id, word, strokes, typed_text, correct, time_ms): 336 with self._get_conn() as conn: 337 conn.execute( 338 """INSERT INTO attempts (session_id, word, strokes, typed_text, correct, time_ms) 339 VALUES (?, ?, ?, ?, ?, ?)""", 340 (session_id, word, strokes, typed_text, correct, time_ms), 341 ) 342 343 conn.execute( 344 """UPDATE sessions SET 345 words_attempted = words_attempted + 1, 346 words_correct = words_correct + ?, 347 total_time_ms = total_time_ms + ? 348 WHERE id = ?""", 349 (1 if correct else 0, time_ms, session_id), 350 ) 351 352 existing = conn.execute( 353 "SELECT * FROM word_stats WHERE user_id = ? AND word = ?", (user_id, word) 354 ).fetchone() 355 356 now = datetime.utcnow().isoformat() 357 358 if existing: 359 old_avg = existing["avg_time_ms"] 360 old_count = existing["attempts"] 361 new_avg = ((old_avg * old_count) + time_ms) / (old_count + 1) 362 363 new_ef, new_interval, new_reps = calculate_review( 364 correct, time_ms, 365 existing["ease_factor"], 366 existing["interval_days"], 367 existing["repetitions"], 368 ) 369 next_review = (datetime.utcnow() + timedelta(days=new_interval)).isoformat() 370 371 conn.execute( 372 """UPDATE word_stats SET 373 strokes = ?, attempts = attempts + 1, correct = correct + ?, 374 avg_time_ms = ?, last_seen = ?, next_review = ?, 375 ease_factor = ?, interval_days = ?, repetitions = ? 376 WHERE user_id = ? AND word = ?""", 377 (strokes, 1 if correct else 0, new_avg, now, 378 next_review, new_ef, new_interval, new_reps, user_id, word), 379 ) 380 else: 381 new_ef, new_interval, new_reps = calculate_review(correct, time_ms, 2.5, 0, 0) 382 next_review = (datetime.utcnow() + timedelta(days=new_interval)).isoformat() 383 384 conn.execute( 385 """INSERT INTO word_stats 386 (user_id, word, strokes, attempts, correct, avg_time_ms, 387 last_seen, next_review, ease_factor, interval_days, repetitions) 388 VALUES (?, ?, ?, 1, ?, ?, ?, ?, ?, ?, ?)""", 389 (user_id, word, strokes, 1 if correct else 0, float(time_ms), 390 now, next_review, new_ef, new_interval, new_reps), 391 ) 392 393 today = datetime.utcnow().strftime("%Y-%m-%d") 394 existing_daily = conn.execute( 395 "SELECT * FROM daily_stats WHERE user_id = ? AND date = ?", (user_id, today) 396 ).fetchone() 397 398 if existing_daily: 399 conn.execute( 400 """UPDATE daily_stats SET 401 words_practiced = words_practiced + 1, 402 words_correct = words_correct + ?, 403 total_time_ms = total_time_ms + ? 404 WHERE user_id = ? AND date = ?""", 405 (1 if correct else 0, time_ms, user_id, today), 406 ) 407 else: 408 conn.execute( 409 """INSERT INTO daily_stats (user_id, date, words_practiced, words_correct, total_time_ms, sessions) 410 VALUES (?, ?, 1, ?, ?, 0)""", 411 (user_id, today, 1 if correct else 0, time_ms), 412 ) 413 414 def get_session(self, session_id): 415 with self._get_conn() as conn: 416 row = conn.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)).fetchone() 417 return dict(row) if row else None 418 419 def get_session_attempts(self, session_id): 420 with self._get_conn() as conn: 421 rows = conn.execute( 422 "SELECT * FROM attempts WHERE session_id = ? ORDER BY created_at", 423 (session_id,), 424 ).fetchall() 425 return [dict(r) for r in rows] 426 427 def get_session_stats(self, session_id): 428 with self._get_conn() as conn: 429 session = conn.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)).fetchone() 430 if not session: 431 return None 432 433 attempts = conn.execute( 434 "SELECT correct, time_ms FROM attempts WHERE session_id = ? ORDER BY created_at", 435 (session_id,), 436 ).fetchall() 437 438 words_attempted = len(attempts) 439 words_correct = sum(1 for a in attempts if a["correct"]) 440 total_time = sum(a["time_ms"] for a in attempts) 441 accuracy = words_correct / words_attempted if words_attempted > 0 else 0 442 443 best_streak = 0 444 current_streak = 0 445 for a in attempts: 446 if a["correct"]: 447 current_streak += 1 448 best_streak = max(best_streak, current_streak) 449 else: 450 current_streak = 0 451 452 return { 453 "words_attempted": words_attempted, 454 "words_correct": words_correct, 455 "accuracy": accuracy, 456 "avg_time_ms": total_time / words_attempted if words_attempted > 0 else 0, 457 "total_time_ms": total_time, 458 "best_streak": best_streak, 459 } 460 461 def get_word_stats(self, user_id, word=None): 462 with self._get_conn() as conn: 463 if word: 464 row = conn.execute( 465 "SELECT * FROM word_stats WHERE user_id = ? AND word = ?", (user_id, word) 466 ).fetchone() 467 return dict(row) if row else None 468 rows = conn.execute( 469 "SELECT * FROM word_stats WHERE user_id = ? ORDER BY last_seen DESC", (user_id,) 470 ).fetchall() 471 return [dict(r) for r in rows] 472 473 def get_words_due_for_review(self, user_id, count=25): 474 now = datetime.utcnow().isoformat() 475 with self._get_conn() as conn: 476 rows = conn.execute( 477 """SELECT word FROM word_stats 478 WHERE user_id = ? AND next_review <= ? 479 ORDER BY next_review ASC LIMIT ?""", 480 (user_id, now, count), 481 ).fetchall() 482 return [r["word"] for r in rows] 483 484 def get_weakest_words(self, user_id, count=25): 485 with self._get_conn() as conn: 486 rows = conn.execute( 487 """SELECT word, attempts, correct, 488 CAST(correct AS REAL) / attempts AS accuracy 489 FROM word_stats 490 WHERE user_id = ? AND attempts >= 3 491 ORDER BY accuracy ASC, attempts DESC LIMIT ?""", 492 (user_id, count), 493 ).fetchall() 494 return [dict(r) for r in rows] 495 496 def get_practiced_words(self, user_id): 497 with self._get_conn() as conn: 498 rows = conn.execute("SELECT word FROM word_stats WHERE user_id = ?", (user_id,)).fetchall() 499 return {r["word"] for r in rows} 500 501 def get_daily_stats(self, user_id, days=30): 502 start_date = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d") 503 with self._get_conn() as conn: 504 rows = conn.execute( 505 "SELECT * FROM daily_stats WHERE user_id = ? AND date >= ? ORDER BY date DESC", 506 (user_id, start_date), 507 ).fetchall() 508 return [dict(r) for r in rows] 509 510 def get_heatmap_data(self, user_id): 511 start_date = (datetime.utcnow() - timedelta(days=365)).strftime("%Y-%m-%d") 512 with self._get_conn() as conn: 513 rows = conn.execute( 514 """SELECT date, words_practiced AS count 515 FROM daily_stats WHERE user_id = ? AND date >= ? ORDER BY date""", 516 (user_id, start_date), 517 ).fetchall() 518 return [dict(r) for r in rows] 519 520 def get_overview_stats(self, user_id): 521 with self._get_conn() as conn: 522 totals = conn.execute( 523 """SELECT 524 COALESCE(SUM(words_practiced), 0) AS total_words, 525 COALESCE(SUM(words_correct), 0) AS total_correct, 526 COALESCE(SUM(total_time_ms), 0) AS total_time_ms, 527 COALESCE(SUM(sessions), 0) AS total_sessions, 528 COUNT(*) AS days_practiced 529 FROM daily_stats WHERE user_id = ?""", 530 (user_id,), 531 ).fetchone() 532 533 streak = self._calculate_streak(conn, user_id) 534 mastered = self.get_words_mastered_count(user_id) 535 536 unique_words = conn.execute( 537 "SELECT COUNT(*) AS count FROM word_stats WHERE user_id = ?", (user_id,) 538 ).fetchone()["count"] 539 540 total_words = totals["total_words"] 541 total_correct = totals["total_correct"] 542 543 return { 544 "total_words_practiced": total_words, 545 "total_correct": total_correct, 546 "overall_accuracy": total_correct / total_words if total_words > 0 else 0, 547 "total_time_ms": totals["total_time_ms"], 548 "total_sessions": totals["total_sessions"], 549 "days_practiced": totals["days_practiced"], 550 "current_streak": streak, 551 "words_mastered": mastered, 552 "unique_words": unique_words, 553 } 554 555 def _calculate_streak(self, conn, user_id): 556 today = datetime.utcnow().date() 557 streak = 0 558 current_date = today 559 while True: 560 date_str = current_date.strftime("%Y-%m-%d") 561 row = conn.execute( 562 "SELECT 1 FROM daily_stats WHERE user_id = ? AND date = ? AND words_practiced > 0", 563 (user_id, date_str), 564 ).fetchone() 565 if row: 566 streak += 1 567 current_date -= timedelta(days=1) 568 else: 569 break 570 return streak 571 572 def get_recent_sessions(self, user_id, limit=5): 573 with self._get_conn() as conn: 574 rows = conn.execute( 575 "SELECT * FROM sessions WHERE user_id = ? ORDER BY started_at DESC LIMIT ?", 576 (user_id, limit), 577 ).fetchall() 578 return [dict(r) for r in rows] 579 580 def get_all_sessions(self, user_id, limit=50): 581 with self._get_conn() as conn: 582 rows = conn.execute( 583 "SELECT * FROM sessions WHERE user_id = ? ORDER BY started_at DESC LIMIT ?", 584 (user_id, limit), 585 ).fetchall() 586 return [dict(r) for r in rows] 587 588 def get_words_mastered_count(self, user_id): 589 with self._get_conn() as conn: 590 row = conn.execute( 591 """SELECT COUNT(*) AS count FROM word_stats 592 WHERE user_id = ? AND attempts >= 5 593 AND CAST(correct AS REAL) / attempts >= 0.9""", 594 (user_id,), 595 ).fetchone() 596 return row["count"] 597 598 def increment_daily_sessions(self, user_id): 599 today = datetime.utcnow().strftime("%Y-%m-%d") 600 with self._get_conn() as conn: 601 existing = conn.execute( 602 "SELECT * FROM daily_stats WHERE user_id = ? AND date = ?", (user_id, today) 603 ).fetchone() 604 if existing: 605 conn.execute( 606 "UPDATE daily_stats SET sessions = sessions + 1 WHERE user_id = ? AND date = ?", 607 (user_id, today), 608 ) 609 else: 610 conn.execute( 611 """INSERT INTO daily_stats (user_id, date, words_practiced, words_correct, total_time_ms, sessions) 612 VALUES (?, ?, 0, 0, 0, 1)""", 613 (user_id, today), 614 ) 615 616 def get_lesson_progress(self, user_id, word_list): 617 """Compute mastery progress for a set of words. 618 619 Returns dict with: 620 total_words, practiced, mastered, avg_wpm, avg_accuracy, 621 per-word breakdown of attempts/accuracy/avg_time. 622 623 Mastery target: 95% accuracy and avg_time < 1500ms (~200 WPM for 624 a 5-char word) with at least 5 attempts. 625 """ 626 if not word_list: 627 return {"total_words": 0, "practiced": 0, "mastered": 0, 628 "avg_wpm": 0, "avg_accuracy": 0, "words": []} 629 630 lower_words = [w.lower() for w in word_list] 631 with self._get_conn() as conn: 632 placeholders = ",".join("?" * len(lower_words)) 633 rows = conn.execute( 634 f"""SELECT word, attempts, correct, avg_time_ms 635 FROM word_stats 636 WHERE user_id = ? AND LOWER(word) IN ({placeholders})""", 637 (user_id, *lower_words), 638 ).fetchall() 639 640 stats_map = {r["word"].lower(): dict(r) for r in rows} 641 642 practiced = 0 643 mastered = 0 644 total_time = 0 645 total_correct = 0 646 total_attempts = 0 647 words = [] 648 649 for w in lower_words: 650 s = stats_map.get(w) 651 if s and s["attempts"] > 0: 652 practiced += 1 653 acc = s["correct"] / s["attempts"] 654 total_correct += s["correct"] 655 total_attempts += s["attempts"] 656 total_time += s["avg_time_ms"] * s["attempts"] 657 if s["attempts"] >= 5 and acc >= 0.95 and s["avg_time_ms"] < 1500: 658 mastered += 1 659 words.append({ 660 "word": w, 661 "attempts": s["attempts"], 662 "accuracy": round(acc * 100, 1), 663 "avg_time_ms": round(s["avg_time_ms"]), 664 }) 665 else: 666 words.append({ 667 "word": w, 668 "attempts": 0, 669 "accuracy": 0, 670 "avg_time_ms": 0, 671 }) 672 673 avg_accuracy = (total_correct / total_attempts * 100) if total_attempts > 0 else 0 674 avg_time = (total_time / total_attempts) if total_attempts > 0 else 0 675 avg_wpm = (5 / (avg_time / 60000)) if avg_time > 0 else 0 676 677 return { 678 "total_words": len(lower_words), 679 "practiced": practiced, 680 "mastered": mastered, 681 "avg_wpm": round(avg_wpm, 1), 682 "avg_accuracy": round(avg_accuracy, 1), 683 "words": words, 684 } 685 686 def clear_user_data(self, user_id): 687 with self._get_conn() as conn: 688 session_ids = [r["id"] for r in conn.execute( 689 "SELECT id FROM sessions WHERE user_id = ?", (user_id,) 690 ).fetchall()] 691 if session_ids: 692 placeholders = ",".join("?" * len(session_ids)) 693 conn.execute(f"DELETE FROM attempts WHERE session_id IN ({placeholders})", session_ids) 694 conn.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) 695 conn.execute("DELETE FROM word_stats WHERE user_id = ?", (user_id,)) 696 conn.execute("DELETE FROM daily_stats WHERE user_id = ?", (user_id,))