app.py (35921B)
1 """Flask application for multi-user stenography practice.""" 2 3 import configparser 4 import json 5 import os 6 import random 7 import shutil 8 import re 9 import time 10 from functools import wraps 11 12 from authlib.integrations.flask_client import OAuth 13 from dotenv import load_dotenv 14 from flask import Flask, jsonify, redirect, render_template, request, session, url_for 15 from werkzeug.utils import secure_filename 16 17 from database import Database 18 from lessons import ( 19 TOP_WORDS, 20 get_lesson_by_id, 21 get_lesson_list, 22 get_lesson_word_list, 23 get_lesson_words, 24 get_sentences, 25 ) 26 from plover_engine import PloverEngine 27 28 load_dotenv() 29 30 app = Flask(__name__) 31 app.secret_key = os.environ.get("SECRET_KEY", "dev-secret-key-change-me") 32 app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0 33 app.config["MAX_CONTENT_LENGTH"] = 50 * 1024 * 1024 34 35 UPLOAD_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads") 36 DEV_MODE = not os.environ.get("GOOGLE_CLIENT_ID") 37 38 _start_time = int(time.time()) 39 40 oauth = OAuth(app) 41 if not DEV_MODE: 42 google = oauth.register( 43 name="google", 44 client_id=os.environ["GOOGLE_CLIENT_ID"], 45 client_secret=os.environ["GOOGLE_CLIENT_SECRET"], 46 server_metadata_url="https://accounts.google.com/.well-known/openid-configuration", 47 client_kwargs={"scope": "openid email profile"}, 48 ) 49 50 db = Database() 51 52 if DEV_MODE: 53 dev_user = db.ensure_dev_user() 54 print(f"DEV MODE: No Google OAuth configured. Using dev user (id={dev_user['id']}).") 55 56 # Per-user engine cache 57 _engine_cache = {} 58 59 60 def _get_user_dict_dir(user_id): 61 return os.path.join(UPLOAD_DIR, str(user_id)) 62 63 64 def get_engine(user_id): 65 if user_id in _engine_cache: 66 return _engine_cache[user_id] 67 engine = PloverEngine() 68 dict_dir = _get_user_dict_dir(user_id) 69 dicts = db.get_user_dictionaries(user_id) 70 if dicts: 71 engine.load_from_uploaded(dict_dir, [dict(d) for d in dicts]) 72 _engine_cache[user_id] = engine 73 return engine 74 75 76 def reload_engine(user_id): 77 _engine_cache.pop(user_id, None) 78 return get_engine(user_id) 79 80 81 @app.context_processor 82 def inject_globals(): 83 user = None 84 if "user_id" in session: 85 user = db.get_user(session["user_id"]) 86 return {"cache_bust": _start_time, "current_user": user, "dev_mode": DEV_MODE} 87 88 89 @app.after_request 90 def add_no_cache(response): 91 if "text/html" in response.content_type or "javascript" in response.content_type: 92 response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" 93 return response 94 95 96 # ── Auth ── 97 98 99 def login_required(f): 100 @wraps(f) 101 def decorated(*args, **kwargs): 102 if DEV_MODE: 103 session.setdefault("user_id", dev_user["id"]) 104 return f(*args, **kwargs) 105 if "user_id" not in session: 106 return redirect(url_for("login")) 107 return f(*args, **kwargs) 108 return decorated 109 110 111 def current_user_id(): 112 return session.get("user_id") 113 114 115 @app.route("/login") 116 def login(): 117 if "user_id" in session: 118 return redirect("/") 119 if DEV_MODE: 120 session["user_id"] = dev_user["id"] 121 return redirect("/") 122 return render_template("login.html") 123 124 125 @app.route("/login/google") 126 def login_google(): 127 if DEV_MODE: 128 session["user_id"] = dev_user["id"] 129 return redirect("/") 130 redirect_uri = url_for("auth_google_callback", _external=True) 131 return google.authorize_redirect(redirect_uri) 132 133 134 @app.route("/auth/google/callback") 135 def auth_google_callback(): 136 if DEV_MODE: 137 session["user_id"] = dev_user["id"] 138 return redirect("/") 139 140 token = google.authorize_access_token() 141 userinfo = token.get("userinfo") 142 if not userinfo: 143 return "Authentication failed", 400 144 145 if not userinfo.get("email_verified"): 146 return render_template("login.html", error="Your Google email is not verified. Please verify it and try again.") 147 148 user = db.get_or_create_user( 149 email=userinfo["email"], 150 name=userinfo.get("name"), 151 avatar_url=userinfo.get("picture"), 152 ) 153 session["user_id"] = user["id"] 154 session.permanent = True 155 return redirect("/") 156 157 158 @app.route("/logout") 159 def logout(): 160 session.clear() 161 if DEV_MODE: 162 return redirect("/") 163 return redirect(url_for("login")) 164 165 166 # ── Page routes ── 167 168 169 MODE_LABELS = { 170 "common": "Common Words", 171 "random": "Random", 172 "new": "New Words", 173 "weak": "Weak Words", 174 "review": "Review", 175 "speed": "Speed Drill", 176 "sentences": "Sentences", 177 "phrases": "Phrases", 178 "lesson": "Lesson", 179 } 180 181 182 @app.route("/") 183 @login_required 184 def dashboard(): 185 uid = current_user_id() 186 daily = db.get_daily_stats(uid, days=1) 187 raw_today = daily[0] if daily else { 188 "words_practiced": 0, "words_correct": 0, "total_time_ms": 0, "sessions": 0 189 } 190 wp = raw_today.get("words_practiced", 0) 191 wc = raw_today.get("words_correct", 0) 192 today = { 193 "words": wp, 194 "correct": wc, 195 "accuracy": (wc / wp * 100) if wp > 0 else 0, 196 "sessions": raw_today.get("sessions", 0), 197 } 198 199 overview = db.get_overview_stats(uid) 200 streak = overview["current_streak"] 201 words_mastered = overview["words_mastered"] 202 total_practiced = overview["unique_words"] 203 review_due = len(db.get_words_due_for_review(uid, count=1000)) 204 205 lesson_list = get_lesson_list() 206 for l in lesson_list: 207 lesson_def = get_lesson_by_id(l["id"]) 208 if lesson_def: 209 word_list = get_lesson_word_list(lesson_def) 210 if word_list: 211 prog = db.get_lesson_progress(uid, word_list) 212 l["progress"] = prog 213 214 raw_sessions = db.get_recent_sessions(uid, limit=5) 215 recent_sessions = [] 216 for s in raw_sessions: 217 wa = s.get("words_attempted", 0) 218 wco = s.get("words_correct", 0) 219 total_ms = s.get("total_time_ms", 0) 220 total_chars = wa * 5 221 wpm = (total_chars / 5) / (total_ms / 60000) if total_ms > 0 else 0 222 recent_sessions.append({ 223 "mode": s.get("mode", ""), 224 "date_str": (s.get("started_at") or "")[:10], 225 "words_attempted": wa, 226 "words_correct": wco, 227 "accuracy": (wco / wa * 100) if wa > 0 else 0, 228 "wpm": round(wpm, 1), 229 }) 230 231 engine = get_engine(uid) 232 has_dicts = bool(engine._practice_words) 233 234 return render_template( 235 "dashboard.html", 236 today=today, 237 streak=streak, 238 words_mastered=words_mastered, 239 total_practiced=total_practiced, 240 review_due=review_due, 241 recent_sessions=recent_sessions, 242 lessons=lesson_list, 243 has_dicts=has_dicts, 244 ) 245 246 247 @app.route("/practice") 248 @login_required 249 def practice(): 250 mode = request.args.get("mode", "common") 251 lesson_id = request.args.get("lesson") 252 mode_label = MODE_LABELS.get(mode, "Practice") 253 if lesson_id: 254 lesson = get_lesson_by_id(lesson_id) 255 if lesson: 256 mode_label = lesson["title"] 257 return render_template("practice.html", mode=mode, mode_label=mode_label, lesson_id=lesson_id or "") 258 259 260 @app.route("/stats") 261 @login_required 262 def stats(): 263 return render_template("stats.html") 264 265 266 @app.route("/settings") 267 @login_required 268 def settings_page(): 269 uid = current_user_id() 270 user_settings = db.get_user_settings(uid) 271 user_dicts = db.get_user_dictionaries(uid) 272 return render_template("settings.html", settings=user_settings, dictionaries=user_dicts) 273 274 275 # ── Session API ── 276 277 session_store = {} 278 279 280 @app.route("/api/session/start", methods=["POST"]) 281 @login_required 282 def api_session_start(): 283 uid = current_user_id() 284 engine = get_engine(uid) 285 data = request.get_json() 286 mode = data.get("mode", "common") 287 lesson_id = data.get("lesson_id") 288 user_settings = db.get_user_settings(uid) 289 count = data.get("count", user_settings.get("words_per_session", 50)) 290 291 if lesson_id: 292 mode = "lesson" 293 lesson = get_lesson_by_id(lesson_id) 294 if not lesson: 295 return jsonify({"error": "Lesson not found"}), 404 296 sentence_count = data.get("sentence_count", 10) 297 items = get_lesson_words(lesson, engine, count=sentence_count) 298 if lesson.get("mode") != "phrases": 299 return _start_sentence_session(uid, engine, mode, items) 300 else: 301 items = _select_words(uid, engine, mode, count) 302 303 if not items: 304 return jsonify({"error": "No words available for this mode"}), 400 305 306 if mode == "sentences": 307 return _start_sentence_session(uid, engine, mode, items) 308 309 session_id = db.create_session(uid, mode) 310 db.increment_daily_sessions(uid) 311 312 words_data = [] 313 for i, word in enumerate(items): 314 display = engine.display_form(word) 315 hints = engine.get_all_hints(word) 316 words_data.append({ 317 "word": display, 318 "index": i, 319 "word_hints": hints["word_hints"], 320 "phrase_hints": hints["phrase_hints"], 321 "fingerspell": hints.get("fingerspell", False), 322 "punct_hints": hints.get("punct_hints", []), 323 "cap_hints": hints.get("cap_hints", []), 324 }) 325 326 session_store[session_id] = { 327 "words": items, 328 "user_id": uid, 329 "index": 0, 330 "streak": 0, 331 "best_streak": 0, 332 } 333 334 return jsonify({ 335 "session_id": session_id, 336 "words": words_data, 337 "total": len(items), 338 }) 339 340 341 _SENT_PUNCT = set('.,;:!?"\')}]>…') 342 343 344 def _tokenize_sentence(sentence): 345 """Split sentence into steno tokens: words and trailing punctuation separately. 346 Each token has a 'prefix' = the expected sentence text up through that token, 347 used for prefix-based input matching.""" 348 tokens = [] 349 for raw_word in sentence.split(): 350 idx = sentence.index(raw_word, tokens[-1]["end"] if tokens else 0) 351 352 word = raw_word 353 while word and word[-1] in _SENT_PUNCT: 354 word = word[:-1] 355 356 if word: 357 word_end = idx + len(word) 358 tokens.append({"text": word, "prefix": sentence[:word_end], 359 "end": word_end, "is_punct": False}) 360 361 for j in range(len(word), len(raw_word)): 362 char_end = idx + j + 1 363 tokens.append({"text": raw_word[j], "prefix": sentence[:char_end], 364 "end": char_end, "is_punct": True}) 365 return tokens 366 367 368 def _start_sentence_session(uid, engine, mode, sentences): 369 session_id = db.create_session(uid, mode) 370 db.increment_daily_sessions(uid) 371 372 # Pass 1: build tokens with cumulative prefixes 373 raw_tokens = [] 374 cumulative_text = "" 375 for sentence in sentences: 376 sent_words = sentence.split() 377 tokens = _tokenize_sentence(sentence) 378 for token in tokens: 379 if token["is_punct"]: 380 cumulative_text += token["text"] 381 else: 382 if cumulative_text: 383 cumulative_text += " " 384 cumulative_text += token["text"] 385 raw_tokens.append({ 386 "text": token["text"], 387 "is_punct": token["is_punct"], 388 "prefix": cumulative_text, 389 "sentence": sentence, 390 "sent_words": sent_words, 391 }) 392 393 # Pass 2: add hints with lookahead context 394 all_tokens = [] 395 for i, tok in enumerate(raw_tokens): 396 if tok["is_punct"]: 397 next_word = None 398 for j in range(i + 1, len(raw_tokens)): 399 if not raw_tokens[j]["is_punct"]: 400 next_word = raw_tokens[j]["text"] 401 break 402 space_after = next_word is not None 403 cap_next = bool(next_word and next_word[0].isupper()) 404 hints = { 405 "word_hints": engine.get_punct_hints( 406 tok["text"], space_after=space_after, cap_next=cap_next), 407 "phrase_hints": [], "fingerspell": False, 408 "punct_hints": [], "cap_hints": [], 409 } 410 else: 411 sent_words = tok["sent_words"] 412 word_idx = next((wi for wi, w in enumerate(sent_words) 413 if w.startswith(tok["text"])), 0) 414 hints = engine.get_all_hints( 415 tok["text"], context_words=sent_words, 416 context_index=word_idx) 417 418 all_tokens.append({ 419 "word": tok["text"], 420 "prefix": tok["prefix"], 421 "is_punct": tok["is_punct"], 422 "index": len(all_tokens), 423 "word_hints": hints["word_hints"], 424 "phrase_hints": hints.get("phrase_hints", []), 425 "fingerspell": hints.get("fingerspell", False), 426 "punct_hints": hints.get("punct_hints", []), 427 "cap_hints": hints.get("cap_hints", []), 428 "sentence": tok["sentence"], 429 }) 430 431 stat_words = [t["word"] for t in all_tokens] 432 433 session_store[session_id] = { 434 "words": stat_words, 435 "user_id": uid, 436 "is_sentence": True, 437 "index": 0, 438 "streak": 0, 439 "best_streak": 0, 440 } 441 442 return jsonify({ 443 "session_id": session_id, 444 "words": all_tokens, 445 "total": len(all_tokens), 446 "is_sentence": True, 447 }) 448 449 450 _TRAILING_PUNCT = re.compile(r'[.!?,;:"\'\)\]\}]+$') 451 452 453 def _strip_word_punctuation(word): 454 """Strip trailing punctuation for stats tracking, but keep contractions intact.""" 455 return _TRAILING_PUNCT.sub("", word) 456 457 458 @app.route("/api/word/submit", methods=["POST"]) 459 @login_required 460 def api_word_submit(): 461 uid = current_user_id() 462 engine = get_engine(uid) 463 data = request.get_json() 464 session_id = data.get("session_id") 465 word = data.get("word", "") 466 typed = data.get("typed", "") 467 time_ms = data.get("time_ms", 0) 468 word_index = data.get("word_index", None) 469 470 if session_id not in session_store: 471 return jsonify({"error": "Session not found"}), 404 472 473 sess = session_store[session_id] 474 if sess.get("user_id") != uid: 475 return jsonify({"error": "Not your session"}), 403 476 477 correct = typed.strip() == word.strip() 478 stat_word = _strip_word_punctuation(word.strip()).lower() 479 strokes = engine.get_strokes(stat_word or word) 480 best_stroke = strokes[0] if strokes else "" 481 482 db.record_attempt(uid, session_id, stat_word or word, best_stroke, typed.strip(), correct, time_ms) 483 484 if correct: 485 sess["streak"] += 1 486 sess["best_streak"] = max(sess["best_streak"], sess["streak"]) 487 else: 488 sess["streak"] = 0 489 490 if word_index is not None: 491 sess["index"] = word_index + 1 492 493 db_session = db.get_session(session_id) 494 words_attempted = db_session["words_attempted"] if db_session else 1 495 words_correct = db_session["words_correct"] if db_session else (1 if correct else 0) 496 accuracy = words_correct / words_attempted if words_attempted > 0 else 0 497 498 word_chars = len(stat_word) if stat_word else len(typed.strip()) 499 wpm = (word_chars / 5) / (time_ms / 60000) if time_ms > 0 else 0 500 501 return jsonify({ 502 "correct": correct, 503 "expected": word, 504 "typed": typed.strip(), 505 "time_ms": time_ms, 506 "accuracy": accuracy, 507 "streak": sess["streak"], 508 "wpm": round(wpm, 1), 509 }) 510 511 512 @app.route("/api/session/end", methods=["POST"]) 513 @login_required 514 def api_session_end(): 515 uid = current_user_id() 516 data = request.get_json() 517 session_id = data.get("session_id") 518 519 if session_id is None: 520 return jsonify({"error": "session_id required"}), 400 521 522 db.end_session(session_id) 523 stats = db.get_session_stats(session_id) 524 if not stats: 525 return jsonify({"error": "Session not found"}), 404 526 527 attempts = db.get_session_attempts(session_id) 528 529 weakest = [] 530 for a in attempts: 531 if not a["correct"]: 532 weakest.append({ 533 "word": a["word"], 534 "typed": a.get("typed_text", ""), 535 "strokes_display": a["strokes"] or "", 536 "time_ms": a["time_ms"], 537 "correct": False, 538 }) 539 540 if len(weakest) < 5: 541 correct_attempts = sorted( 542 [a for a in attempts if a["correct"]], 543 key=lambda x: x["time_ms"], 544 reverse=True, 545 ) 546 for a in correct_attempts: 547 if len(weakest) >= 5: 548 break 549 weakest.append({ 550 "word": a["word"], 551 "typed": a.get("typed_text", ""), 552 "strokes_display": a["strokes"] or "", 553 "time_ms": a["time_ms"], 554 "correct": True, 555 }) 556 557 # Full per-word results for detailed accuracy review 558 all_results = [] 559 for a in attempts: 560 all_results.append({ 561 "word": a["word"], 562 "typed": a.get("typed_text", ""), 563 "correct": bool(a["correct"]), 564 "time_ms": a["time_ms"], 565 "strokes": a["strokes"] or "", 566 }) 567 568 total_chars = sum(len(_strip_word_punctuation(a["word"] or "")) for a in attempts) 569 total_time = stats["total_time_ms"] 570 avg_wpm = (total_chars / 5) / (total_time / 60000) if total_time > 0 else 0 571 572 session_store.pop(session_id, None) 573 574 return jsonify({ 575 "session_id": session_id, 576 "words_attempted": stats["words_attempted"], 577 "words_correct": stats["words_correct"], 578 "accuracy": stats["accuracy"], 579 "avg_wpm": round(avg_wpm, 1), 580 "total_time_ms": stats["total_time_ms"], 581 "best_streak": stats["best_streak"], 582 "weakest_words": weakest, 583 "all_results": all_results, 584 }) 585 586 587 # ── Settings API ── 588 589 590 @app.route("/api/settings", methods=["GET"]) 591 @login_required 592 def api_settings_get(): 593 return jsonify(db.get_user_settings(current_user_id())) 594 595 596 @app.route("/api/settings", methods=["POST"]) 597 @login_required 598 def api_settings_set(): 599 uid = current_user_id() 600 data = request.get_json() 601 if "words_per_session" in data: 602 val = data["words_per_session"] 603 if isinstance(val, int) and 10 <= val <= 200: 604 db.update_user_settings(uid, words_per_session=val) 605 return jsonify({"ok": True, "settings": db.get_user_settings(uid)}) 606 607 608 @app.route("/api/data/clear", methods=["POST"]) 609 @login_required 610 def api_data_clear(): 611 db.clear_user_data(current_user_id()) 612 return jsonify({"ok": True}) 613 614 615 # ── Dictionary Upload API ── 616 617 ALLOWED_DICT_EXTENSIONS = {".json", ".py"} 618 619 620 @app.route("/api/dictionaries/upload-config", methods=["POST"]) 621 @login_required 622 def api_upload_config(): 623 uid = current_user_id() 624 625 if "config" not in request.files: 626 return jsonify({"error": "No config file provided"}), 400 627 628 cfg_file = request.files["config"] 629 if not cfg_file.filename: 630 return jsonify({"error": "No file selected"}), 400 631 632 cfg_content = cfg_file.read().decode("utf-8", errors="replace") 633 entries = PloverEngine.parse_plover_config(cfg_content) 634 635 if not entries: 636 return jsonify({"error": "No dictionaries found in config. Make sure it has a [System: English Stenotype] section."}), 400 637 638 user_dir = _get_user_dict_dir(uid) 639 os.makedirs(user_dir, exist_ok=True) 640 641 cfg_path = os.path.join(user_dir, "plover.cfg") 642 with open(cfg_path, "w", encoding="utf-8") as f: 643 f.write(cfg_content) 644 645 db.clear_user_dictionaries(uid) 646 647 existing_files = set(os.listdir(user_dir)) if os.path.isdir(user_dir) else set() 648 649 result_entries = [] 650 for entry in entries: 651 filename = entry["filename"] 652 safe = secure_filename(filename) 653 if not safe: 654 continue 655 656 ext = os.path.splitext(safe)[1].lower() 657 if ext not in ALLOWED_DICT_EXTENSIONS: 658 continue 659 660 uploaded = safe in existing_files 661 entry_count = 0 662 if uploaded and ext == ".json": 663 try: 664 with open(os.path.join(user_dir, safe), "r", encoding="utf-8") as f: 665 entry_count = len(json.load(f)) 666 except Exception: 667 pass 668 669 dict_id = db.add_user_dictionary( 670 uid, safe, filename, entry["order"], entry_count 671 ) 672 673 result_entries.append({ 674 "id": dict_id, 675 "filename": safe, 676 "display_name": filename, 677 "order": entry["order"], 678 "enabled": entry["enabled"], 679 "uploaded": uploaded, 680 "entry_count": entry_count, 681 }) 682 683 if any(e["uploaded"] for e in result_entries): 684 reload_engine(uid) 685 686 return jsonify({"ok": True, "dictionaries": result_entries}) 687 688 689 @app.route("/api/dictionaries/upload-folder", methods=["POST"]) 690 @login_required 691 def api_upload_folder(): 692 """Receive plover.cfg + only the referenced dict files (pre-filtered client-side).""" 693 uid = current_user_id() 694 cfg_file = request.files.get("config") 695 if not cfg_file: 696 return jsonify({"error": "No plover.cfg provided"}), 400 697 698 cfg_content = cfg_file.read().decode("utf-8", errors="replace") 699 entries = PloverEngine.parse_plover_config(cfg_content) 700 if not entries: 701 return jsonify({"error": "No dictionaries found in plover.cfg."}), 400 702 703 user_dir = _get_user_dict_dir(uid) 704 os.makedirs(user_dir, exist_ok=True) 705 706 with open(os.path.join(user_dir, "plover.cfg"), "w", encoding="utf-8") as out: 707 out.write(cfg_content) 708 709 dict_files = request.files.getlist("dicts") 710 file_map = {} 711 for f in dict_files: 712 name = os.path.basename(f.filename or "") 713 if name: 714 file_map[name] = f 715 716 db.clear_user_dictionaries(uid) 717 718 result_entries = [] 719 saved = 0 720 for entry in entries: 721 filename = entry["filename"] 722 safe = secure_filename(filename) 723 if not safe: 724 continue 725 ext = os.path.splitext(safe)[1].lower() 726 if ext not in ALLOWED_DICT_EXTENSIONS: 727 continue 728 729 uploaded_file = file_map.get(filename) or file_map.get(safe) 730 found = False 731 entry_count = 0 732 if uploaded_file: 733 dest = os.path.join(user_dir, safe) 734 uploaded_file.save(dest) 735 found = True 736 saved += 1 737 if ext == ".json": 738 try: 739 with open(dest, "r", encoding="utf-8") as jf: 740 entry_count = len(json.load(jf)) 741 except Exception: 742 pass 743 744 dict_id = db.add_user_dictionary( 745 uid, safe, filename, entry["order"], entry_count 746 ) 747 result_entries.append({ 748 "id": dict_id, 749 "filename": safe, 750 "display_name": filename, 751 "order": entry["order"], 752 "enabled": entry["enabled"], 753 "uploaded": found, 754 "entry_count": entry_count, 755 }) 756 757 reload_engine(uid) 758 759 missing = [e for e in result_entries if not e["uploaded"]] 760 return jsonify({ 761 "ok": True, 762 "dictionaries": result_entries, 763 "saved": saved, 764 "missing": len(missing), 765 }) 766 767 768 @app.route("/api/dictionaries/upload", methods=["POST"]) 769 @login_required 770 def api_upload_dicts(): 771 uid = current_user_id() 772 user_dir = _get_user_dict_dir(uid) 773 os.makedirs(user_dir, exist_ok=True) 774 775 files = request.files.getlist("files") 776 if not files: 777 return jsonify({"error": "No files provided"}), 400 778 779 known = {d["filename"]: d for d in db.get_user_dictionaries(uid)} 780 781 uploaded = [] 782 for f in files: 783 if not f.filename: 784 continue 785 safe = secure_filename(f.filename) 786 if not safe: 787 continue 788 ext = os.path.splitext(safe)[1].lower() 789 if ext not in ALLOWED_DICT_EXTENSIONS: 790 continue 791 792 dest = os.path.join(user_dir, safe) 793 f.save(dest) 794 795 entry_count = 0 796 if ext == ".json": 797 try: 798 with open(dest, "r", encoding="utf-8") as jf: 799 entry_count = len(json.load(jf)) 800 except Exception: 801 pass 802 803 if safe in known: 804 db.update_dictionary_entry_count(known[safe]["id"], uid, entry_count) 805 else: 806 order = max((d.get("dict_order", 0) for d in known.values()), default=-1) + 1 807 db.add_user_dictionary(uid, safe, f.filename, order, entry_count) 808 809 uploaded.append({"filename": safe, "entry_count": entry_count}) 810 811 reload_engine(uid) 812 813 all_dicts = db.get_user_dictionaries(uid) 814 existing_files = set(os.listdir(user_dir)) if os.path.isdir(user_dir) else set() 815 dict_list = [] 816 for d in all_dicts: 817 dict_list.append({ 818 "id": d["id"], 819 "filename": d["filename"], 820 "display_name": d["display_name"], 821 "order": d["dict_order"], 822 "enabled": bool(d["enabled"]), 823 "uploaded": d["filename"] in existing_files, 824 "entry_count": d["entry_count"], 825 }) 826 827 return jsonify({"ok": True, "uploaded": uploaded, "dictionaries": dict_list}) 828 829 830 @app.route("/api/dictionaries", methods=["GET"]) 831 @login_required 832 def api_get_dicts(): 833 uid = current_user_id() 834 all_dicts = db.get_user_dictionaries(uid) 835 user_dir = _get_user_dict_dir(uid) 836 existing_files = set(os.listdir(user_dir)) if os.path.isdir(user_dir) else set() 837 838 result = [] 839 for d in all_dicts: 840 result.append({ 841 "id": d["id"], 842 "filename": d["filename"], 843 "display_name": d["display_name"], 844 "order": d["dict_order"], 845 "enabled": bool(d["enabled"]), 846 "uploaded": d["filename"] in existing_files, 847 "entry_count": d["entry_count"], 848 }) 849 return jsonify(result) 850 851 852 @app.route("/api/dictionaries/<int:dict_id>/toggle", methods=["POST"]) 853 @login_required 854 def api_toggle_dict(dict_id): 855 uid = current_user_id() 856 data = request.get_json() 857 enabled = data.get("enabled", True) 858 db.toggle_dictionary(dict_id, uid, enabled) 859 reload_engine(uid) 860 return jsonify({"ok": True}) 861 862 863 @app.route("/api/dictionaries/reorder", methods=["POST"]) 864 @login_required 865 def api_reorder_dicts(): 866 uid = current_user_id() 867 data = request.get_json() 868 order = data.get("order", []) 869 if not order: 870 return jsonify({"error": "No order provided"}), 400 871 with db._get_conn() as conn: 872 for i, dict_id in enumerate(order): 873 conn.execute( 874 "UPDATE user_dictionaries SET dict_order = ? WHERE id = ? AND user_id = ?", 875 (i, dict_id, uid), 876 ) 877 reload_engine(uid) 878 return jsonify({"ok": True}) 879 880 881 @app.route("/api/dictionaries/<int:dict_id>/replace", methods=["POST"]) 882 @login_required 883 def api_replace_dict(dict_id): 884 uid = current_user_id() 885 dicts = {d["id"]: d for d in db.get_user_dictionaries(uid)} 886 if dict_id not in dicts: 887 return jsonify({"error": "Dictionary not found"}), 404 888 889 f = request.files.get("file") 890 if not f or not f.filename: 891 return jsonify({"error": "No file provided"}), 400 892 893 ext = os.path.splitext(f.filename)[1].lower() 894 if ext not in ALLOWED_DICT_EXTENSIONS: 895 return jsonify({"error": "Only .json and .py files are allowed"}), 400 896 897 old = dicts[dict_id] 898 user_dir = _get_user_dict_dir(uid) 899 os.makedirs(user_dir, exist_ok=True) 900 901 old_path = os.path.join(user_dir, old["filename"]) 902 if os.path.exists(old_path): 903 os.remove(old_path) 904 905 safe = secure_filename(f.filename) 906 dest = os.path.join(user_dir, safe) 907 f.save(dest) 908 909 entry_count = 0 910 if ext == ".json": 911 try: 912 with open(dest, "r", encoding="utf-8") as jf: 913 entry_count = len(json.load(jf)) 914 except Exception: 915 pass 916 917 with db._get_conn() as conn: 918 conn.execute( 919 "UPDATE user_dictionaries SET filename = ?, display_name = ?, entry_count = ? WHERE id = ? AND user_id = ?", 920 (safe, f.filename, entry_count, dict_id, uid), 921 ) 922 923 reload_engine(uid) 924 return jsonify({ 925 "ok": True, 926 "id": dict_id, 927 "filename": safe, 928 "display_name": f.filename, 929 "entry_count": entry_count, 930 }) 931 932 933 @app.route("/api/dictionaries/<int:dict_id>", methods=["DELETE"]) 934 @login_required 935 def api_delete_dict(dict_id): 936 uid = current_user_id() 937 filename = db.delete_dictionary(dict_id, uid) 938 if filename: 939 fpath = os.path.join(_get_user_dict_dir(uid), filename) 940 if os.path.exists(fpath): 941 os.remove(fpath) 942 reload_engine(uid) 943 return jsonify({"ok": True}) 944 945 946 @app.route("/api/dictionaries/clear", methods=["POST"]) 947 @login_required 948 def api_clear_dicts(): 949 uid = current_user_id() 950 user_dir = _get_user_dict_dir(uid) 951 db.clear_user_dictionaries(uid) 952 if os.path.isdir(user_dir): 953 shutil.rmtree(user_dir) 954 reload_engine(uid) 955 return jsonify({"ok": True}) 956 957 958 # ── Stats API ── 959 960 961 @app.route("/api/stats/overview") 962 @login_required 963 def api_stats_overview(): 964 uid = current_user_id() 965 raw = db.get_overview_stats(uid) 966 total_ms = raw.get("total_time_ms", 0) 967 return jsonify({ 968 "total_sessions": raw.get("total_sessions", 0), 969 "total_words": raw.get("total_words_practiced", 0), 970 "words_mastered": raw.get("words_mastered", 0), 971 "avg_accuracy": raw.get("overall_accuracy", 0) * 100, 972 "avg_wpm": 0, 973 "total_time_seconds": total_ms // 1000, 974 "best_streak": raw.get("current_streak", 0), 975 }) 976 977 978 @app.route("/api/stats/daily") 979 @login_required 980 def api_stats_daily(): 981 uid = current_user_id() 982 days = request.args.get("days", 30, type=int) 983 raw = db.get_daily_stats(uid, days=days) 984 result = [] 985 for d in raw: 986 wp = d.get("words_practiced", 0) 987 wc = d.get("words_correct", 0) 988 result.append({ 989 "date": d.get("date", ""), 990 "words_total": wp, 991 "words_correct": wc, 992 "accuracy": (wc / wp * 100) if wp > 0 else 0, 993 "avg_wpm": 0, 994 "sessions": d.get("sessions", 0), 995 }) 996 return jsonify(result) 997 998 999 @app.route("/api/stats/heatmap") 1000 @login_required 1001 def api_stats_heatmap(): 1002 return jsonify(db.get_heatmap_data(current_user_id())) 1003 1004 1005 @app.route("/api/stats/words") 1006 @login_required 1007 def api_stats_words(): 1008 uid = current_user_id() 1009 sort_by = request.args.get("sort", "accuracy") 1010 order = request.args.get("order", "asc") 1011 limit = request.args.get("limit", 50, type=int) 1012 1013 all_stats = db.get_word_stats(uid) 1014 result = [] 1015 for ws in all_stats: 1016 att = ws.get("attempts", 0) 1017 cor = ws.get("correct", 0) 1018 result.append({ 1019 "word": ws.get("word", ""), 1020 "strokes": ws.get("strokes", ""), 1021 "attempts": att, 1022 "correct": cor, 1023 "accuracy": (cor / att * 100) if att > 0 else 0, 1024 "avg_time": ws.get("avg_time_ms", 0), 1025 }) 1026 1027 reverse = order == "desc" 1028 if sort_by in ("accuracy", "attempts", "correct", "avg_time"): 1029 result.sort(key=lambda x: x.get(sort_by, 0), reverse=reverse) 1030 elif sort_by == "word": 1031 result.sort(key=lambda x: x.get("word", ""), reverse=reverse) 1032 else: 1033 result.sort(key=lambda x: x.get("accuracy", 0), reverse=reverse) 1034 1035 return jsonify(result[:limit]) 1036 1037 1038 @app.route("/api/stats/sessions") 1039 @login_required 1040 def api_stats_sessions(): 1041 uid = current_user_id() 1042 limit = request.args.get("limit", 20, type=int) 1043 raw = db.get_all_sessions(uid, limit=limit) 1044 result = [] 1045 for s in raw: 1046 wa = s.get("words_attempted", 0) 1047 wc = s.get("words_correct", 0) 1048 total_ms = s.get("total_time_ms", 0) 1049 total_chars = wa * 5 1050 wpm = (total_chars / 5) / (total_ms / 60000) if total_ms > 0 else 0 1051 started = s.get("started_at") or "" 1052 result.append({ 1053 "date": started[:10] if started else "", 1054 "date_str": started[:10] if started else "", 1055 "mode": s.get("mode", ""), 1056 "words_attempted": wa, 1057 "words_correct": wc, 1058 "accuracy": (wc / wa * 100) if wa > 0 else 0, 1059 "wpm": round(wpm, 1), 1060 "duration_seconds": total_ms // 1000, 1061 }) 1062 return jsonify(result) 1063 1064 1065 @app.route("/api/lessons") 1066 @login_required 1067 def api_lessons(): 1068 uid = current_user_id() 1069 lessons = get_lesson_list() 1070 for l in lessons: 1071 lesson_def = get_lesson_by_id(l["id"]) 1072 if lesson_def: 1073 word_list = get_lesson_word_list(lesson_def) 1074 if word_list: 1075 prog = db.get_lesson_progress(uid, word_list) 1076 l["progress"] = { 1077 "total_words": prog["total_words"], 1078 "practiced": prog["practiced"], 1079 "mastered": prog["mastered"], 1080 "avg_wpm": prog["avg_wpm"], 1081 "avg_accuracy": prog["avg_accuracy"], 1082 } 1083 return jsonify(lessons) 1084 1085 1086 @app.route("/api/lessons/<lesson_id>/progress") 1087 @login_required 1088 def api_lesson_progress(lesson_id): 1089 uid = current_user_id() 1090 lesson = get_lesson_by_id(lesson_id) 1091 if not lesson: 1092 return jsonify({"error": "Lesson not found"}), 404 1093 word_list = get_lesson_word_list(lesson) 1094 if not word_list: 1095 return jsonify({"error": "No word list for this lesson"}), 400 1096 progress = db.get_lesson_progress(uid, word_list) 1097 return jsonify(progress) 1098 1099 1100 # ── Helpers ── 1101 1102 1103 def _select_words(uid, engine, mode, count): 1104 has_words = bool(engine.practice_words) 1105 1106 if mode == "common": 1107 if has_words: 1108 words = [w for w in TOP_WORDS[:500] if engine.word_exists(w)] 1109 else: 1110 words = list(TOP_WORDS[:500]) 1111 return words[:count] 1112 elif mode == "random": 1113 if has_words: 1114 words = engine.practice_words 1115 return random.sample(words, min(count, len(words))) 1116 return list(TOP_WORDS[:count]) 1117 elif mode == "new": 1118 practiced = db.get_practiced_words(uid) 1119 if has_words: 1120 available = [w for w in engine.practice_words if w not in practiced] 1121 else: 1122 available = [w for w in TOP_WORDS[:500] if w not in practiced] 1123 return random.sample(available, min(count, len(available))) if available else [] 1124 elif mode == "weak": 1125 word_stats = db.get_word_stats(uid) 1126 weak = [] 1127 for ws in word_stats: 1128 if ws["attempts"] >= 3: 1129 accuracy = ws["correct"] / ws["attempts"] 1130 weak.append((accuracy, ws["word"])) 1131 weak.sort() 1132 words = [w for _, w in weak[:count]] 1133 if len(words) < count and has_words: 1134 extra = random.sample(engine.practice_words, min(count - len(words), len(engine.practice_words))) 1135 words.extend(extra) 1136 return words[:count] 1137 elif mode == "review": 1138 words = db.get_words_due_for_review(uid, count) 1139 if len(words) < count and has_words: 1140 extra = random.sample(engine.practice_words, min(count - len(words), len(engine.practice_words))) 1141 words.extend(extra) 1142 return words[:count] 1143 elif mode == "speed": 1144 if has_words: 1145 words = engine.single_stroke_words 1146 return random.sample(words, min(count, len(words))) 1147 return list(TOP_WORDS[:count]) 1148 elif mode == "phrases": 1149 phrases = engine.all_phrases + engine.generate_phrasing_phrases(200) 1150 return random.sample(phrases, min(count, len(phrases))) if phrases else [] 1151 elif mode == "sentences": 1152 return get_sentences("intermediate", count=10) 1153 else: 1154 if has_words: 1155 words = engine.practice_words 1156 return random.sample(words, min(count, len(words))) 1157 return list(TOP_WORDS[:count]) 1158 1159 1160 print("Initializing StenoDojo...") 1161 db_check = Database() 1162 from sentence_generator import _load_corpus 1163 _load_corpus() 1164 print("Ready!\n") 1165 1166 if __name__ == "__main__": 1167 app.run(debug=True, port=5000)