quiz/app/routes_admin.py

"""Instructor routes (single-session deployment).

The deployment runs exactly one quiz session at a time, loaded from
`QUIZ_POOL_PATH` at startup. There is no per-quiz CRUD; the operator
edits the pool JSON on disk and restarts the service when they want
a new pool. The admin UI is therefore a thin control panel for the
single canonical session whose id is `Settings.default_session_id`.
"""

from __future__ import annotations

import base64
from io import BytesIO
from pathlib import Path

import qrcode
import qrcode.image.svg
from fastapi import APIRouter, HTTPException, Request, Response, WebSocket
from fastapi.responses import FileResponse, PlainTextResponse

from app import auth
from app.config import Settings
from app.csv_export import export_session_csv
from app.models import AdminLoginRequest
from app.rate_limit import TokenBucket, client_ip
from app.room import RoomManager


def router(settings: Settings, rooms: RoomManager) -> APIRouter:
    api = APIRouter()

    # Per-app instance so test apps get fresh state.
    # 10 attempts/minute/IP — generous for the instructor, hostile to brute
    # force without locking out the campus network on student endpoints
    # (which are not rate-limited at all, see rate_limit.py).
    login_bucket = TokenBucket(capacity=10, refill_per_minute=10)

    def require_admin(request: Request) -> None:
        auth.require_admin_request(settings, request)

    @api.post("/admin/login")
    async def login(body: AdminLoginRequest, request: Request, response: Response):
        ip = client_ip(request)
        if not login_bucket.take(ip):
            raise HTTPException(
                status_code=429,
                detail="Too many login attempts; try again in a minute.",
            )
        if not auth.verify_admin_password(settings, body.password):
            raise HTTPException(status_code=401, detail="Invalid admin password")
        auth.set_admin_cookie(settings, response, auth.sign_admin(settings))
        return {"ok": True}

    @api.post("/admin/logout")
    async def logout(response: Response):
        response.delete_cookie(auth.ADMIN_COOKIE, path="/")
        return {"ok": True}

    @api.get("/admin/")
    async def admin_page():
        # No auth gate; the SPA fetches /admin/api/state and renders
        # the login form on 401 or the dashboard on 200.
        return FileResponse(Path("static/admin.html"))

    @api.get("/admin/api/state")
    async def admin_state(request: Request):
        require_admin(request)
        sid = rooms.canonical_sid or settings.default_session_id
        if not await rooms.session_exists(sid):
            raise HTTPException(status_code=503, detail="Session is not initialised")
        session = await rooms.get_session(sid)
        pool = await rooms.get_pool_for_session(sid)
        join_url = f"{settings.public_url}/?sid={sid}"
        return {
            "sid": sid,
            "title": session["title"],
            "state": session["state"],
            "current_question_idx": session["current_question_idx"],
            "join_url": join_url,
            "qr_url": _qr_data_url(join_url),
            "pool_meta": {
                "score_fn": pool["score_fn"],
                "time_limit_default": pool["time_limit_default"],
                "question_count": len(pool["questions"]),
            },
        }

    @api.post("/admin/api/reset")
    async def admin_reset(request: Request):
        require_admin(request)
        sid = rooms.canonical_sid or settings.default_session_id
        if not await rooms.session_exists(sid):
            raise HTTPException(status_code=503, detail="Session is not initialised")
        await rooms.reset(sid)
        return {"ok": True}

    @api.get("/admin/api/csv")
    async def csv_download(request: Request):
        require_admin(request)
        sid = rooms.canonical_sid or settings.default_session_id
        if not await rooms.session_exists(sid):
            raise HTTPException(status_code=503, detail="Session is not initialised")
        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


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}"