quiz/app/csv_export.py

"""CSV export helpers."""

from __future__ import annotations

import csv
from io import StringIO

from app.db import connect


async def export_session_csv(db_path: str, sid: str) -> str:
    out = StringIO()
    writer = csv.writer(out)
    writer.writerow(
        [
            "sid",
            "student_id",
            "name",
            "question_idx",
            "answer",
            "elapsed_ms",
            "score",
            "status",
            "blur_count",
            "hidden_count",
            "duplicate_join_attempts",
        ]
    )
    async with connect(db_path) as db:
        cursor = await db.execute(
            """
            SELECT p.sid, p.student_id, p.name, s.question_idx, s.answer, s.elapsed_ms, s.score, s.status
            FROM participants p
            LEFT JOIN submissions s ON s.sid = p.sid AND s.student_id = p.student_id
            WHERE p.sid = ?
            ORDER BY p.student_id, s.question_idx
            """,
            (sid,),
        )
        rows = await cursor.fetchall()
        events_cur = await db.execute(
            """
            SELECT student_id, kind, COUNT(*) AS c
            FROM student_events
            WHERE sid = ? AND student_id IS NOT NULL
            GROUP BY student_id, kind
            """,
            (sid,),
        )
        events = await events_cur.fetchall()
    counts: dict[str, dict[str, int]] = {}
    for row in events:
        counts.setdefault(row["student_id"], {})[row["kind"]] = int(row["c"])
    for row in rows:
        per = counts.get(row["student_id"], {})
        writer.writerow(
            [
                row["sid"],
                row["student_id"],
                row["name"],
                "" if row["question_idx"] is None else row["question_idx"],
                row["answer"] or "",
                "" if row["elapsed_ms"] is None else row["elapsed_ms"],
                "" if row["score"] is None else row["score"],
                row["status"] or "",
                per.get("blur", 0),
                per.get("visibility_hidden", 0),
                per.get("duplicate_join", 0),
            ]
        )
    return out.getvalue()