"""Instructor routes.""" from __future__ import annotations import base64 import json import secrets from io import BytesIO from typing import Any import qrcode import qrcode.image.svg from fastapi import APIRouter, File, HTTPException, Request, Response, UploadFile, WebSocket from fastapi.responses import FileResponse, HTMLResponse, PlainTextResponse from app import auth from app.config import Settings from app.csv_export import export_session_csv from app.db import connect from app.models import QuizCreateRequest, SessionCreateRequest from app.pool import PoolValidationError, parse_pool_json from app.room import RoomManager CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" def router(settings: Settings, rooms: RoomManager) -> APIRouter: api = APIRouter() def require_admin(request: Request) -> None: auth.require_admin_request(settings, request) @api.get("/admin/login") async def login_form(): return HTMLResponse( "Admin Login
" "" "
" ) @api.post("/admin/login") async def login(request: Request, response: Response): password = "" content_type = request.headers.get("content-type", "") if "application/json" in content_type: data = await request.json() password = str(data.get("password", "")) else: form = await request.form() password = str(form.get("password", "")) if not auth.verify_admin_password(settings, password): raise HTTPException(status_code=401, detail="Invalid admin password") auth.set_admin_cookie(settings, response, auth.sign_admin(settings)) return {"ok": True} @api.get("/admin/") async def admin_page(request: Request): require_admin(request) return FileResponse("static/admin.html") @api.get("/admin/api/quizzes") async def list_quizzes(request: Request): require_admin(request) async with connect(settings.db_path) as db: cursor = await db.execute( "SELECT id, title, time_limit_default, score_fn_name, created_at FROM quizzes ORDER BY created_at DESC, id DESC" ) rows = await cursor.fetchall() return {"quizzes": [dict(row) for row in rows]} @api.post("/admin/api/quizzes") async def create_quiz(request: Request, body: QuizCreateRequest): require_admin(request) try: pool = parse_pool_json(body.pool_json) except PoolValidationError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc if body.time_limit_default is not None: pool["time_limit_default"] = body.time_limit_default title = (body.title or pool["title"]).strip() async with connect(settings.db_path) as db: cursor = await db.execute( "INSERT INTO quizzes (title, pool_json, time_limit_default, score_fn_name) VALUES (?, ?, ?, ?)", (title, json.dumps(pool), pool["time_limit_default"], pool["score_fn"]), ) await db.commit() quiz_id = cursor.lastrowid return {"ok": True, "quiz_id": quiz_id} @api.post("/admin/api/quizzes/upload") async def upload_quiz(request: Request, file: UploadFile = File(...)): require_admin(request) raw = (await file.read()).decode("utf-8") try: pool = parse_pool_json(raw) except PoolValidationError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc async with connect(settings.db_path) as db: cursor = await db.execute( "INSERT INTO quizzes (title, pool_json, time_limit_default, score_fn_name) VALUES (?, ?, ?, ?)", (pool["title"], json.dumps(pool), pool["time_limit_default"], pool["score_fn"]), ) await db.commit() quiz_id = cursor.lastrowid return {"ok": True, "quiz_id": quiz_id} @api.get("/admin/api/sessions") async def list_sessions(request: Request): require_admin(request) async with connect(settings.db_path) as db: cursor = await db.execute( """ SELECT s.sid, s.quiz_id, s.title, s.state, s.current_question_idx, s.started_at, s.finished_at, COUNT(p.student_id) AS participant_count FROM quiz_sessions s LEFT JOIN participants p ON p.sid = s.sid GROUP BY s.sid ORDER BY s.started_at DESC """ ) rows = await cursor.fetchall() return {"sessions": [dict(row) for row in rows]} @api.post("/admin/api/sessions") async def create_session(request: Request, body: SessionCreateRequest): require_admin(request) async with connect(settings.db_path) as db: quiz_cursor = await db.execute("SELECT id, title FROM quizzes WHERE id = ?", (body.quiz_id,)) quiz = await quiz_cursor.fetchone() if quiz is None: raise HTTPException(status_code=404, detail="Quiz not found") sid = await _generate_sid(db) await db.execute( "INSERT INTO quiz_sessions (sid, quiz_id, title) VALUES (?, ?, ?)", (sid, body.quiz_id, quiz["title"]), ) await db.commit() join_url = f"{settings.public_url}/?sid={sid}" return {"sid": sid, "join_url": join_url, "qr_url": _qr_data_url(join_url)} @api.get("/admin/api/sessions/{sid}/csv") async def csv_download(sid: str, request: Request): require_admin(request) if not await rooms.session_exists(sid): raise HTTPException(status_code=404, detail="Session not found") csv_text = await export_session_csv(settings.db_path, sid) return PlainTextResponse( csv_text, media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{sid}-results.csv"'}, ) @api.websocket("/ws/instructor/{sid}") async def instructor_socket(websocket: WebSocket, sid: str): if not auth.is_admin_ws(settings, websocket) or not await rooms.session_exists(sid): await websocket.close(code=4001) return await rooms.instructor_ws(websocket, sid) return api async def _generate_sid(db: Any) -> str: for _ in range(5): sid = "".join(secrets.choice(CROCKFORD) for _ in range(6)) cursor = await db.execute("SELECT 1 FROM quiz_sessions WHERE sid = ?", (sid,)) if await cursor.fetchone() is None: return sid raise HTTPException(status_code=500, detail="Could not allocate session ID") def _qr_data_url(value: str) -> str: image = qrcode.make(value, image_factory=qrcode.image.svg.SvgPathImage) buf = BytesIO() image.save(buf) encoded = base64.b64encode(buf.getvalue()).decode("ascii") return f"data:image/svg+xml;base64,{encoded}"