# utils/db.py (top of file) import os import json import certifi from contextlib import contextmanager from datetime import date, timedelta # password hashing import bcrypt # --- Feature flag: DB off by default in frontend Space --- DISABLE_DB = os.getenv("DISABLE_DB", "1") == "1" # Import mysql connector only when DB is enabled MYSQL_AVAILABLE = False if not DISABLE_DB: try: import mysql.connector # provided by mysql-connector-python from mysql.connector import Error # noqa: F401 MYSQL_AVAILABLE = True except Exception: MYSQL_AVAILABLE = False # will raise a friendly error if used def _db_disabled_error(): raise RuntimeError( "Database access is disabled in this frontend (DISABLE_DB=1). " "Route calls through your backend Space instead." ) def get_db_connection(): if DISABLE_DB or not MYSQL_AVAILABLE: _db_disabled_error() ssl_enabled = os.getenv("TIDB_ENABLE_SSL", "false").lower() == "true" ssl_ca = certifi.where() if ssl_enabled else None return mysql.connector.connect( host=os.getenv("TIDB_HOST"), port=int(os.getenv("TIDB_PORT", 4000)), user=os.getenv("TIDB_USER"), password=os.getenv("TIDB_PASSWORD"), database=os.getenv("TIDB_DATABASE", "fin_ed_agentic"), ssl_ca=ssl_ca, ssl_verify_cert=ssl_enabled, autocommit=True, ) @contextmanager def cursor(dict_rows=True): if DISABLE_DB or not MYSQL_AVAILABLE: _db_disabled_error() conn = get_db_connection() try: cur = conn.cursor(dictionary=dict_rows) yield cur conn.commit() finally: cur.close() conn.close() # password hashing import bcrypt # ----------- label <-> slug mappers for UI selects ----------- COUNTRY_SLUG = { "Jamaica": "jamaica", "USA": "usa", "UK": "uk", "India": "india", "Canada": "canada", "Other": "other", "N/A": "na" } LEVEL_SLUG = { "Beginner": "beginner", "Intermediate": "intermediate", "Advanced": "advanced", "N/A": "na" } ROLE_SLUG = {"Student": "student", "Teacher": "teacher"} def _slug(s: str) -> str: return (s or "").strip().lower() def hash_password(plain: str) -> bytes: return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt()) def verify_password(plain: str, hashed: bytes | None) -> bool: if not plain or not hashed: return False try: return bcrypt.checkpw(plain.encode("utf-8"), hashed) except Exception: return False def _ensure_na_slugs(): """ Make sure 'na' exists in countries/levels for teacher rows. Harmless if already present. """ with cursor() as cur: cur.execute("INSERT IGNORE INTO countries(slug,label) VALUES('na','N/A')") cur.execute("INSERT IGNORE INTO levels(slug,label) VALUES('na','N/A')") # def get_db_connection(): # ssl_enabled = os.getenv("TIDB_ENABLE_SSL", "false").lower() == "true" # ssl_ca = certifi.where() if ssl_enabled else None # return mysql.connector.connect( # host=os.getenv("TIDB_HOST"), # port=int(os.getenv("TIDB_PORT", 4000)), # user=os.getenv("TIDB_USER"), # password=os.getenv("TIDB_PASSWORD"), # database=os.getenv("TIDB_DATABASE", "agenticfinance"), # ssl_ca=ssl_ca, # ssl_verify_cert=ssl_enabled, # autocommit=True, # ) @contextmanager def cursor(dict_rows=True): conn = get_db_connection() try: cur = conn.cursor(dictionary=dict_rows) yield cur conn.commit() finally: cur.close() conn.close() # ---------- USERS ---------- def create_user(name:str, email:str, country:str, level:str, role:str): slug = lambda s: s.strip().lower() with cursor() as cur: cur.execute(""" INSERT INTO users(name,email,country_slug,level_slug,role_slug) VALUES (%s,%s,%s,%s,%s) """, (name, email.strip().lower(), slug(country), slug(level), slug(role))) return True # role-specific creators def create_student(*, name:str, email:str, password:str, level_label:str, country_label:str) -> bool: """ level_label/country_label are UI labels (e.g., 'Beginner', 'Jamaica'). """ level_slug = LEVEL_SLUG.get(level_label, _slug(level_label)) country_slug = COUNTRY_SLUG.get(country_label, _slug(country_label)) with cursor() as cur: cur.execute(""" INSERT INTO users (name,email,password_hash,title,country_slug,level_slug,role_slug) VALUES (%s,%s,%s,NULL,%s,%s,'student') """, (name.strip(), email.strip().lower(), hash_password(password), country_slug, level_slug)) return True def create_teacher(*, title:str, name:str, email:str, password:str) -> bool: """ Teachers do not provide level/country; we store 'na' for both. """ _ensure_na_slugs() with cursor() as cur: cur.execute(""" INSERT INTO users (title,name,email,password_hash,country_slug,level_slug,role_slug) VALUES (%s,%s,%s,%s,'na','na','teacher') """, (title.strip(), name.strip(), email.strip().lower(), hash_password(password))) return True def get_user_by_email(email:str): with cursor() as cur: cur.execute(""" SELECT u.user_id, u.title, u.name, u.email, u.password_hash, u.country_slug, c.label AS country, u.level_slug, l.label AS level, u.role_slug, r.label AS role FROM users u JOIN countries c ON c.slug = u.country_slug JOIN levels l ON l.slug = u.level_slug JOIN roles r ON r.slug = u.role_slug WHERE u.email=%s LIMIT 1 """, (email.strip().lower(),)) u = cur.fetchone() if not u: return None u["role"] = "Teacher" if u["role_slug"] == "teacher" else "Student" return u def check_password(email: str, plain_password: str) -> dict | None: """ Returns the user dict if password is correct, else None. """ user = get_user_by_email(email) if not user: return None if verify_password(plain_password, user.get("password_hash")): return user return None # ---------- CLASSES ---------- import random, string def _code(): return "".join(random.choices(string.ascii_uppercase + string.digits, k=6)) def create_class(teacher_id:int, name:str): # ensure unique code for _ in range(20): code = _code() with cursor() as cur: cur.execute("SELECT 1 FROM classes WHERE code=%s", (code,)) if not cur.fetchone(): cur.execute("INSERT INTO classes(teacher_id,name,code) VALUES(%s,%s,%s)", (teacher_id, name, code)) cur.execute("SELECT LAST_INSERT_ID() AS id") cid = cur.fetchone()["id"] return {"class_id": cid, "code": code} raise RuntimeError("Could not generate unique class code") def join_class_by_code(student_id:int, code:str): with cursor() as cur: cur.execute("SELECT class_id FROM classes WHERE code=%s", (code.strip().upper(),)) row = cur.fetchone() if not row: raise ValueError("Invalid class code") cur.execute("INSERT IGNORE INTO class_students(class_id,student_id) VALUES(%s,%s)", (row["class_id"], student_id)) return row["class_id"] def list_students_in_class(class_id:int): with cursor() as cur: cur.execute(""" SELECT u.user_id, u.name, u.email, u.level_slug, cs.joined_at, -- <- show true join date u.created_at FROM class_students cs JOIN users u ON u.user_id = cs.student_id WHERE cs.class_id = %s ORDER BY u.name """, (class_id,)) return cur.fetchall() def class_analytics(class_id:int): """ Returns: class_avg -> 0..1 average quiz score for the class (from v_class_stats) total_xp -> sum of xp_log.delta for students in this class lessons_completed -> count of completed lesson_progress entries for lessons assigned to this class """ out = {"class_avg": 0.0, "total_xp": 0, "lessons_completed": 0} with cursor() as cur: # class average from view cur.execute("SELECT class_avg FROM v_class_stats WHERE class_id=%s", (class_id,)) row = cur.fetchone() if row: out["class_avg"] = float(row["class_avg"] or 0) # total XP for all students in this class cur.execute(""" SELECT COALESCE(SUM(x.delta),0) AS total_xp FROM xp_log x JOIN class_students cs ON cs.student_id = x.user_id WHERE cs.class_id = %s """, (class_id,)) out["total_xp"] = int((cur.fetchone() or {"total_xp": 0})["total_xp"]) # lessons completed that were actually assigned to this class cur.execute(""" SELECT COUNT(*) AS n FROM lesson_progress lp JOIN class_students cs ON cs.student_id = lp.user_id JOIN assignments a ON a.lesson_id = lp.lesson_id WHERE cs.class_id = %s AND a.class_id = %s AND lp.status = 'completed' """, (class_id, class_id)) out["lessons_completed"] = int((cur.fetchone() or {"n": 0})["n"]) return out # ---------- Teacher dash for real time data - Class Helpers ---------- def class_content_counts(class_id:int): # counts of distinct lessons and quizzes assigned to this class with cursor() as cur: cur.execute(""" SELECT COUNT(DISTINCT lesson_id) AS lessons, COUNT(DISTINCT quiz_id) AS quizzes FROM assignments WHERE class_id=%s """, (class_id,)) row = cur.fetchone() or {"lessons": 0, "quizzes": 0} return row def list_class_assignments(class_id:int): with cursor() as cur: cur.execute(""" SELECT a.assignment_id, a.created_at, l.lesson_id, l.title, l.subject, l.level, a.quiz_id FROM assignments a JOIN lessons l ON l.lesson_id = a.lesson_id WHERE a.class_id=%s ORDER BY a.created_at DESC """, (class_id,)) return cur.fetchall() def list_classes_by_teacher(teacher_id:int): with cursor() as cur: cur.execute(""" SELECT s.*, c.code FROM v_class_stats s JOIN classes c USING (class_id) WHERE s.teacher_id=%s ORDER BY c.created_at DESC """, (teacher_id,)) return cur.fetchall() def get_class(class_id:int): with cursor() as cur: cur.execute("SELECT class_id, name, code, teacher_id FROM classes WHERE class_id=%s", (class_id,)) return cur.fetchone() def class_student_metrics(class_id: int): """ Returns one row per student in the class with: name, email, joined_at, lessons_completed, total_assigned_lessons, avg_score (0..1), streak_days, total_xp """ with cursor() as cur: cur.execute(""" /* total assigned lessons for the class */ WITH assigned AS ( SELECT DISTINCT lesson_id FROM assignments WHERE class_id = %s ) SELECT cs.student_id, u.name, u.email, cs.joined_at, /* lessons completed by this student that were assigned to this class */ COALESCE( (SELECT COUNT(*) FROM lesson_progress lp WHERE lp.user_id = cs.student_id AND lp.status = 'completed' AND lp.lesson_id IN (SELECT lesson_id FROM assigned) ), 0 ) AS lessons_completed, /* total lessons assigned to this class */ (SELECT COUNT(*) FROM assigned) AS total_assigned_lessons, /* average quiz score only for submissions tied to this class */ COALESCE(sc.avg_score, 0) AS avg_score, /* streak days from streaks table */ COALESCE(str.days, 0) AS streak_days, /* total XP across the app */ COALESCE(xp.total_xp, 0) AS total_xp FROM class_students cs JOIN users u ON u.user_id = cs.student_id LEFT JOIN ( SELECT s.student_id, AVG(s.score * 1.0 / NULLIF(s.total,0)) AS avg_score FROM submissions s JOIN assignments a ON a.assignment_id = s.assignment_id WHERE a.class_id = %s GROUP BY s.student_id ) sc ON sc.student_id = cs.student_id LEFT JOIN streaks str ON str.user_id = cs.student_id LEFT JOIN (SELECT user_id, SUM(delta) AS total_xp FROM xp_log GROUP BY user_id) xp ON xp.user_id = cs.student_id WHERE cs.class_id = %s ORDER BY u.name; """, (class_id, class_id, class_id)) return cur.fetchall() def level_from_xp(total_xp: int) -> int: try: xp = int(total_xp or 0) except Exception: xp = 0 return 1 + xp // 500 def list_classes_for_student(student_id: int): with cursor() as cur: cur.execute(""" SELECT c.class_id, c.name, c.code, c.teacher_id, t.name AS teacher_name, cs.joined_at FROM class_students cs JOIN classes c ON c.class_id = cs.class_id JOIN users t ON t.user_id = c.teacher_id WHERE cs.student_id = %s ORDER BY cs.joined_at DESC """, (student_id,)) return cur.fetchall() def leave_class(student_id: int, class_id: int): with cursor() as cur: cur.execute("DELETE FROM class_students WHERE student_id=%s AND class_id=%s", (student_id, class_id)) return True def student_class_progress(student_id: int, class_id: int): """ Per-student view of progress inside ONE class. Returns: dict(overall_progress 0..1, lessons_completed int, total_assigned_lessons int, avg_score 0..1) """ with cursor() as cur: # total distinct lessons assigned to this class cur.execute("SELECT COUNT(DISTINCT lesson_id) AS n FROM assignments WHERE class_id=%s", (class_id,)) total_assigned = int((cur.fetchone() or {"n": 0})["n"]) # lessons completed among the class's assigned lessons cur.execute(""" WITH assigned AS (SELECT DISTINCT lesson_id FROM assignments WHERE class_id = %s) SELECT COUNT(*) AS n FROM lesson_progress lp WHERE lp.user_id = %s AND lp.status = 'completed' AND lp.lesson_id IN (SELECT lesson_id FROM assigned) """, (class_id, student_id)) completed = int((cur.fetchone() or {"n": 0})["n"]) # student’s avg quiz score but only for submissions tied to this class cur.execute(""" SELECT AVG(s.score * 1.0 / NULLIF(s.total,0)) AS avg_ratio FROM submissions s JOIN assignments a ON a.assignment_id = s.assignment_id WHERE a.class_id = %s AND s.student_id = %s """, (class_id, student_id)) avg_score = float((cur.fetchone() or {"avg_ratio": 0.0})["avg_ratio"] or 0.0) overall = (completed / float(total_assigned)) if total_assigned else 0.0 return dict( overall_progress=overall, lessons_completed=completed, total_assigned_lessons=total_assigned, avg_score=avg_score ) def student_assignments_for_class(student_id: int, class_id: int): """ All assignments in a class, annotated with THIS student's status/progress and (if applicable) their quiz score for that assignment. Deduplicates by lesson_id (keeps the most recent assignment per lesson). """ with cursor() as cur: cur.execute(""" SELECT a.assignment_id, a.lesson_id, l.title, l.subject, l.level, a.quiz_id, a.due_at, COALESCE(lp.status,'not_started') AS status, lp.current_pos, /* student's latest submission on this assignment (if any) */ (SELECT MAX(s.submitted_at) FROM submissions s WHERE s.assignment_id = a.assignment_id AND s.student_id = %s) AS last_submit_at, (SELECT s2.score FROM submissions s2 WHERE s2.assignment_id = a.assignment_id AND s2.student_id = %s ORDER BY s2.submitted_at DESC LIMIT 1) AS score, (SELECT s3.total FROM submissions s3 WHERE s3.assignment_id = a.assignment_id AND s3.student_id = %s ORDER BY s3.submitted_at DESC LIMIT 1) AS total FROM ( SELECT a.*, ROW_NUMBER() OVER ( PARTITION BY a.lesson_id ORDER BY a.created_at DESC, a.assignment_id DESC ) AS rn FROM assignments a WHERE a.class_id = %s ) AS a JOIN lessons l ON l.lesson_id = a.lesson_id LEFT JOIN lesson_progress lp ON lp.user_id = %s AND lp.lesson_id = a.lesson_id WHERE a.rn = 1 ORDER BY a.created_at DESC """, (student_id, student_id, student_id, class_id, student_id)) return cur.fetchall() def update_quiz(quiz_id:int, teacher_id:int, title:str, items:list[dict], settings:dict|None=None) -> bool: with cursor() as cur: # only the teacher who owns the linked lesson can edit cur.execute(""" SELECT 1 FROM quizzes q JOIN lessons l ON l.lesson_id = q.lesson_id WHERE q.quiz_id = %s AND l.teacher_id = %s LIMIT 1 """, (quiz_id, teacher_id)) if not cur.fetchone(): return False cur.execute("UPDATE quizzes SET title=%s, settings=%s WHERE quiz_id=%s", (title, json.dumps(settings or {}), quiz_id)) cur.execute("DELETE FROM quiz_items WHERE quiz_id=%s", (quiz_id,)) for i, it in enumerate(items, start=1): cur.execute(""" INSERT INTO quiz_items(quiz_id, position, question, options, answer_key, points) VALUES (%s, %s, %s, %s, %s, %s) """, ( quiz_id, i, it["question"], json.dumps(it.get("options", [])), json.dumps(it.get("answer_key")), # single letter as JSON string int(it.get("points", 1)) )) return True def class_weekly_activity(class_id:int): start = date.today() - timedelta(days=6) with cursor() as cur: cur.execute(""" SELECT DATE(lp.last_accessed) d, COUNT(*) n FROM lesson_progress lp JOIN class_students cs ON cs.student_id = lp.user_id WHERE cs.class_id=%s AND lp.last_accessed >= %s GROUP BY DATE(lp.last_accessed) """, (class_id, start)) lessons = {r["d"]: r["n"] for r in cur.fetchall()} cur.execute(""" SELECT DATE(s.submitted_at) d, COUNT(*) n FROM submissions s JOIN assignments a ON a.assignment_id = s.assignment_id WHERE a.class_id=%s AND s.submitted_at >= %s GROUP BY DATE(s.submitted_at) """, (class_id, start)) quizzes = {r["d"]: r["n"] for r in cur.fetchall()} cur.execute(""" SELECT DATE(g.started_at) d, COUNT(*) n FROM game_sessions g JOIN class_students cs ON cs.student_id = g.user_id WHERE cs.class_id=%s AND g.started_at >= %s GROUP BY DATE(g.started_at) """, (class_id, start)) games = {r["d"]: r["n"] for r in cur.fetchall()} out = [] for i in range(7): d = start + timedelta(days=i) out.append({ "date": d, "lessons": lessons.get(d, 0), "quizzes": quizzes.get(d, 0), "games": games.get(d, 0), }) return out def update_lesson(lesson_id:int, teacher_id:int, title:str, description:str, subject:str, level_slug:str, sections:list[dict]) -> bool: with cursor() as cur: # ownership check cur.execute("SELECT 1 FROM lessons WHERE lesson_id=%s AND teacher_id=%s", (lesson_id, teacher_id)) if not cur.fetchone(): return False cur.execute(""" UPDATE lessons SET title=%s, description=%s, subject=%s, level=%s WHERE lesson_id=%s AND teacher_id=%s """, (title, description, subject, level_slug, lesson_id, teacher_id)) # simplest and safest: rebuild sections in order cur.execute("DELETE FROM lesson_sections WHERE lesson_id=%s", (lesson_id,)) for i, sec in enumerate(sections, start=1): cur.execute(""" INSERT INTO lesson_sections(lesson_id,position,title,content) VALUES(%s,%s,%s,%s) """, (lesson_id, i, sec.get("title"), sec.get("content"))) return True # --- Class progress overview (overall progress, quiz performance, totals) def class_progress_overview(class_id: int): """ Returns: { "overall_progress": 0..1, "quiz_performance": 0..1, "lessons_completed": int, "class_xp": int } """ with cursor() as cur: # total distinct lessons assigned to this class cur.execute("SELECT COUNT(DISTINCT lesson_id) AS n FROM assignments WHERE class_id=%s", (class_id,)) total_assigned = int((cur.fetchone() or {"n": 0})["n"]) # number of enrolled students cur.execute("SELECT COUNT(*) AS n FROM class_students WHERE class_id=%s", (class_id,)) num_students = int((cur.fetchone() or {"n": 0})["n"]) # sum of completed lessons by all students (for assigned lessons) cur.execute(""" WITH assigned AS ( SELECT DISTINCT lesson_id FROM assignments WHERE class_id = %s ), enrolled AS ( SELECT student_id FROM class_students WHERE class_id = %s ), per_student AS ( SELECT e.student_id, COUNT(DISTINCT CASE WHEN lp.status='completed' AND lp.lesson_id IN (SELECT lesson_id FROM assigned) THEN lp.lesson_id END) AS completed FROM enrolled e LEFT JOIN lesson_progress lp ON lp.user_id = e.student_id GROUP BY e.student_id ) SELECT COALESCE(SUM(completed),0) AS total_completed FROM per_student """, (class_id, class_id)) total_completed = int((cur.fetchone() or {"total_completed": 0})["total_completed"] or 0) # quiz performance: average percentage for submissions tied to this class cur.execute(""" SELECT AVG(s.score * 1.0 / NULLIF(s.total,0)) AS avg_ratio FROM submissions s JOIN assignments a ON a.assignment_id = s.assignment_id WHERE a.class_id = %s """, (class_id,)) quiz_perf_row = cur.fetchone() or {"avg_ratio": 0} quiz_perf = float(quiz_perf_row["avg_ratio"] or 0) # total class XP (sum of xp for enrolled students) cur.execute(""" SELECT COALESCE(SUM(x.delta),0) AS xp FROM xp_log x WHERE x.user_id IN (SELECT student_id FROM class_students WHERE class_id=%s) """, (class_id,)) class_xp = int((cur.fetchone() or {"xp": 0})["xp"] or 0) if total_assigned and num_students: denominator = float(total_assigned * num_students) overall = float(total_completed) / denominator else: overall = 0.0 return dict( overall_progress=float(overall), quiz_performance=float(quiz_perf), lessons_completed=int(total_completed), class_xp=int(class_xp), ) # --- Recent student activity with total_xp for level badge def class_recent_activity(class_id:int, limit:int=6, days:int=30): """ Returns latest activity rows with fields: ts, kind('lesson'|'quiz'|'game'), student_id, student_name, item_title, extra, total_xp """ with cursor() as cur: cur.execute(f""" WITH enrolled AS ( SELECT student_id FROM class_students WHERE class_id = %s ), xp AS ( SELECT user_id, COALESCE(SUM(delta),0) AS total_xp FROM xp_log GROUP BY user_id ) SELECT * FROM ( /* completed lessons */ SELECT lp.last_accessed AS ts, 'lesson' AS kind, u.user_id AS student_id, u.name AS student_name, l.title AS item_title, NULL AS extra, COALESCE(xp.total_xp,0) AS total_xp FROM lesson_progress lp JOIN enrolled e ON e.student_id = lp.user_id JOIN users u ON u.user_id = lp.user_id JOIN lessons l ON l.lesson_id = lp.lesson_id LEFT JOIN xp ON xp.user_id = u.user_id WHERE lp.status = 'completed' AND lp.last_accessed >= NOW() - INTERVAL {days} DAY UNION ALL /* quiz submissions */ SELECT s.submitted_at AS ts, 'quiz' AS kind, u.user_id AS student_id, u.name AS student_name, l.title AS item_title, CONCAT(ROUND(s.score*100.0/NULLIF(s.total,0)),'%') AS extra, COALESCE(xp.total_xp,0) AS total_xp FROM submissions s JOIN assignments a ON a.assignment_id = s.assignment_id AND a.class_id = %s JOIN users u ON u.user_id = s.student_id JOIN lessons l ON l.lesson_id = a.lesson_id LEFT JOIN xp ON xp.user_id = u.user_id WHERE s.submitted_at >= NOW() - INTERVAL {days} DAY UNION ALL /* games */ SELECT g.started_at AS ts, 'game' AS kind, u.user_id AS student_id, u.name AS student_name, g.game_slug AS item_title, NULL AS extra, COALESCE(xp.total_xp,0) AS total_xp FROM game_sessions g JOIN enrolled e ON e.student_id = g.user_id JOIN users u ON u.user_id = g.user_id LEFT JOIN xp ON xp.user_id = u.user_id WHERE g.started_at >= NOW() - INTERVAL {days} DAY ) x ORDER BY ts DESC LIMIT %s """, (class_id, class_id, limit)) return cur.fetchall() def list_quizzes_by_teacher(teacher_id:int): with cursor() as cur: cur.execute(""" SELECT q.quiz_id, q.title, q.created_at, l.title AS lesson_title, (SELECT COUNT(*) FROM quiz_items qi WHERE qi.quiz_id=q.quiz_id) AS num_items FROM quizzes q JOIN lessons l ON l.lesson_id=q.lesson_id WHERE l.teacher_id=%s ORDER BY q.created_at DESC """, (teacher_id,)) return cur.fetchall() def list_all_students_for_teacher(teacher_id:int): with cursor() as cur: cur.execute(""" SELECT DISTINCT u.user_id, u.name, u.email FROM classes c JOIN class_students cs ON cs.class_id=c.class_id JOIN users u ON u.user_id=cs.student_id WHERE c.teacher_id=%s ORDER BY u.name """, (teacher_id,)) return cur.fetchall() # ----- ASSIGNEES (students) ----- def list_assigned_students_for_lesson(lesson_id:int): with cursor() as cur: cur.execute(""" WITH direct AS ( SELECT student_id FROM assignments WHERE lesson_id=%s AND student_id IS NOT NULL ), via_class AS ( SELECT cs.student_id FROM assignments a JOIN class_students cs ON cs.class_id=a.class_id WHERE a.lesson_id=%s AND a.class_id IS NOT NULL ), all_students AS ( SELECT student_id FROM direct UNION SELECT student_id FROM via_class ) SELECT u.user_id, u.name, u.email FROM users u JOIN all_students s ON s.student_id=u.user_id ORDER BY u.name """, (lesson_id, lesson_id)) return cur.fetchall() def list_assigned_students_for_quiz(quiz_id:int): with cursor() as cur: cur.execute(""" WITH direct AS ( SELECT student_id FROM assignments WHERE quiz_id=%s AND student_id IS NOT NULL ), via_class AS ( SELECT cs.student_id FROM assignments a JOIN class_students cs ON cs.class_id=a.class_id WHERE a.quiz_id=%s AND a.class_id IS NOT NULL ), all_students AS ( SELECT student_id FROM direct UNION SELECT student_id FROM via_class ) SELECT u.user_id, u.name, u.email FROM users u JOIN all_students s ON s.student_id=u.user_id ORDER BY u.name """, (quiz_id, quiz_id)) return cur.fetchall() # ----- ASSIGN ACTIONS ----- def assign_lesson_to_students(lesson_id:int, student_ids:list[int], teacher_id:int, due_at:str|None=None): # bulk insert; quiz_id stays NULL with cursor() as cur: for sid in student_ids: cur.execute(""" INSERT INTO assignments(lesson_id, quiz_id, student_id, assigned_by, due_at) VALUES(%s, NULL, %s, %s, %s) ON DUPLICATE KEY UPDATE due_at=VALUES(due_at) """, (lesson_id, sid, teacher_id, due_at)) return True def assign_quiz_to_students(quiz_id:int, student_ids:list[int], teacher_id:int, due_at:str|None=None): # get lesson_id for integrity with cursor() as cur: cur.execute("SELECT lesson_id FROM quizzes WHERE quiz_id=%s", (quiz_id,)) row = cur.fetchone() if not row: raise ValueError("Quiz not found") lesson_id = row["lesson_id"] for sid in student_ids: cur.execute(""" INSERT INTO assignments(lesson_id, quiz_id, student_id, assigned_by, due_at) VALUES(%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE due_at=VALUES(due_at) """, (lesson_id, quiz_id, sid, teacher_id, due_at)) return True # ----- SAFE DELETE ----- def delete_lesson(lesson_id:int, teacher_id:int): with cursor() as cur: # ownership check cur.execute("SELECT 1 FROM lessons WHERE lesson_id=%s AND teacher_id=%s", (lesson_id, teacher_id)) if not cur.fetchone(): return False, "You can only delete own lesson." # block if assigned or quizzed cur.execute("SELECT COUNT(*) AS n FROM assignments WHERE lesson_id=%s", (lesson_id,)) if cur.fetchone()["n"] > 0: return False, "Remove assignments first." cur.execute("SELECT COUNT(*) AS n FROM quizzes WHERE lesson_id=%s", (lesson_id,)) if cur.fetchone()["n"] > 0: return False, "Delete quizzes for this lesson first." # delete sections then lesson cur.execute("DELETE FROM lesson_sections WHERE lesson_id=%s", (lesson_id,)) cur.execute("DELETE FROM lessons WHERE lesson_id=%s AND teacher_id=%s", (lesson_id, teacher_id)) return True, "Deleted." def delete_quiz(quiz_id:int, teacher_id:int): with cursor() as cur: cur.execute(""" SELECT 1 FROM quizzes q JOIN lessons l ON l.lesson_id=q.lesson_id WHERE q.quiz_id=%s AND l.teacher_id=%s """, (quiz_id, teacher_id)) if not cur.fetchone(): return False, "You can only delete own quiz." cur.execute("SELECT COUNT(*) AS n FROM submissions WHERE quiz_id=%s", (quiz_id,)) if cur.fetchone()["n"] > 0: return False, "This quiz has submissions. Deleting is blocked." cur.execute("DELETE FROM quiz_items WHERE quiz_id=%s", (quiz_id,)) cur.execute("DELETE FROM assignments WHERE quiz_id=%s", (quiz_id,)) cur.execute("DELETE FROM quizzes WHERE quiz_id=%s", (quiz_id,)) return True, "Deleted." def _bump_game_stats(user_id:int, slug:str, *, gained_xp:int, matched:int|None=None, level_inc:int=0): with cursor() as cur: cur.execute(""" INSERT INTO game_stats(user_id,game_slug,total_xp,matches,level) VALUES(%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), matches = matches + VALUES(matches), level = GREATEST(level, VALUES(level)) """, (user_id, slug, int(gained_xp), int(matched or 1), level_inc)) # ---------- LESSONS ---------- def create_lesson(teacher_id:int, title:str, description:str, subject:str, level_slug:str, sections:list[dict]): with cursor() as cur: cur.execute(""" INSERT INTO lessons(teacher_id,title,description,subject,level,duration_min) VALUES(%s,%s,%s,%s,%s,%s) """, (teacher_id, title, description, subject, level_slug, 60)) cur.execute("SELECT LAST_INSERT_ID() AS id") lesson_id = cur.fetchone()["id"] for i, sec in enumerate(sections, start=1): cur.execute(""" INSERT INTO lesson_sections(lesson_id,position,title,content) VALUES(%s,%s,%s,%s) """, (lesson_id, i, sec.get("title"), sec.get("content"))) return lesson_id def list_lessons_by_teacher(teacher_id:int): with cursor() as cur: cur.execute("SELECT * FROM lessons WHERE teacher_id=%s ORDER BY created_at DESC", (teacher_id,)) return cur.fetchall() def get_lesson(lesson_id:int): with cursor() as cur: cur.execute("SELECT * FROM lessons WHERE lesson_id=%s", (lesson_id,)) lesson = cur.fetchone() cur.execute("SELECT * FROM lesson_sections WHERE lesson_id=%s ORDER BY position", (lesson_id,)) sections = cur.fetchall() return {"lesson": lesson, "sections": sections} # ---------- QUIZZES ---------- def create_quiz(lesson_id:int, title:str, items:list[dict], settings:dict|None=None): with cursor() as cur: cur.execute("INSERT INTO quizzes(lesson_id,title,settings) VALUES(%s,%s,%s)", (lesson_id, title, json.dumps(settings or {}))) cur.execute("SELECT LAST_INSERT_ID() AS id") quiz_id = cur.fetchone()["id"] for i, it in enumerate(items, start=1): cur.execute(""" INSERT INTO quiz_items(quiz_id,position,question,options,answer_key,points) VALUES(%s,%s,%s,%s,%s,%s) """, (quiz_id, i, it["question"], json.dumps(it.get("options", [])), json.dumps(it.get("answer_key")), int(it.get("points", 1)))) return quiz_id def get_quiz(quiz_id:int): with cursor() as cur: cur.execute("SELECT * FROM quizzes WHERE quiz_id=%s", (quiz_id,)) quiz = cur.fetchone() cur.execute("SELECT * FROM quiz_items WHERE quiz_id=%s ORDER BY position", (quiz_id,)) items = cur.fetchall() return {"quiz": quiz, "items": items} # ---------- ASSIGNMENTS ---------- def assign_to_class(lesson_id:int, quiz_id:int|None, class_id:int, teacher_id:int, due_at:str|None=None): with cursor() as cur: cur.execute(""" INSERT INTO assignments(lesson_id,quiz_id,class_id,assigned_by,due_at) VALUES(%s,%s,%s,%s,%s) """, (lesson_id, quiz_id, class_id, teacher_id, due_at)) cur.execute("SELECT LAST_INSERT_ID() AS id") return cur.fetchone()["id"] def assign_to_student(lesson_id:int, quiz_id:int|None, student_id:int, teacher_id:int, due_at:str|None=None): with cursor() as cur: cur.execute(""" INSERT INTO assignments(lesson_id,quiz_id,student_id,assigned_by,due_at) VALUES(%s,%s,%s,%s,%s) """, (lesson_id, quiz_id, student_id, teacher_id, due_at)) cur.execute("SELECT LAST_INSERT_ID() AS id") return cur.fetchone()["id"] def list_assignments_for_student(student_id:int): with cursor() as cur: cur.execute(""" SELECT a.assignment_id, a.lesson_id, l.title, l.subject, l.level, a.quiz_id, a.due_at, COALESCE(lp.status,'not_started') AS status, lp.current_pos FROM ( SELECT a.*, ROW_NUMBER() OVER ( PARTITION BY a.lesson_id ORDER BY a.created_at DESC, a.assignment_id DESC ) AS rn FROM assignments a WHERE a.student_id = %s OR a.class_id IN (SELECT class_id FROM class_students WHERE student_id = %s) ) AS a JOIN lessons l ON l.lesson_id = a.lesson_id LEFT JOIN lesson_progress lp ON lp.user_id = %s AND lp.lesson_id = a.lesson_id WHERE a.rn = 1 ORDER BY a.created_at DESC """, (student_id, student_id, student_id)) return cur.fetchall() # ---------- PROGRESS and SUBMISSIONS ---------- def save_progress(user_id:int, lesson_id:int, current_pos:int, status:str): with cursor() as cur: cur.execute(""" INSERT INTO lesson_progress(user_id,lesson_id,current_pos,status) VALUES(%s,%s,%s,%s) ON DUPLICATE KEY UPDATE current_pos=VALUES(current_pos), status=VALUES(status) """, (user_id, lesson_id, current_pos, status)) return True def submit_quiz(student_id:int, assignment_id:int, quiz_id:int, score:int, total:int, details:dict): with cursor() as cur: cur.execute(""" INSERT INTO submissions(assignment_id,quiz_id,student_id,score,total,details) VALUES(%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE score=VALUES(score), total=VALUES(total), details=VALUES(details), submitted_at=CURRENT_TIMESTAMP """, (assignment_id, quiz_id, student_id, score, total, json.dumps(details))) return True # ---------- DASHBOARD SHORTCUTS ---------- def teacher_tiles(teacher_id:int): with cursor() as cur: cur.execute("SELECT * FROM v_class_stats WHERE teacher_id=%s", (teacher_id,)) rows = cur.fetchall() total_students = sum(r["total_students"] for r in rows) lessons_created = _count_lessons(teacher_id) # use simple averages; adjust later as needed class_avg = round(sum(r["class_avg"] for r in rows)/len(rows), 2) if rows else 0 active_students = sum(1 for r in rows if r.get("recent_submissions",0) > 0) return dict(total_students=total_students, class_avg=class_avg, lessons_created=lessons_created, active_students=active_students) def _count_lessons(teacher_id:int): with cursor() as cur: cur.execute("SELECT COUNT(*) AS n FROM lessons WHERE teacher_id=%s", (teacher_id,)) return cur.fetchone()["n"] # --- XP and streak helpers --- def user_xp_and_level(user_id: int, base: int = 500): with cursor() as cur: cur.execute("SELECT COALESCE(SUM(delta),0) AS xp FROM xp_log WHERE user_id=%s", (user_id,)) xp = int((cur.fetchone() or {"xp": 0})["xp"]) cur.execute("SELECT COALESCE(days,0) AS days FROM streaks WHERE user_id=%s", (user_id,)) streak = int((cur.fetchone() or {"days": 0})["days"]) # level math level = max(1, xp // base + 1) start_of_level = (level - 1) * base into = xp - start_of_level need = base # exact boundary should flip level and reset progress if into == need: level += 1 into = 0 return { "xp": xp, # lifetime XP from the DB "level": level, # current level "into": into, # XP inside this level "need": need, # XP needed to reach next level "streak": streak, } def recent_lessons_for_student(user_id:int, limit:int=5): with cursor() as cur: cur.execute(""" SELECT l.title, CASE WHEN lp.status='completed' THEN 100 WHEN lp.current_pos IS NULL THEN 0 ELSE LEAST(95, lp.current_pos * 10) END AS progress FROM lessons l LEFT JOIN lesson_progress lp ON lp.lesson_id=l.lesson_id AND lp.user_id=%s WHERE l.lesson_id IN ( SELECT lesson_id FROM assignments WHERE student_id=%s OR class_id IN (SELECT class_id FROM class_students WHERE student_id=%s) ) ORDER BY l.created_at DESC LIMIT %s """, (user_id, user_id, user_id, limit)) return cur.fetchall() def student_quiz_average(student_id: int) -> int: """ Returns the student's average quiz percentage (0–100) using the latest submission per quiz from the `submissions` table. """ with cursor() as cur: cur.execute(""" WITH latest AS ( SELECT quiz_id, MAX(submitted_at) AS last_ts FROM submissions WHERE student_id = %s GROUP BY quiz_id ) SELECT ROUND(AVG(s.score * 100.0 / NULLIF(s.total,0))) AS pct FROM latest t JOIN submissions s ON s.quiz_id = t.quiz_id AND s.submitted_at = t.last_ts WHERE s.student_id = %s """, (student_id, student_id)) row = cur.fetchone() or {} return int(row.get("pct") or 0) # --- Generic XP bump and streak touch --- def add_xp(user_id:int, delta:int, source:str, meta:dict|None=None): with cursor() as cur: cur.execute( "INSERT INTO xp_log(user_id,source,delta,meta) VALUES(%s,%s,%s,%s)", (user_id, source, int(delta), json.dumps(meta or {})) ) # streak touch cur.execute("SELECT days, last_active FROM streaks WHERE user_id=%s", (user_id,)) row = cur.fetchone() today = date.today() if not row: cur.execute("INSERT INTO streaks(user_id,days,last_active) VALUES(%s,%s,%s)", (user_id, 1, today)) else: last = row["last_active"] days = int(row["days"] or 0) if last is None or last < today: # if we missed a day, reset to 1 else +1 if last and (today - last) > timedelta(days=1): days = 1 else: days = max(1, days + 1) cur.execute("UPDATE streaks SET days=%s,last_active=%s WHERE user_id=%s", (days, today, user_id)) # -- leaderboard helpders --- def leaderboard_for_class(class_id: int, limit: int = 10): """ Returns: [{'user_id': int, 'name': str, 'xp': int, 'level': int}, ...] Sorted by XP (desc) for students in a specific class. """ with cursor() as cur: cur.execute(""" SELECT u.user_id, u.name, COALESCE(x.total_xp, 0) AS xp FROM class_students cs JOIN users u ON u.user_id = cs.student_id LEFT JOIN ( SELECT user_id, SUM(delta) AS total_xp FROM xp_log GROUP BY user_id ) x ON x.user_id = u.user_id WHERE cs.class_id = %s ORDER BY COALESCE(x.total_xp, 0) DESC, u.name LIMIT %s """, (class_id, limit)) rows = cur.fetchall() or [] # attach levels using curve for r in rows: r["level"] = level_from_xp(r.get("xp", 0)) return rows def leaderboard_global(limit: int = 10): """ Returns: [{'user_id': int, 'name': str, 'xp': int, 'level': int}, ...] Top students across the whole app by XP. """ with cursor() as cur: cur.execute(""" SELECT u.user_id, u.name, COALESCE(x.total_xp, 0) AS xp FROM users u LEFT JOIN ( SELECT user_id, SUM(delta) AS total_xp FROM xp_log GROUP BY user_id ) x ON x.user_id = u.user_id WHERE u.role_slug = 'student' ORDER BY COALESCE(x.total_xp, 0) DESC, u.name LIMIT %s """, (limit,)) rows = cur.fetchall() or [] for r in rows: r["level"] = level_from_xp(r.get("xp", 0)) return rows # --- Game logging helpers --- def record_money_match_play(user_id:int, *, target:int, total:int, elapsed_ms:int, matched:bool, gained_xp:int): with cursor() as cur: cur.execute(""" INSERT INTO game_sessions(user_id,game_slug,target,total,elapsed_ms,matched,gained_xp,ended_at) VALUES(%s,'money_match',%s,%s,%s,%s,%s,NOW()) """, (user_id, target, total, elapsed_ms, 1 if matched else 0, gained_xp)) cur.execute(""" INSERT INTO money_match_history(user_id,target,total,elapsed_ms,gained_xp,matched) VALUES(%s,%s,%s,%s,%s,%s) """, (user_id, target, total, elapsed_ms, gained_xp, 1 if matched else 0)) cur.execute(""" INSERT INTO money_match_stats(user_id,total_xp,matches,best_time_ms,best_target) VALUES(%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), matches = matches + VALUES(matches), best_time_ms = LEAST(COALESCE(best_time_ms, VALUES(best_time_ms)), VALUES(best_time_ms)), best_target = COALESCE(best_target, VALUES(best_target)) """, (user_id, gained_xp, 1 if matched else 0, elapsed_ms if matched else None, target if matched else None)) _bump_game_stats(user_id, "money_match", gained_xp=gained_xp, matched=1 if matched else 0) add_xp(user_id, gained_xp, "game", {"game":"money_match","target":target,"total":total,"elapsed_ms":elapsed_ms,"matched":matched}) def record_budget_builder_save(user_id:int, *, weekly_allowance:int, allocations:list[dict]): total_allocated = sum(int(x.get("amount",0)) for x in allocations) remaining = int(weekly_allowance) - total_allocated gained_xp = 150 if remaining == 0 else 100 if remaining > 0 else 50 with cursor() as cur: cur.execute(""" INSERT INTO game_sessions(user_id,game_slug,gained_xp,ended_at) VALUES(%s,'budget_builder',%s,NOW()) """, (user_id, gained_xp)) cur.execute(""" INSERT INTO budget_builder_history(user_id,weekly_allowance,allocations,total_allocated,remaining,gained_xp) VALUES(%s,%s,%s,%s,%s,%s) """, (user_id, weekly_allowance, json.dumps(allocations), total_allocated, remaining, gained_xp)) cur.execute(""" INSERT INTO budget_builder_stats(user_id,total_xp,plays,best_balance) VALUES(%s,%s,1,%s) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), plays = plays + 1, best_balance = GREATEST(COALESCE(best_balance, 0), VALUES(best_balance)) """, (user_id, gained_xp, remaining)) _bump_game_stats(user_id, "budget_builder", gained_xp=gained_xp, matched=1) add_xp(user_id, gained_xp, "game", {"game":"budget_builder","remaining":remaining}) def record_debt_dilemma_round( user_id:int, *, level:int, round_no:int, wallet:int, health:int, happiness:int, credit_score:int, event_json:dict, outcome:str, gained_xp:int ): with cursor() as cur: cur.execute(""" INSERT INTO game_sessions(user_id,game_slug,gained_xp,ended_at) VALUES(%s,'debt_dilemma',%s,NOW()) """, (user_id, gained_xp)) cur.execute(""" INSERT INTO debt_dilemma_history(user_id,level,round_no,wallet,health,happiness,credit_score,event_json,outcome,gained_xp) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, (user_id, level, round_no, wallet, health, happiness, credit_score, json.dumps(event_json or {}), outcome, gained_xp)) cur.execute(""" INSERT INTO debt_dilemma_stats(user_id,total_xp,plays,highest_level,last_outcome) VALUES(%s,%s,1,%s,%s) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), plays = plays + 1, highest_level = GREATEST(COALESCE(highest_level,0), VALUES(highest_level)), last_outcome = VALUES(last_outcome) """, (user_id, gained_xp, level, outcome)) # Treat a completed month/level as a "match" _bump_game_stats(user_id, "debt_dilemma", gained_xp=gained_xp, matched=1, level_inc=level) add_xp(user_id, gained_xp, "game", { "game":"debt_dilemma","level":level,"round":round_no,"outcome":outcome }) def record_profit_puzzle_result( user_id:int, *, scenario_id:str, title:str, units:int, price:int, cost:int, user_answer:float, actual_profit:float, is_correct:bool, gained_xp:int ): with cursor() as cur: # generic session row for cross-game views cur.execute(""" INSERT INTO game_sessions(user_id,game_slug,gained_xp,ended_at) VALUES(%s,'profit_puzzle',%s,NOW()) """, (user_id, int(gained_xp))) # detailed history cur.execute(""" INSERT INTO profit_puzzle_history (user_id,scenario_id,title,units,price,cost,user_answer,actual_profit,is_correct,gained_xp) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) """, (user_id, scenario_id, title, int(units), int(price), int(cost), float(user_answer), float(actual_profit), 1 if is_correct else 0, int(gained_xp))) # per-game stats cur.execute(""" INSERT INTO profit_puzzle_stats(user_id,total_xp,plays,correct,last_score) VALUES(%s,%s,1,%s,%s) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), plays = plays + 1, correct = correct + VALUES(correct), last_score = VALUES(last_score), last_played = CURRENT_TIMESTAMP """, (user_id, int(gained_xp), 1 if is_correct else 0, int(gained_xp))) # game_stats rollup like other games cur.execute(""" INSERT INTO game_stats(user_id,game_slug,total_xp,matches,level) VALUES(%s,'profit_puzzle',%s,%s,1) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), matches = matches + VALUES(matches) """, (user_id, int(gained_xp), 1 if is_correct else 0)) # global XP and streak add_xp(user_id, int(gained_xp), "game", {"game":"profit_puzzle","scenario":scenario_id,"correct":bool(is_correct)}) # --- Profit Puzzle logging --- def record_profit_puzzle_progress(user_id:int, *, scenario_title:str, correct:bool, gained_xp:int): """ Log a Profit Puzzle step and bump XP. - Writes to generic game_sessions and game_stats - Writes to xp_log via add_xp """ with cursor() as cur: # session line item cur.execute(""" INSERT INTO game_sessions(user_id, game_slug, gained_xp, ended_at) VALUES(%s, 'profit_puzzle', %s, NOW()) """, (user_id, int(gained_xp))) # aggregate by game cur.execute(""" INSERT INTO game_stats(user_id, game_slug, total_xp, matches, level) VALUES(%s, 'profit_puzzle', %s, %s, 1) ON DUPLICATE KEY UPDATE total_xp = total_xp + VALUES(total_xp), matches = matches + VALUES(matches) """, (user_id, int(gained_xp), 1 if correct else 0)) add_xp( user_id, int(gained_xp), "game", {"game": "profit_puzzle", "scenario": scenario_title, "correct": bool(correct)} )