chatkit/backend/app/database.py

"""Data-access layer for ClassLens, backed by SQLAlchemy async (Supabase Postgres).

Public function names/signatures are kept stable for existing callers:
  - "session" in the API == a row in the ``quizzes`` table (a quiz/exam session)
  - "user" == a row in the ``teachers`` table

New capabilities (students as first-class entities, per-student results, and
analytics) are added alongside the legacy CRUD helpers.
"""

from __future__ import annotations

import random
import re
import secrets
import unicodedata
from datetime import datetime, date, timezone
from typing import Optional

from sqlalchemy import Date, cast, delete, func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm.attributes import flag_modified

from .config import get_settings
from .db import get_engine, get_sessionmaker
from .models import (
    AnswerGrid as AnswerGridRecord,
    Base,
    ParsedData,
    PracticeQuiz,
    Prompt,
    Quiz,
    QuestionBank,
    RawFile,
    Report,
    Student,
    StudentResult,
    Teacher,
)



def _now() -> datetime:
    return datetime.now(timezone.utc)


def _row(obj) -> dict:
    """Serialize an ORM object to a plain dict of its columns."""
    return {c.key: getattr(obj, c.key) for c in obj.__table__.columns}


def normalize_name(name: str) -> str:
    """Canonical key for matching the same student across quizzes."""
    s = unicodedata.normalize("NFKC", (name or "").strip()).casefold()
    return re.sub(r"\s+", " ", s)


# =============================================================================
# Schema bootstrap
# =============================================================================

async def init_database():
    """Ensure schema exists and bootstrap admins.

    On Supabase, tables are managed by Alembic migrations; here we only create
    tables for the SQLite dev/test fallback. Admin bootstrap runs for both.
    """
    settings = get_settings()

    if settings.database_provider != "supabase":
        async with get_engine().begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

    await _bootstrap_admins()


async def _bootstrap_admins():
    settings = get_settings()
    emails = [e.strip().lower() for e in (settings.admin_emails or "").split(",") if e.strip()]
    if not emails:
        return
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Teacher).where(func.lower(Teacher.email).in_(emails)).values(is_admin=True)
        )
        await db.commit()


# =============================================================================
# Teachers (users)
# =============================================================================

async def create_user(
    email: str,
    password_hash: str,
    display_name: str = "",
    full_name: str = "",
    school: str = "",
) -> int:
    async with get_sessionmaker()() as db:
        teacher = Teacher(
            email=email,
            password_hash=password_hash,
            display_name=display_name or email.split("@")[0],
            full_name=full_name,
            school=school,
        )
        db.add(teacher)
        await db.flush()
        teacher_id = teacher.id
        await db.commit()
        return teacher_id


async def get_user_by_email(email: str) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(select(Teacher).where(Teacher.email == email))
        teacher = res.scalar_one_or_none()
        return _row(teacher) if teacher else None


async def get_user_by_id(user_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        teacher = await db.get(Teacher, user_id)
        return _row(teacher) if teacher else None


async def update_last_login(user_id: int):
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Teacher).where(Teacher.id == user_id).values(last_login=_now())
        )
        await db.commit()


async def set_user_admin(user_id: int, is_admin: bool):
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Teacher).where(Teacher.id == user_id).values(is_admin=is_admin)
        )
        await db.commit()


async def list_all_users() -> list[dict]:
    """Admin: all teachers (without password hashes), with per-teacher counts."""
    async with get_sessionmaker()() as db:
        res = await db.execute(select(Teacher).order_by(Teacher.created_at.desc()))
        teachers = res.scalars().all()
        out = []
        for t in teachers:
            d = _row(t)
            d.pop("password_hash", None)
            d["quiz_count"] = (
                await db.execute(select(func.count(Quiz.id)).where(Quiz.teacher_id == t.id))
            ).scalar_one()
            d["student_count"] = (
                await db.execute(select(func.count(Student.id)).where(Student.teacher_id == t.id))
            ).scalar_one()
            out.append(d)
        return out


async def delete_user(user_id: int):
    async with get_sessionmaker()() as db:
        await db.execute(delete(Teacher).where(Teacher.id == user_id))
        await db.commit()


# =============================================================================
# Quizzes (sessions)
# =============================================================================

async def create_session(user_id: int, title: str = "", quiz_date: Optional[date] = None) -> int:
    async with get_sessionmaker()() as db:
        if not title:
            # Number sequentially by quizzes the teacher actually finished and
            # saved — not every draft session ever created (a fresh draft is
            # created every time the upload flow is opened, so counting those
            # inflated this quickly, e.g. "Quiz 36" after just a few real quizzes).
            count = (await db.execute(
                select(func.count(Quiz.id)).where(
                    Quiz.teacher_id == user_id, Quiz.status == "saved"
                )
            )).scalar_one()
            title = f"Quiz {count + 1}"
        quiz = Quiz(teacher_id=user_id, title=title, status="draft", quiz_date=quiz_date)
        db.add(quiz)
        await db.flush()
        quiz_id = quiz.id
        await db.commit()
        return quiz_id


async def get_sessions(user_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Quiz).where(Quiz.teacher_id == user_id).order_by(Quiz.created_at.desc())
        )
        return [_row(q) for q in res.scalars().all()]


async def get_session(session_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        quiz = await db.get(Quiz, session_id)
        if not quiz:
            return None
        d = _row(quiz)
        d["user_id"] = quiz.teacher_id  # back-compat key for existing callers
        return d


async def update_session_status(session_id: int, status: str):
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Quiz).where(Quiz.id == session_id).values(status=status, updated_at=_now())
        )
        await db.commit()


async def update_session_meta(session_id: int, title: Optional[str] = None, subject: Optional[str] = None, quiz_date: Optional[date] = None):
    values: dict = {"updated_at": _now()}
    if title is not None:
        values["title"] = title
    if subject is not None:
        values["subject"] = subject
    if quiz_date is not None:
        values["quiz_date"] = quiz_date
    async with get_sessionmaker()() as db:
        await db.execute(update(Quiz).where(Quiz.id == session_id).values(**values))
        await db.commit()


async def delete_session(session_id: int):
    async with get_sessionmaker()() as db:
        await db.execute(delete(Quiz).where(Quiz.id == session_id))
        await db.commit()


async def delete_all_sessions(teacher_id: Optional[int] = None) -> int:
    """Delete every quiz for a teacher (or all teachers if None), cascading to every
    piece of derived data: results, reports, weaknesses/recommendations, answer grids,
    parsed data, and raw files. Returns the number of quizzes deleted."""
    async with get_sessionmaker()() as db:
        stmt = delete(Quiz)
        if teacher_id is not None:
            stmt = stmt.where(Quiz.teacher_id == teacher_id)
        result = await db.execute(stmt)
        await db.commit()
        return result.rowcount or 0


# =============================================================================
# Parsed data
# =============================================================================

async def save_parsed_data(
    session_id: int,
    question_num: int,
    question_str: str,
    answer: str,
    main_category: str,
    tags: list,
    options: Optional[list] = None,
) -> None:
    """Upsert a ParsedData row by (quiz_id, question_num).

    Uses ON CONFLICT instead of a plain insert so concurrent/retried uploads
    for the same quiz can't create duplicate rows per question number.
    """
    async with get_sessionmaker()() as db:
        stmt = pg_insert(ParsedData).values(
            quiz_id=session_id,
            question_num=question_num,
            question_str=question_str,
            options=options or [],
            answer=answer,
            main_category=main_category,
            tags=tags,
        )
        stmt = stmt.on_conflict_do_update(
            constraint="uq_parsed_data_quiz_question",
            set_={
                "question_str": stmt.excluded.question_str,
                "options": stmt.excluded.options,
                "answer": stmt.excluded.answer,
                "main_category": stmt.excluded.main_category,
                "tags": stmt.excluded.tags,
            },
        )
        await db.execute(stmt)
        await db.commit()


async def get_parsed_data(session_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(ParsedData)
            .where(ParsedData.quiz_id == session_id)
            .order_by(ParsedData.question_num)
        )
        return [_row(p) for p in res.scalars().all()]


async def delete_parsed_data(session_id: int, data_type: str = ""):
    """Delete all ParsedData rows for a session. data_type accepted but unused (no such column)."""
    async with get_sessionmaker()() as db:
        await db.execute(delete(ParsedData).where(ParsedData.quiz_id == session_id))
        await db.commit()


async def update_parsed_data_categories(
    session_id: int, categories: list[dict]
) -> None:
    """Bulk-update main_category + tags for parsed questions.

    `categories` is a list of {question_num, main_category, tags}.
    """
    async with get_sessionmaker()() as db:
        for entry in categories:
            await db.execute(
                update(ParsedData)
                .where(ParsedData.quiz_id == session_id)
                .where(ParsedData.question_num == entry["question_num"])
                .values(
                    main_category=entry.get("main_category", ""),
                    tags=entry.get("tags", []),
                )
            )
        await db.commit()


# =============================================================================
# Question bank (shared across all teachers)
# =============================================================================

async def save_question_bank_batch(
    quiz_id: int,
    teacher_id: int,
    questions: list[dict],
) -> int:
    """Insert categorized questions into the shared question bank.

    `questions` is a list of {question_text, answer, main_category, tags}.
    Skips questions with empty question_text and exact-text duplicates.
    Returns the number of new questions inserted.
    """
    texts = [q["question_text"] for q in questions if q.get("question_text")]
    if not texts:
        return 0

    async with get_sessionmaker()() as db:
        result = await db.execute(
            select(QuestionBank.question_text).where(QuestionBank.question_text.in_(texts))
        )
        existing = {row[0] for row in result}

        inserted = 0
        for q in questions:
            if not q.get("question_text") or q["question_text"] in existing:
                continue
            db.add(QuestionBank(
                quiz_id=quiz_id,
                teacher_id=teacher_id,
                question_text=q["question_text"],
                answer=q.get("answer", ""),
                main_category=q.get("main_category", ""),
                tags=q.get("tags", []),
            ))
            inserted += 1
        await db.commit()
    return inserted


async def query_question_bank(
    main_category: str,
    tags: list[str],
    limit: int = 30,
) -> list[dict]:
    """Return questions matching a category (and optionally any of the given tags),
    randomly sampled from the full matching set.

    If tags are specified, returns questions that contain AT LEAST ONE matching tag.
    Falls back to category-only matches when no tag-matched questions are found.
    """
    async with get_sessionmaker()() as db:
        # Fetch every question in the category — capping this to a
        # created_at-ordered window before filtering by tag meant older
        # (but still tag-matching) questions were never even considered, and
        # the same newest slice got returned on every call regardless of any
        # later shuffle. Random sampling needs to see the whole pool first.
        res = await db.execute(
            select(QuestionBank).where(QuestionBank.main_category == main_category)
        )
        rows = [_row(r) for r in res.scalars().all()]

        if tags:
            tag_set = set(tags)
            matched = [r for r in rows if tag_set & set(r.get("tags") or [])]
            if matched:
                random.shuffle(matched)
                return matched[:limit]

        random.shuffle(rows)
        return rows[:limit]


# ---- Answer grid (own table — one confirmed grid per quiz) ----

async def save_answer_grid(session_id: int, grid_dict: dict) -> int:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(AnswerGridRecord).where(AnswerGridRecord.quiz_id == session_id)
        )
        existing = res.scalar_one_or_none()
        if existing:
            existing.grid_json = grid_dict
            flag_modified(existing, "grid_json")
            await db.commit()
            return existing.id
        row = AnswerGridRecord(quiz_id=session_id, grid_json=grid_dict)
        db.add(row)
        await db.commit()
        return 0


async def get_answer_grid(session_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(AnswerGridRecord.grid_json).where(AnswerGridRecord.quiz_id == session_id)
        )
        row = res.scalar_one_or_none()
        return row if row else None


async def delete_answer_grid(session_id: int):
    async with get_sessionmaker()() as db:
        await db.execute(delete(AnswerGridRecord).where(AnswerGridRecord.quiz_id == session_id))
        await db.commit()


# =============================================================================
# Raw files (metadata; bytes live in Supabase Storage)
# =============================================================================

async def save_raw_file(
    session_id: int,
    data_type: str,
    file_name: str,
    content_type: str,
    size_bytes: int,
    storage_path: str,
) -> int:
    async with get_sessionmaker()() as db:
        row = RawFile(
            quiz_id=session_id,
            data_type=data_type,
            file_name=file_name,
            content_type=content_type,
            size_bytes=size_bytes,
            storage_path=storage_path,
        )
        db.add(row)
        await db.flush()
        row_id = row.id
        await db.commit()
        return row_id


async def get_raw_files(session_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(RawFile).where(RawFile.quiz_id == session_id).order_by(RawFile.created_at)
        )
        return [_row(f) for f in res.scalars().all()]


# =============================================================================
# Students (first-class, matched across quizzes by normalized name)
# =============================================================================

async def save_grid_results_batch(
    teacher_id: int, quiz_id: int, students: list[tuple[str, float]]
) -> list[dict]:
    """Upsert a Student + StudentResult row for every student in a grid.

    Runs in a single DB session instead of one per student (as
    match_or_create_student + save_student_result do when called in a loop) —
    this is called on every answer-grid save and every save-to-db, so for a
    class of 30 it turned ~60 sequential connection round-trips into 1.

    Each student is wrapped in its own savepoint so one bad row (a data quirk,
    a constraint violation) is skipped without rolling back everyone else in
    the same batch — matching the old per-student try/except isolation.
    Never overwrites an existing result's weaknesses/study_recommendations.
    """
    saved: list[dict] = []
    async with get_sessionmaker()() as db:
        for name, score in students:
            try:
                async with db.begin_nested():
                    norm = normalize_name(name)
                    res = await db.execute(
                        select(Student)
                        .where(Student.teacher_id == teacher_id)
                        .where(Student.normalized_name == norm)
                    )
                    student = res.scalar_one_or_none()
                    if not student:
                        student = Student(teacher_id=teacher_id, name=name, normalized_name=norm)
                        db.add(student)
                        await db.flush()
                    student_id = student.id

                    res2 = await db.execute(
                        select(StudentResult)
                        .where(StudentResult.quiz_id == quiz_id)
                        .where(StudentResult.student_id == student_id)
                    )
                    existing = res2.scalar_one_or_none()
                    if existing:
                        existing.score = score
                    else:
                        db.add(StudentResult(quiz_id=quiz_id, student_id=student_id, score=score))

                saved.append({"student_id": student_id, "name": name, "score": score})
            except Exception:
                continue  # isolate this student's failure from the rest of the batch
        await db.commit()
    return saved


async def match_or_create_student(
    teacher_id: int, name: str, external_id: Optional[str] = None
) -> int:
    """Return the student id for this teacher, creating the row if new."""
    norm = normalize_name(name)
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Student)
            .where(Student.teacher_id == teacher_id)
            .where(Student.normalized_name == norm)
        )
        student = res.scalar_one_or_none()
        if student:
            if external_id and not student.external_id:
                student.external_id = external_id
                await db.commit()
            return student.id
        student = Student(
            teacher_id=teacher_id, name=name, normalized_name=norm, external_id=external_id
        )
        db.add(student)
        await db.flush()
        student_id = student.id
        await db.commit()
        return student_id


async def update_student_email(student_id: int, teacher_id: int, email: str) -> bool:
    """Set a student's email (for practice-quiz delivery). Returns False if
    the student doesn't exist or isn't owned by this teacher.
    """
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Student)
            .where(Student.id == student_id)
            .where(Student.teacher_id == teacher_id)
        )
        student = res.scalar_one_or_none()
        if not student:
            return False
        student.email = email.strip() or None
        await db.commit()
        return True


async def get_students(teacher_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Student).where(Student.teacher_id == teacher_id).order_by(Student.name)
        )
        return [_row(s) for s in res.scalars().all()]


async def get_student(student_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        student = await db.get(Student, student_id)
        return _row(student) if student else None


async def save_student_ai_summary(
    student_id: int, summary: str, focus_areas: str, study_recommendations: str
) -> None:
    """Cache the cross-quiz AI synthesis on the student row.

    Refreshed whenever a new single-quiz report is generated for this
    student, so the dashboard can read it instantly instead of calling the
    LLM on every view.
    """
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Student)
            .where(Student.id == student_id)
            .values(
                ai_summary=summary,
                ai_focus_areas=focus_areas,
                ai_study_recommendations=study_recommendations,
                ai_summary_updated_at=_now(),
            )
        )
        await db.commit()


async def save_student_result(
    quiz_id: int,
    student_id: int,
    score: float,
    weaknesses: Optional[str] = None,
    study_recommendations: Optional[str] = None,
) -> int:
    """Upsert one student's result for a quiz (unique per quiz+student)."""
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(StudentResult)
            .where(StudentResult.quiz_id == quiz_id)
            .where(StudentResult.student_id == student_id)
        )
        existing = res.scalar_one_or_none()
        if existing:
            existing.score = score
            if weaknesses is not None:
                existing.weaknesses = weaknesses
            if study_recommendations is not None:
                existing.study_recommendations = study_recommendations
            await db.commit()
            return existing.id
        row = StudentResult(
            quiz_id=quiz_id,
            student_id=student_id,
            score=score,
            weaknesses=weaknesses,
            study_recommendations=study_recommendations,
        )
        db.add(row)
        await db.flush()
        row_id = row.id
        await db.commit()
        return row_id


async def wipe_quiz_data(quiz_id: int) -> dict:
    """Delete all data for a quiz (results, reports, grids, parsed_data, files) but keep the quiz record."""
    async with get_sessionmaker()() as db:
        r1 = await db.execute(delete(Report).where(Report.quiz_id == quiz_id))
        r2 = await db.execute(delete(StudentResult).where(StudentResult.quiz_id == quiz_id))
        r3 = await db.execute(delete(AnswerGridRecord).where(AnswerGridRecord.quiz_id == quiz_id))
        r4 = await db.execute(delete(ParsedData).where(ParsedData.quiz_id == quiz_id))
        r5 = await db.execute(delete(RawFile).where(RawFile.quiz_id == quiz_id))
        await db.commit()
        return {
            "reports": r1.rowcount,
            "results": r2.rowcount,
            "grids": r3.rowcount,
            "parsed_data": r4.rowcount,
            "files": r5.rowcount,
        }


async def wipe_question_bank(teacher_id: int | None = None) -> int:
    """Delete question bank rows. If teacher_id given, only that teacher's rows; else all rows."""
    async with get_sessionmaker()() as db:
        q = delete(QuestionBank)
        if teacher_id is not None:
            q = q.where(QuestionBank.teacher_id == teacher_id)
        r = await db.execute(q)
        await db.commit()
        return r.rowcount


async def delete_student_result(quiz_id: int, student_id: int) -> bool:
    """Delete one student's result for a quiz. Returns True if a row was deleted."""
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(StudentResult)
            .where(StudentResult.quiz_id == quiz_id)
            .where(StudentResult.student_id == student_id)
        )
        row = res.scalar_one_or_none()
        if not row:
            return False
        await db.delete(row)
        await db.commit()
        return True


def _normalize_name_simple(s: str) -> str:
    return " ".join((s or "").strip().lower().split())


async def get_student_timeline(student_id: int) -> list[dict]:
    """One student's results across all quizzes, oldest first — the trajectory.

    Enriches each entry with total_questions, correct_count, and per-answer data
    derived from the stored answer grid (answer_grids table).
    """
    async with get_sessionmaker()() as db:
        # Fetch results + quiz metadata + student name in one query
        res = await db.execute(
            select(StudentResult, Quiz.title, Quiz.subject, Quiz.created_at, Quiz.quiz_date, Student.name)
            .join(Quiz, StudentResult.quiz_id == Quiz.id)
            .join(Student, StudentResult.student_id == Student.id)
            .where(StudentResult.student_id == student_id)
            .where(Quiz.teacher_id == Student.teacher_id)
            # Order by the exam's actual date (quiz_date), not upload order —
            # a quiz entered later for an earlier exam date must still land in
            # its correct chronological spot on the student's graph. Fall back
            # to created_at's date for quizzes with no quiz_date set, and use
            # created_at as a tiebreaker for same-day quizzes.
            .order_by(func.coalesce(Quiz.quiz_date, cast(Quiz.created_at, Date)), Quiz.created_at)
        )
        rows = res.all()
        if not rows:
            return []

        # Batch-load answer grids for all quizzes in one query
        quiz_ids = [sr.quiz_id for sr, *_ in rows]
        grid_res = await db.execute(
            select(AnswerGridRecord.quiz_id, AnswerGridRecord.grid_json)
            .where(AnswerGridRecord.quiz_id.in_(quiz_ids))
        )
        grids: dict[int, dict] = {qid: gj for qid, gj in grid_res.all()}

        out = []
        for sr, title, subject, created, quiz_date, student_name in rows:
            d = _row(sr)
            d["quiz_title"] = title
            d["quiz_subject"] = subject
            d["quiz_created_at"] = created
            d["quiz_date"] = str(quiz_date) if quiz_date else None

            # Derive correct_count, total_questions, answers from answer grid
            grid_json = grids.get(sr.quiz_id)
            if grid_json and isinstance(grid_json, dict):
                official: list = grid_json.get("official_answers") or []
                total = grid_json.get("total_questions") or len(official)
                norm = _normalize_name_simple(student_name)
                for gs in grid_json.get("students") or []:
                    if _normalize_name_simple(gs.get("name", "")) == norm:
                        raw_answers = gs.get("answers") or []
                        correct = 0
                        for j, cell in enumerate(raw_answers):
                            eff = official[j] if cell == "=" and j < len(official) else cell
                            if eff and j < len(official) and eff == official[j]:
                                correct += 1
                        d["total_questions"] = total
                        d["correct_count"] = correct
                        d["answers"] = {"answers": raw_answers}
                        break
                else:
                    # Student not found in grid (name mismatch) — fall back to score
                    d["total_questions"] = total or 0
                    d["correct_count"] = None
                    d["answers"] = {"answers": []}
            else:
                d["total_questions"] = 0
                d["correct_count"] = None
                d["answers"] = {"answers": []}

            out.append(d)
        return out


# =============================================================================
# Prompts
# =============================================================================

async def save_prompt(user_id: int, name: str, content: str, is_default: bool = False) -> int:
    async with get_sessionmaker()() as db:
        row = Prompt(teacher_id=user_id, name=name, content=content, is_default=is_default)
        db.add(row)
        await db.flush()
        row_id = row.id
        await db.commit()
        return row_id


async def get_prompts(user_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Prompt)
            .where((Prompt.teacher_id == user_id) | (Prompt.is_default.is_(True)))
            .order_by(Prompt.is_default.desc(), Prompt.updated_at.desc())
        )
        return [_row(p) for p in res.scalars().all()]


async def get_all_prompts() -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Prompt, Teacher.email)
            .join(Teacher, Prompt.teacher_id == Teacher.id)
            .order_by(Prompt.updated_at.desc())
        )
        out = []
        for p, email in res.all():
            d = _row(p)
            d["author_email"] = email
            out.append(d)
        return out


async def get_prompt(prompt_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        p = await db.get(Prompt, prompt_id)
        if not p:
            return None
        d = _row(p)
        d["user_id"] = p.teacher_id  # back-compat key
        return d


async def update_prompt(prompt_id: int, name: str, content: str):
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Prompt)
            .where(Prompt.id == prompt_id)
            .values(name=name, content=content, updated_at=_now())
        )
        await db.commit()


async def delete_prompt(prompt_id: int):
    async with get_sessionmaker()() as db:
        await db.execute(delete(Prompt).where(Prompt.id == prompt_id))
        await db.commit()


# =============================================================================
# Reports
# =============================================================================

async def save_report(
    session_id: int,
    prompt_id: Optional[int],
    html_content: str,
    student_id: Optional[int] = None,
    model: str = "",
) -> int:
    async with get_sessionmaker()() as db:
        row = Report(
            quiz_id=session_id,
            prompt_id=prompt_id,
            student_id=student_id,
            html_content=html_content,
            model=model,
        )
        db.add(row)
        await db.flush()
        row_id = row.id
        await db.commit()
        return row_id


async def get_reports(session_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Report).where(Report.quiz_id == session_id).order_by(Report.created_at.desc())
        )
        return [_row(r) for r in res.scalars().all()]


async def get_report(report_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        r = await db.get(Report, report_id)
        return _row(r) if r else None


# =============================================================================
# Analytics (admin dashboard)
# =============================================================================

async def get_stats(teacher_id: Optional[int] = None) -> dict:
    """Global stats (admin) or stats scoped to one teacher's own data."""
    async with get_sessionmaker()() as db:
        async def count(stmt):
            return (await db.execute(stmt)).scalar_one()

        if teacher_id is None:
            return {
                "teachers": await count(select(func.count(Teacher.id))),
                "students": await count(select(func.count(Student.id))),
                "quizzes": await count(select(func.count(Quiz.id)).where(Quiz.status != "draft")),
                "reports": await count(select(func.count(Report.id))),
                "student_results": await count(select(func.count(StudentResult.id))),
                "prompts": await count(select(func.count(Prompt.id))),
            }

        # Scoped to one teacher: own students/quizzes/prompts directly; reports and
        # results are owned transitively through the teacher's own quizzes.
        own_quiz = select(Quiz.id).where(Quiz.teacher_id == teacher_id).scalar_subquery()
        return {
            "teachers": 1,
            "students": await count(
                select(func.count(Student.id)).where(Student.teacher_id == teacher_id)
            ),
            "quizzes": await count(
                select(func.count(Quiz.id)).where(Quiz.teacher_id == teacher_id, Quiz.status != "draft")
            ),
            "reports": await count(
                select(func.count(Report.id)).where(Report.quiz_id.in_(own_quiz))
            ),
            "student_results": await count(
                select(func.count(StudentResult.id)).where(StudentResult.quiz_id.in_(own_quiz))
            ),
            "prompts": await count(
                select(func.count(Prompt.id)).where(Prompt.teacher_id == teacher_id)
            ),
        }


async def admin_list_students(teacher_id: Optional[int] = None, limit: int = 500) -> list[dict]:
    """Students with quiz count + avg score; all teachers (admin) or one teacher's own."""
    async with get_sessionmaker()() as db:
        stmt = (
            select(Student, Teacher.email)
            .join(Teacher, Student.teacher_id == Teacher.id)
            .order_by(Student.name)
            .limit(limit)
        )
        if teacher_id is not None:
            stmt = stmt.where(Student.teacher_id == teacher_id)
        res = await db.execute(stmt)
        rows = res.all()

        # One aggregate query for all students instead of one round-trip per
        # student — the per-student version turned a class of 30 into 31
        # sequential DB calls before the list could render.
        student_ids = [s.id for s, _ in rows]
        agg_map: dict[int, tuple[int, Optional[float]]] = {}
        if student_ids:
            agg_res = await db.execute(
                select(
                    StudentResult.student_id,
                    func.count(StudentResult.id),
                    func.avg(StudentResult.score),
                )
                .where(StudentResult.student_id.in_(student_ids))
                .group_by(StudentResult.student_id)
            )
            agg_map = {sid: (count, avg) for sid, count, avg in agg_res.all()}

        out = []
        for s, email in rows:
            d = _row(s)
            d["teacher_email"] = email
            count, avg = agg_map.get(s.id, (0, None))
            d["result_count"] = count
            d["avg_score"] = round(float(avg), 1) if avg is not None else None
            out.append(d)
        return out


async def get_quiz_results(quiz_id: int) -> list[dict]:
    """All student results for a quiz, joined with student name (admin detail)."""
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(StudentResult, Student.name, Student.id)
            .join(Student, StudentResult.student_id == Student.id)
            .join(Quiz, StudentResult.quiz_id == Quiz.id)
            .where(StudentResult.quiz_id == quiz_id)
            # defense-in-depth: only students belonging to this quiz's teacher
            .where(Student.teacher_id == Quiz.teacher_id)
            .order_by(Student.name)
        )
        out = []
        for sr, name, sid in res.all():
            d = _row(sr)
            d["student_name"] = name
            d["student_id"] = sid
            out.append(d)
        return out


async def get_schema_version() -> Optional[str]:
    """Current Alembic migration revision applied to the DB."""
    from sqlalchemy import text

    from .db import active_schema

    schema = active_schema()
    table = f'"{schema}".alembic_version' if schema else "alembic_version"
    async with get_sessionmaker()() as db:
        try:
            res = await db.execute(text(f"SELECT version_num FROM {table} LIMIT 1"))
            row = res.first()
            return row[0] if row else None
        except Exception:
            return None


async def admin_list_sessions(teacher_id: Optional[int] = None, limit: int = 200) -> list[dict]:
    """Quizzes with author email + counts; all teachers (admin) or one teacher's own.

    Only includes status "saved" — a quiz becomes a record here only once the
    teacher finishes the whole flow and clicks "儲存至資料庫" in step 3.
    Uploaded-but-abandoned drafts and half-finished sessions (uploaded, maybe a
    few reports generated, but never explicitly saved) never show up here, so
    this list never contains a quiz with 0 reports.
    """
    async with get_sessionmaker()() as db:
        stmt = (
            select(Quiz, Teacher.email)
            .join(Teacher, Quiz.teacher_id == Teacher.id)
            .where(Quiz.status == "saved")
            .order_by(Quiz.created_at.desc())
            .limit(limit)
        )
        if teacher_id is not None:
            stmt = stmt.where(Quiz.teacher_id == teacher_id)
        res = await db.execute(stmt)
        rows = res.all()

        # One aggregate query for all quizzes instead of one COUNT round-trip
        # per quiz — same fix as admin_list_students' N+1.
        quiz_ids = [q.id for q, _ in rows]
        report_counts: dict[int, int] = {}
        if quiz_ids:
            count_res = await db.execute(
                select(Report.quiz_id, func.count(Report.id))
                .where(Report.quiz_id.in_(quiz_ids))
                .group_by(Report.quiz_id)
            )
            report_counts = dict(count_res.all())

        out = []
        for q, email in rows:
            d = _row(q)
            d["teacher_email"] = email
            d["report_count"] = report_counts.get(q.id, 0)
            out.append(d)
        return out


# =============================================================================
# Practice quizzes (individualized, student-facing, no login required)
# =============================================================================

async def create_practice_quiz(
    teacher_id: int,
    student_id: int,
    title: str,
    questions: list[dict],
    feed_to_dashboard: bool = True,
) -> dict:
    """Create an individualized practice quiz for one student.

    `questions` is a list of {number, question, options: {A,B,C,D}, answer,
    category}. The share_token is the sole credential for the public
    student-facing link — unguessable and unique, no separate login.
    """
    async with get_sessionmaker()() as db:
        quiz = PracticeQuiz(
            teacher_id=teacher_id,
            student_id=student_id,
            title=title or "Practice Quiz",
            share_token=secrets.token_urlsafe(24),
            questions=questions,
            feed_to_dashboard=feed_to_dashboard,
            status="sent",
        )
        db.add(quiz)
        await db.flush()
        row = _row(quiz)
        await db.commit()
        return row


async def get_practice_quiz(practice_quiz_id: int, teacher_id: int) -> Optional[dict]:
    """Teacher-owned lookup (for the management view), scoped to the caller."""
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(PracticeQuiz)
            .where(PracticeQuiz.id == practice_quiz_id)
            .where(PracticeQuiz.teacher_id == teacher_id)
        )
        quiz = res.scalar_one_or_none()
        return _row(quiz) if quiz else None


async def get_practice_quiz_by_token(share_token: str) -> Optional[dict]:
    """Public lookup by share token — the token itself is the authorization,
    there is no teacher_id scoping here by design.
    """
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(PracticeQuiz).where(PracticeQuiz.share_token == share_token)
        )
        quiz = res.scalar_one_or_none()
        return _row(quiz) if quiz else None


async def list_practice_quizzes(teacher_id: int, student_id: Optional[int] = None) -> list[dict]:
    async with get_sessionmaker()() as db:
        stmt = (
            select(PracticeQuiz)
            .where(PracticeQuiz.teacher_id == teacher_id)
            .order_by(PracticeQuiz.created_at.desc())
        )
        if student_id is not None:
            stmt = stmt.where(PracticeQuiz.student_id == student_id)
        res = await db.execute(stmt)
        return [_row(q) for q in res.scalars().all()]


async def submit_practice_quiz(share_token: str, answers: dict[str, str]) -> Optional[dict]:
    """Grade and record a single-attempt submission.

    Returns None if the token doesn't match any quiz. If the quiz was already
    completed, returns the existing (already-graded) row unchanged rather
    than re-grading — single attempt only, resubmits are a no-op.
    """
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(PracticeQuiz).where(PracticeQuiz.share_token == share_token)
        )
        quiz = res.scalar_one_or_none()
        if not quiz:
            return None
        if quiz.status == "completed":
            return _row(quiz)

        questions = quiz.questions or []
        correct = 0
        for q in questions:
            num = str(q.get("number"))
            given = (answers.get(num) or "").strip().upper()
            if given and given == (q.get("answer") or "").strip().upper():
                correct += 1
        total = len(questions)
        score = round((correct / total) * 100, 1) if total else 0.0

        quiz.answers = answers
        quiz.score = score
        quiz.correct_count = correct
        quiz.status = "completed"
        quiz.completed_at = _now()
        await db.commit()
        result = _row(quiz)

    if result.get("feed_to_dashboard"):
        await _feed_practice_quiz_to_dashboard(result)

    return result


async def _feed_practice_quiz_to_dashboard(pq: dict) -> None:
    """Record this practice attempt as a StudentResult (under a lightweight,
    synthetic Quiz row) so it slots into the student's existing timeline,
    average score, and AI-summary pipeline for free.

    Uses status="practice" (not "saved") so it never shows up in the
    teacher's 我的資料 exam/session list — that list is for real uploaded
    exams, and a practice quiz has no report to show there.
    """
    async with get_sessionmaker()() as db:
        quiz = Quiz(
            teacher_id=pq["teacher_id"],
            title=pq.get("title") or "Practice Quiz",
            status="practice",
            quiz_date=_now().date(),
        )
        db.add(quiz)
        await db.flush()
        db.add(StudentResult(
            quiz_id=quiz.id,
            student_id=pq["student_id"],
            score=pq["score"],
        ))
        await db.commit()