Spaces:
Sleeping
Sleeping
taboola
add student metrics dashboard, save quiz answer records, save original uploaded files
25cb9ab | """Dashboard API. | |
| Two access levels share these routes: | |
| - Admins (is_admin) see ALL teachers' data and can manage users. | |
| - Regular teachers see ONLY their own data (their quizzes/students/results) and | |
| cannot reach user-management or other teachers' records. | |
| Scoping is enforced server-side: list endpoints filter by the caller's teacher | |
| id, and detail endpoints verify ownership before returning anything. | |
| """ | |
| from __future__ import annotations | |
| from typing import Optional | |
| from fastapi import APIRouter, Depends, HTTPException | |
| from fastapi.responses import FileResponse | |
| from pydantic import BaseModel | |
| from .auth import get_current_admin, get_current_user | |
| from .crypto import decrypt | |
| from .database import ( | |
| admin_list_sessions, | |
| admin_list_students, | |
| delete_session, | |
| delete_user, | |
| dismiss_issue_report, | |
| get_active_users, | |
| get_answer_grid, | |
| get_assessments_processed, | |
| get_feedback_loop_funnel, | |
| get_parsed_data, | |
| get_practice_quiz_funnel, | |
| get_quiz_results, | |
| get_raw_file, | |
| get_raw_files, | |
| get_schema_version, | |
| get_session, | |
| get_signup_growth, | |
| get_stats, | |
| get_student, | |
| get_student_improvement_summary, | |
| get_student_timeline, | |
| get_teacher_retention, | |
| get_teacher_vocabulary, | |
| list_all_users, | |
| list_open_issue_reports, | |
| set_teacher_vocabulary, | |
| set_user_admin, | |
| wipe_quiz_data, | |
| wipe_question_bank, | |
| list_question_bank, | |
| list_question_bank_tags, | |
| delete_question_bank_row, | |
| ) | |
| from .student_credentials import ensure_student_login | |
| router = APIRouter(prefix="/api/admin", tags=["admin"]) | |
| def _scope(user: dict) -> Optional[int]: | |
| """None for admins (see everything), else the teacher id to scope to.""" | |
| return None if user.get("is_admin") else user["id"] | |
| def _owns_quiz(user: dict, quiz: dict) -> bool: | |
| return user.get("is_admin") or quiz.get("teacher_id") == user["id"] | |
| def _owns_student(user: dict, student: dict) -> bool: | |
| return user.get("is_admin") or student.get("teacher_id") == user["id"] | |
| async def dashboard_stats(user=Depends(get_current_user)): | |
| stats = await get_stats(_scope(user)) | |
| stats["is_admin"] = bool(user.get("is_admin")) | |
| if user.get("is_admin"): | |
| stats["schema_version"] = await get_schema_version() | |
| return stats | |
| async def dashboard_analytics(_admin=Depends(get_current_admin)): | |
| """Signup growth over time โ admin-only, global (not scoped per teacher).""" | |
| return await get_signup_growth() | |
| async def dashboard_active_users(weeks: int = 12, _admin=Depends(get_current_admin)): | |
| return await get_active_users(weeks=weeks) | |
| async def dashboard_assessments_processed( | |
| months: int = 6, _admin=Depends(get_current_admin) | |
| ): | |
| return await get_assessments_processed(months=months) | |
| async def dashboard_practice_quiz_funnel(_admin=Depends(get_current_admin)): | |
| return await get_practice_quiz_funnel() | |
| async def dashboard_student_improvement(_admin=Depends(get_current_admin)): | |
| return await get_student_improvement_summary() | |
| async def dashboard_teacher_retention(_admin=Depends(get_current_admin)): | |
| return await get_teacher_retention() | |
| async def dashboard_feedback_loop(_admin=Depends(get_current_admin)): | |
| return await get_feedback_loop_funnel() | |
| # ---- Issue reports (admin only) ---- | |
| async def dashboard_list_issues(_admin=Depends(get_current_admin)): | |
| return {"issues": await list_open_issue_reports()} | |
| async def dashboard_dismiss_issue(issue_id: int, _admin=Depends(get_current_admin)): | |
| ok = await dismiss_issue_report(issue_id) | |
| if not ok: | |
| raise HTTPException( | |
| status_code=404, detail="Issue not found or already dismissed" | |
| ) | |
| return {"status": "dismissed"} | |
| # ---- User management (admin only) ---- | |
| async def admin_users(_admin=Depends(get_current_admin)): | |
| return {"users": await list_all_users()} | |
| class SetAdminRequest(BaseModel): | |
| is_admin: bool | |
| async def admin_set_admin( | |
| user_id: int, body: SetAdminRequest, admin=Depends(get_current_admin) | |
| ): | |
| if user_id == admin["id"] and not body.is_admin: | |
| raise HTTPException( | |
| status_code=400, detail="You cannot remove your own admin access" | |
| ) | |
| await set_user_admin(user_id, body.is_admin) | |
| return {"status": "ok", "user_id": user_id, "is_admin": body.is_admin} | |
| async def admin_delete_user(user_id: int, admin=Depends(get_current_admin)): | |
| if user_id == admin["id"]: | |
| raise HTTPException( | |
| status_code=400, detail="You cannot delete your own account here" | |
| ) | |
| await delete_user(user_id) | |
| return {"status": "deleted"} | |
| # ---- Sessions / quizzes (scoped) ---- | |
| async def dashboard_sessions(user=Depends(get_current_user)): | |
| return {"sessions": await admin_list_sessions(_scope(user))} | |
| async def dashboard_session_detail(session_id: int, user=Depends(get_current_user)): | |
| """Full quiz detail: structured parsed data, results.""" | |
| quiz = await get_session(session_id) | |
| if not quiz or not _owns_quiz(user, quiz): | |
| raise HTTPException(status_code=404, detail="Session not found") | |
| questions = await get_parsed_data(session_id) | |
| grid = await get_answer_grid(session_id) | |
| parsed: dict[str, object] = {"questions": questions} | |
| if grid: | |
| parsed["answer_grid"] = grid | |
| return { | |
| "quiz": quiz, | |
| "parsed_data": parsed, | |
| "results": await get_quiz_results(session_id), | |
| "raw_files": await get_raw_files(session_id), | |
| } | |
| async def dashboard_raw_file( | |
| session_id: int, raw_file_id: int, user=Depends(get_current_user) | |
| ): | |
| """Serve the original uploaded scan/PDF for a quiz record.""" | |
| quiz = await get_session(session_id) | |
| if not quiz or not _owns_quiz(user, quiz): | |
| raise HTTPException(status_code=404, detail="Session not found") | |
| raw_file = await get_raw_file(raw_file_id) | |
| if not raw_file or raw_file["quiz_id"] != session_id: | |
| raise HTTPException(status_code=404, detail="File not found") | |
| return FileResponse( | |
| raw_file["storage_path"], | |
| media_type=raw_file["content_type"] or "application/octet-stream", | |
| filename=raw_file["file_name"], | |
| ) | |
| async def dashboard_delete_session(session_id: int, user=Depends(get_current_user)): | |
| quiz = await get_session(session_id) | |
| if not quiz or not _owns_quiz(user, quiz): | |
| raise HTTPException(status_code=404, detail="Session not found") | |
| await delete_session(session_id) | |
| return {"status": "deleted"} | |
| async def dashboard_wipe_session_data(session_id: int, user=Depends(get_current_user)): | |
| """Wipe all data for a quiz (results, reports, grids, parsed_data, files) without deleting the quiz.""" | |
| quiz = await get_session(session_id) | |
| if not quiz or not _owns_quiz(user, quiz): | |
| raise HTTPException(status_code=404, detail="Session not found") | |
| counts = await wipe_quiz_data(session_id) | |
| return {"status": "wiped", **counts} | |
| async def dashboard_wipe_question_bank(user=Depends(get_current_user)): | |
| """Delete question bank rows contributed by the current teacher (or all, for admins).""" | |
| count = await wipe_question_bank(teacher_id=_scope(user)) | |
| return {"status": "deleted", "count": count} | |
| _DEV_WIPE_ALLOWED_EMAIL = "melfurever21@gmail.com" | |
| async def dev_wipe_all_question_bank(user=Depends(get_current_user)): | |
| """TEMPORARY dev-only tool: wipes the ENTIRE shared question bank (every | |
| teacher's contributions). Restricted to one account by email โ the | |
| button is hidden from everyone else in the UI, but that's not real | |
| enforcement on its own, so it's checked here too. | |
| Delete this endpoint and its frontend button once the fresh question set | |
| re-upload is done โ it's intentionally a blunt, unscoped wipe and should | |
| not stay in the app long-term. Kept separate from | |
| dashboard_wipe_question_bank above (which is scoped/admin-aware and | |
| meant to be permanent) so removing this later is a clean, contained | |
| deletion. | |
| """ | |
| if user.get("email") != _DEV_WIPE_ALLOWED_EMAIL: | |
| raise HTTPException(status_code=403, detail="Not authorized") | |
| count = await wipe_question_bank(teacher_id=None) | |
| return {"status": "deleted", "count": count} | |
| async def dashboard_list_question_bank( | |
| category: Optional[str] = None, | |
| tag: Optional[str] = None, | |
| search: Optional[str] = None, | |
| limit: int = 50, | |
| offset: int = 0, | |
| user=Depends(get_current_user), | |
| ): | |
| """Browse the shared question bank (the pool practice-quiz generation | |
| draws from โ every teacher's contributions, visible to every teacher).""" | |
| rows, total = await list_question_bank( | |
| category, tag, search, limit=limit, offset=offset | |
| ) | |
| return {"questions": rows, "total": total} | |
| async def dashboard_question_bank_tags(user=Depends(get_current_user)): | |
| return {"categories": await list_question_bank_tags()} | |
| async def dashboard_delete_question_bank_row( | |
| question_id: int, user=Depends(get_current_user) | |
| ): | |
| """Remove a single question from the shared bank โ any teacher can curate | |
| it, since generation draws from the whole shared pool, not just their own | |
| contributions.""" | |
| ok = await delete_question_bank_row(question_id) | |
| if not ok: | |
| raise HTTPException(status_code=404, detail="Question not found") | |
| return {"status": "deleted"} | |
| # ---- Vocabulary (personal, per-teacher โ not shared) ---- | |
| class VocabWord(BaseModel): | |
| word: str | |
| class SetVocabularyRequest(BaseModel): | |
| words: list[VocabWord] | |
| async def dashboard_get_vocabulary(user=Depends(get_current_user)): | |
| return {"words": await get_teacher_vocabulary(user["id"])} | |
| async def dashboard_set_vocabulary( | |
| body: SetVocabularyRequest, user=Depends(get_current_user) | |
| ): | |
| words = [w.model_dump() for w in body.words] | |
| saved = await set_teacher_vocabulary(user["id"], words) | |
| return {"words": saved} | |
| # ---- Students (scoped) ---- | |
| async def _ensure_login(student: dict) -> dict: | |
| """Auto-issue a login the first time a student shows up here without | |
| one (see student_credentials.ensure_student_login). Swaps the stored | |
| ciphertext for the plaintext password so ๆ็่ณๆ shows credentials on an | |
| ongoing basis, not just once at issuance.""" | |
| student, new_password = await ensure_student_login(student) | |
| if new_password is not None: | |
| student.pop("password_encrypted", None) | |
| student["password"] = new_password | |
| return student | |
| encrypted = student.pop("password_encrypted", None) | |
| student["password"] = decrypt(encrypted) if encrypted else None | |
| return student | |
| async def dashboard_students(user=Depends(get_current_user)): | |
| students = await admin_list_students(_scope(user)) | |
| return {"students": [await _ensure_login(s) for s in students]} | |
| async def dashboard_student_detail(student_id: int, user=Depends(get_current_user)): | |
| student = await get_student(student_id) | |
| if not student or not _owns_student(user, student): | |
| raise HTTPException(status_code=404, detail="Student not found") | |
| return { | |
| "student": await _ensure_login(student), | |
| "timeline": await get_student_timeline(student_id), | |
| } | |