ClassLensPortal

Sleeping

ClassLensPortal / chatkit /backend /app /database.py
chih.yikuan
Claude Opus 4.8 (1M context)
feat: add "Sign in with Google" (Google SSO) for teachers/admins
6726707 about 2 months ago
81.2 kB
"""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, timedelta, timezone
from typing import Awaitable, Callable, 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 .taxonomy import MAIN_CATEGORIES
from .models import (
    AnswerGrid as AnswerGridRecord,
    Base,
    IssueReport,
    LoginEvent,
    ParsedData,
    PracticeQuiz,
    Prompt,
    Quiz,
    QuestionBank,
    RawFile,
    Report,
    Student,
    StudentResult,
    Teacher,
)


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


def _aware(dt: datetime) -> datetime:
    """Postgres (DateTime(timezone=True)) round-trips tz-aware datetimes;
    SQLite always hands back naive ones regardless of column type. Anything
    comparing a DB-fetched datetime against another in plain Python (not a
    SQL WHERE clause, which each dialect handles consistently on its own)
    needs both sides on the same footing — assume naive values are UTC,
    since every datetime this app writes is generated via _now()."""
    return dt if dt.tzinfo else dt.replace(tzinfo=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 get_user_by_google_id(google_id: str) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(select(Teacher).where(Teacher.google_id == google_id))
        teacher = res.scalar_one_or_none()
        return _row(teacher) if teacher else None


async def create_google_user(
    email: str, google_id: str, full_name: str = "", is_admin: bool = False
) -> int:
    """Create a Google-only teacher (no password) and return its id."""
    async with get_sessionmaker()() as db:
        teacher = Teacher(
            email=email,
            password_hash=None,
            google_id=google_id,
            display_name=full_name or email.split("@")[0],
            full_name=full_name,
            is_admin=is_admin,
        )
        db.add(teacher)
        await db.flush()
        teacher_id = teacher.id
        await db.commit()
        return teacher_id


async def link_google_id(user_id: int, google_id: str) -> None:
    """Attach a Google account to an existing (password) teacher."""
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Teacher).where(Teacher.id == user_id).values(google_id=google_id)
        )
        await db.commit()


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 get_teacher_vocabulary(teacher_id: int) -> list[dict]:
    async with get_sessionmaker()() as db:
        teacher = await db.get(Teacher, teacher_id)
        return list(teacher.vocabulary) if teacher else []


async def set_teacher_vocabulary(teacher_id: int, words: list[dict]) -> list[dict]:
    async with get_sessionmaker()() as db:
        await db.execute(
            update(Teacher).where(Teacher.id == teacher_id).values(vocabulary=words)
        )
        await db.commit()
        return words


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 get_admin_emails() -> list[str]:
    """Every teacher with is_admin=True in the DB — used as the recipient
    list for issue-report notifications. Doesn't include email-based admin
    overrides granted only in code (see auth._apply_admin_override); callers
    that need those too should merge them in separately."""
    async with get_sessionmaker()() as db:
        res = await db.execute(select(Teacher.email).where(Teacher.is_admin.is_(True)))
        return [e for (e,) in res.all()]


async def create_issue_report(
    reporter_email: str,
    category: str,
    subject: str,
    description: str,
    page_context: str = "",
) -> dict:
    async with get_sessionmaker()() as db:
        report = IssueReport(
            reporter_email=reporter_email,
            category=category,
            subject=subject,
            description=description,
            page_context=page_context,
        )
        db.add(report)
        await db.flush()
        await db.refresh(report)
        row = _row(report)
        await db.commit()
        return row


async def list_open_issue_reports() -> list[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(IssueReport)
            .where(IssueReport.status == "open")
            .order_by(IssueReport.created_at.desc())
        )
        return [_row(r) for r in res.scalars().all()]


async def dismiss_issue_report(report_id: int) -> bool:
    async with get_sessionmaker()() as db:
        res = await db.execute(
            update(IssueReport)
            .where(IssueReport.id == report_id, IssueReport.status == "open")
            .values(status="dismissed", dismissed_at=_now())
        )
        await db.commit()
        return res.rowcount > 0  # type: ignore[attr-defined]


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]:
    """A teacher's saved quiz records — filtered to status == "saved" so an
    upload that's mid-flow or was abandoned before step 3's explicit save
    never appears as a record here, matching admin_list_sessions."""
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(Quiz)
            .where(Quiz.teacher_id == user_id, Quiz.status == "saved")
            .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()


# =============================================================================
# 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 save_raw_file(
    quiz_id: int,
    data_type: str,
    file_name: str,
    content_type: str,
    size_bytes: int,
    storage_path: str,
) -> int:
    """Record an uploaded file's on-disk location so the original scan/PDF
    can be viewed later instead of only the OCR-extracted structured data."""
    async with get_sessionmaker()() as db:
        raw_file = RawFile(
            quiz_id=quiz_id,
            data_type=data_type,
            file_name=file_name,
            content_type=content_type,
            size_bytes=size_bytes,
            storage_path=storage_path,
        )
        db.add(raw_file)
        await db.commit()
        await db.refresh(raw_file)
        return raw_file.id


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


async def get_raw_file(raw_file_id: int) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(select(RawFile).where(RawFile.id == raw_file_id))
        f = res.scalar_one_or_none()
        return _row(f) if f else None


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],
) -> dict[str, int]:
    """Insert categorized questions into the shared question bank.
    `questions` is a list of {question_text, answer, main_category, tags,
    passage}. `passage` is optional — for a passage/stimulus-based set (one
    reading passage with several questions attached), pass the *same*
    passage text on each sibling question's dict; each becomes its own row
    (the model has no separate passages table — see QuestionBank.passage),
    so the text is duplicated across siblings, but every row stays
    self-contained and query_question_bank/generation code needs no
    special-casing for "which questions share a passage."
    Skips questions with empty question_text and exact-text duplicates.
    Completeness (answer present, no unresolved image reference) is not
    checked here — that heuristic was rejecting verified-complete uploads,
    so it was removed from this save path. The equivalent safeguard for
    generation-time context sufficiency still lives in
    question_generator._has_answerable_context. Returns
    {"inserted", "duplicates", "incomplete"} for API-shape compatibility;
    "incomplete" is always 0 now.
    """
    texts = [q["question_text"] for q in questions if q.get("question_text")]
    if not texts:
        return {"inserted": 0, "duplicates": 0, "incomplete": 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
        duplicates = 0
        for q in questions:
            if not q.get("question_text"):
                continue
            if q["question_text"] in existing:
                duplicates += 1
                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", []),
                    passage=q.get("passage") or None,
                )
            )
            inserted += 1
        await db.commit()
    return {"inserted": inserted, "duplicates": duplicates, "incomplete": 0}


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]


async def backfill_question_bank_answers(
    quiz_id: int, answers_by_text: dict[str, str]
) -> int:
    """Fill in `answer` on question_bank rows from this quiz's confirmed answer grid.
    Called once a teacher confirms the answer grid (the point the correct
    answer per question becomes actually known) — lets question generation
    skip the LLM answer-check for these rows from then on. Only overwrites
    rows that don't already have an answer; matches by exact question_text
    since that's how these rows were seeded (see `save_question_bank_batch`).
    Returns the number of rows updated.
    """
    if not answers_by_text:
        return 0
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(QuestionBank.id, QuestionBank.question_text)
            .where(QuestionBank.quiz_id == quiz_id)
            .where((QuestionBank.answer == "") | (QuestionBank.answer.is_(None)))
        )
        updated = 0
        for row_id, text in res.all():
            answer = answers_by_text.get((text or "").strip())
            if answer:
                await db.execute(
                    update(QuestionBank)
                    .where(QuestionBank.id == row_id)
                    .values(answer=answer)
                )
                updated += 1
        await db.commit()
    return updated


async def save_question_bank_passage(question_id: int, passage: str) -> None:
    """Backfill a generated reading passage onto a question_bank row.
    Best-effort — called after the LLM writes a passage for a row that was
    sampled without one, so future draws of the same row reuse it instead of
    paying to regenerate.
    """
    async with get_sessionmaker()() as db:
        await db.execute(
            update(QuestionBank)
            .where(QuestionBank.id == question_id)
            .values(passage=passage)
        )
        await db.commit()


# ---- 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()


# =============================================================================
# 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_student_by_username(username: str) -> Optional[dict]:
    async with get_sessionmaker()() as db:
        res = await db.execute(select(Student).where(Student.username == username))
        student = res.scalar_one_or_none()
        return _row(student) if student else None


async def username_taken(username: str) -> bool:
    async with get_sessionmaker()() as db:
        res = await db.execute(select(Student.id).where(Student.username == username))
        return res.scalar_one_or_none() is not None


async def set_student_login(
    student_id: int, teacher_id: int, username: str, password_encrypted: str
) -> bool:
    """Issue or reset a student's login credentials. 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.username = username
        student.password_encrypted = password_encrypted
        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,  # type: ignore[attr-defined]
            "results": r2.rowcount,  # type: ignore[attr-defined]
            "grids": r3.rowcount,  # type: ignore[attr-defined]
            "parsed_data": r4.rowcount,  # type: ignore[attr-defined]
            "files": r5.rowcount,  # type: ignore[attr-defined]
        }


async def list_question_bank(
    category: str | None = None,
    tag: str | None = None,
    search: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[dict], int]:
    """Browse the shared question bank for the 題庫管理 UI.
    Filters by exact category, a single tag (checked in Python since `tags`
    is a JSON array column), and/or a substring search on question_text.
    Returns (page of rows, total matching count) — rows sorted by category
    then newest first, so browsing naturally groups by tag/category.
    """
    async with get_sessionmaker()() as db:
        q = select(QuestionBank)
        if category:
            q = q.where(QuestionBank.main_category == category)
        if search:
            q = q.where(QuestionBank.question_text.ilike(f"%{search}%"))
        res = await db.execute(
            q.order_by(QuestionBank.main_category, QuestionBank.created_at.desc())
        )
        rows = [_row(r) for r in res.scalars().all()]

    if tag:
        rows = [r for r in rows if tag in (r.get("tags") or [])]

    return rows[offset : offset + limit], len(rows)


async def list_question_bank_tags() -> dict[str, list[str]]:
    """Distinct (category -> tags actually in use) for the 題庫管理 filter UI.
    Always includes all 6 main categories, even ones with zero questions
    right now (e.g. a teacher just cleared out everything in one category) —
    otherwise that category's filter button would vanish entirely instead of
    just showing empty, which reads as "this category doesn't exist" rather
    than "this category has nothing in it yet."
    """
    by_cat: dict[str, set[str]] = {cat: set() for cat in MAIN_CATEGORIES}
    async with get_sessionmaker()() as db:
        res = await db.execute(select(QuestionBank.main_category, QuestionBank.tags))
        for cat, tags in res.all():
            bucket = by_cat.setdefault(cat or "", set())
            bucket.update(tags or [])
    return {cat: sorted(tags) for cat, tags in by_cat.items()}


async def delete_question_bank_row(row_id: int) -> bool:
    async with get_sessionmaker()() as db:
        res = await db.execute(select(QuestionBank).where(QuestionBank.id == row_id))
        row = res.scalar_one_or_none()
        if not row:
            return False
        await db.delete(row)
        await db.commit()
        return True


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  # type: ignore[attr-defined]


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)
            # Only quizzes the teacher actually saved (or practice-quiz
            # completions) belong on the dashboard — a report generated in
            # Step 3 writes a StudentResult immediately, before "Save to
            # Database" is clicked, so without this filter the dashboard
            # would show results for quizzes that never got saved.
            .where(Quiz.status.in_(["saved", "practice"]))
            # 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()}

        # Batch-load this student's own report id per quiz, for the report/raw-file
        # links on the dashboard — a quiz can have reports for other students too,
        # so this is scoped by student_id, not just quiz_id.
        report_res = await db.execute(
            select(Report.quiz_id, Report.id)
            .where(Report.quiz_id.in_(quiz_ids))
            .where(Report.student_id == student_id)
        )
        report_ids: dict[int, int] = {qid: rid for qid, rid in report_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
            d["report_id"] = report_ids.get(sr.quiz_id)

            # 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 == "saved")
                ),
                "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 == "saved"
                )
            ),
            "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 get_signup_growth() -> dict:
    """Cumulative teacher/student signups by day, for the admin analytics chart.
    Computed directly from each row's created_at rather than a separate
    snapshot table — signup date is immutable once a row exists, so this
    gives a fully accurate history (including before this feature shipped)
    with nothing to backfill or keep in sync.
    """
    async with get_sessionmaker()() as db:

        async def daily_counts(model) -> list[tuple[str, int]]:
            # func.date() (not cast(..., Date)) — casting a datetime column to
            # Date and selecting it directly trips a SQLAlchemy/SQLite result-
            # decoding bug (the cast is a no-op at the SQLite level, so the
            # value comes back as a full timestamp string that the Date type's
            # Python-side parser then rejects). func.date() truncates to
            # 'YYYY-MM-DD' at the SQL level instead, which is portable to
            # Postgres too and never hits that decode path.
            day = func.date(model.created_at)
            res = await db.execute(
                select(day, func.count()).group_by(day).order_by(day)
            )
            return [(str(d), c) for d, c in res.all()]

        def cumulative(rows: list[tuple[str, int]]) -> list[dict]:
            out = []
            total = 0
            for d, c in rows:
                total += c
                out.append({"date": d, "new": c, "total": total})
            return out

        return {
            "teachers": cumulative(await daily_counts(Teacher)),
            "students": cumulative(await daily_counts(Student)),
        }


async def record_login_event(role: str, user_id: int) -> None:
    """Append-only log of a successful login — see LoginEvent for why this
    is a separate table rather than reusing Teacher.last_login (which is
    overwritten each time, so it can't answer "how many distinct users
    logged in this week/month")."""
    async with get_sessionmaker()() as db:
        db.add(LoginEvent(role=role, user_id=user_id))
        await db.commit()


async def get_active_users(weeks: int = 12) -> dict:
    """Weekly active teachers/students (distinct users with >=1 login) over
    the last `weeks` weeks, for the admin analytics line chart.
    Bucketed in Python (not SQL date-trunc) — the two DB backends this app
    runs on (SQLite locally, Postgres in production) don't share a portable
    "start of ISO week" expression, and login-event volume is small enough
    that fetching raw rows and bucketing here is simpler than reconciling
    dialect-specific date functions.
    """
    cutoff = _now() - timedelta(weeks=weeks)
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(LoginEvent.role, LoginEvent.user_id, LoginEvent.created_at).where(
                LoginEvent.created_at >= cutoff
            )
        )
        rows = res.all()

    def week_start(dt: datetime) -> str:
        d = dt.date() if hasattr(dt, "date") else dt
        return (d - timedelta(days=d.weekday())).isoformat()

    buckets: dict[str, dict[str, set]] = {}
    for role, user_id, created_at in rows:
        wk = week_start(created_at)
        buckets.setdefault(wk, {"teacher": set(), "student": set()})[role].add(user_id)

    weeks_sorted = sorted(buckets)
    return {
        "teachers": [
            {"date": wk, "count": len(buckets[wk]["teacher"])} for wk in weeks_sorted
        ],
        "students": [
            {"date": wk, "count": len(buckets[wk]["student"])} for wk in weeks_sorted
        ],
    }


async def get_teacher_retention() -> dict:
    """Of teachers whose account is old enough to judge, what share came back
    and logged in again *after* their first week / first month — not just
    during initial signup/onboarding. Shows sustained use rather than
    one-time testing.
    A teacher only counts toward a window's denominator once their account
    has existed at least that long (a 3-day-old account can't yet prove or
    disprove week-1 retention), and toward the numerator if any login_event
    for them lands strictly after `created_at + window`.
    """
    now = _now()
    async with get_sessionmaker()() as db:
        teachers_res = await db.execute(select(Teacher.id, Teacher.created_at))
        teachers = [(tid, _aware(created_at)) for tid, created_at in teachers_res.all()]

        logins_res = await db.execute(
            select(LoginEvent.user_id, LoginEvent.created_at).where(
                LoginEvent.role == "teacher"
            )
        )
        logins_by_teacher: dict[int, list[datetime]] = {}
        for user_id, created_at in logins_res.all():
            logins_by_teacher.setdefault(user_id, []).append(_aware(created_at))

    def window_stats(days: int) -> dict:
        eligible = 0
        retained = 0
        for teacher_id, created_at in teachers:
            cutoff = created_at + timedelta(days=days)
            if cutoff > now:
                continue  # account not old enough to judge this window yet
            eligible += 1
            if any(t > cutoff for t in logins_by_teacher.get(teacher_id, [])):
                retained += 1
        pct = round(100 * retained / eligible, 1) if eligible else 0.0
        return {"eligible": eligible, "retained": retained, "pct": pct}

    return {
        "week": window_stats(7),
        "month": window_stats(30),
    }


async def get_assessments_processed(months: int = 6) -> dict:
    """Monthly counts of exams uploaded (saved quizzes) and student reports
    generated, for the admin analytics bar chart."""
    cutoff = _now() - timedelta(days=months * 31)
    async with get_sessionmaker()() as db:
        quiz_res = await db.execute(
            select(Quiz.created_at).where(
                Quiz.status == "saved", Quiz.created_at >= cutoff
            )
        )
        report_res = await db.execute(
            select(Report.created_at).where(Report.created_at >= cutoff)
        )

    def month_bucket(dt: datetime) -> str:
        return f"{dt.year:04d}-{dt.month:02d}"

    def counts_by_month(rows) -> dict[str, int]:
        out: dict[str, int] = {}
        for (dt,) in rows:
            key = month_bucket(dt)
            out[key] = out.get(key, 0) + 1
        return out

    quizzes_by_month = counts_by_month(quiz_res.all())
    reports_by_month = counts_by_month(report_res.all())
    all_months = sorted(set(quizzes_by_month) | set(reports_by_month))
    return {
        "months": [
            {
                "month": m,
                "quizzes_uploaded": quizzes_by_month.get(m, 0),
                "reports_generated": reports_by_month.get(m, 0),
            }
            for m in all_months
        ]
    }


async def mark_practice_quiz_opened(share_token: str) -> None:
    """Set opened_at the first time a student loads the quiz link — a no-op
    if already set, so repeated page loads/refreshes don't matter."""
    async with get_sessionmaker()() as db:
        await db.execute(
            update(PracticeQuiz)
            .where(
                PracticeQuiz.share_token == share_token,
                PracticeQuiz.opened_at.is_(None),
            )
            .values(opened_at=func.now())
        )
        await db.commit()


async def get_practice_quiz_funnel(teacher_id: Optional[int] = None) -> dict:
    """generated -> assigned -> opened -> completed counts for the admin
    funnel chart. "generated" and "assigned" are currently always equal (a
    PracticeQuiz row is created with status="sent" in the same insert, see
    create_practice_quiz) — kept as separate stages since sharing is a
    distinct product step even though today it's not a separate DB event."""
    async with get_sessionmaker()() as db:
        stmt = select(PracticeQuiz)
        if teacher_id is not None:
            stmt = stmt.where(PracticeQuiz.teacher_id == teacher_id)
        res = await db.execute(stmt)
        rows = res.scalars().all()

    generated = len(rows)
    assigned = sum(1 for r in rows if r.status in ("sent", "completed"))
    opened = sum(1 for r in rows if r.opened_at is not None)
    completed = sum(1 for r in rows if r.status == "completed")
    return {
        "generated": generated,
        "assigned": assigned,
        "opened": opened,
        "completed": completed,
    }


async def get_feedback_loop_funnel(teacher_id: Optional[int] = None) -> dict:
    """Exam result → weakness identified → practice generated → practice
    completed → later reassessment, for the admin funnel chart.
    The unit here is one student's result on one real exam (a
    StudentResult row on a Quiz with status="saved" and no
    source_practice_quiz_id — i.e. not the synthetic quiz row a completed
    practice quiz feeds into the dashboard as). PracticeQuiz has no FK back
    to the exam that triggered it, so each stage is correlated by student +
    timestamp ordering instead: the practice quiz counted for a given exam
    result is the earliest one generated for that student after the exam,
    and a "reassessment" is any later real exam for that same student after
    that practice was completed.
    """
    async with get_sessionmaker()() as db:
        quiz_stmt = select(Quiz.id, Quiz.created_at).where(
            Quiz.status == "saved", Quiz.source_practice_quiz_id.is_(None)
        )
        if teacher_id is not None:
            quiz_stmt = quiz_stmt.where(Quiz.teacher_id == teacher_id)
        quiz_res = await db.execute(quiz_stmt)
        quiz_created_at: dict[int, datetime] = {
            qid: _aware(dt) for qid, dt in quiz_res.all()
        }

        empty = {
            "uploaded": 0,
            "weakness_identified": 0,
            "practice_generated": 0,
            "practice_completed": 0,
            "reassessed": 0,
        }
        if not quiz_created_at:
            return empty

        result_res = await db.execute(
            select(
                StudentResult.quiz_id,
                StudentResult.student_id,
                StudentResult.weaknesses,
            ).where(StudentResult.quiz_id.in_(quiz_created_at.keys()))
        )
        results = result_res.all()
        if not results:
            return empty

        student_ids = {sid for _, sid, _ in results}

        pq_stmt = select(
            PracticeQuiz.student_id,
            PracticeQuiz.created_at,
            PracticeQuiz.completed_at,
            PracticeQuiz.status,
        ).where(PracticeQuiz.student_id.in_(student_ids))
        if teacher_id is not None:
            pq_stmt = pq_stmt.where(PracticeQuiz.teacher_id == teacher_id)
        pq_res = await db.execute(pq_stmt)
        practice_by_student: dict[int, list[tuple]] = {}
        for sid, created_at, completed_at, status in pq_res.all():
            practice_by_student.setdefault(sid, []).append(
                (
                    _aware(created_at),
                    _aware(completed_at) if completed_at else None,
                    status,
                )
            )

        # Every real exam per student (quiz_id kept so a student's own
        # triggering exam is never counted as its own "later" reassessment).
        exam_stmt = (
            select(StudentResult.student_id, Quiz.id, Quiz.created_at)
            .join(Quiz, StudentResult.quiz_id == Quiz.id)
            .where(Quiz.status == "saved", Quiz.source_practice_quiz_id.is_(None))
        )
        if teacher_id is not None:
            exam_stmt = exam_stmt.where(Quiz.teacher_id == teacher_id)
        exam_res = await db.execute(exam_stmt)
        exams_by_student: dict[int, list[tuple]] = {}
        for sid, qid, created_at in exam_res.all():
            exams_by_student.setdefault(sid, []).append((qid, _aware(created_at)))

    uploaded = len(results)
    weakness_identified = 0
    practice_generated = 0
    practice_completed = 0
    reassessed = 0

    for quiz_id, student_id, weaknesses in results:
        if not (weaknesses or "").strip():
            continue
        weakness_identified += 1

        quiz_time = quiz_created_at[quiz_id]
        practices_after = sorted(
            (p for p in practice_by_student.get(student_id, []) if p[0] > quiz_time),
            key=lambda p: p[0],
        )
        if not practices_after:
            continue
        practice_generated += 1

        _, completed_at, status = practices_after[0]
        if status != "completed" or completed_at is None:
            continue
        practice_completed += 1

        later_exams = [
            t
            for other_qid, t in exams_by_student.get(student_id, [])
            if other_qid != quiz_id and t > completed_at
        ]
        if later_exams:
            reassessed += 1

    return {
        "uploaded": uploaded,
        "weakness_identified": weakness_identified,
        "practice_generated": practice_generated,
        "practice_completed": practice_completed,
        "reassessed": reassessed,
    }


async def get_all_students_category_history(
    teacher_id: Optional[int] = None,
) -> dict[int, dict[str, list[dict]]]:
    """Per-student category accuracy breakdown (results+quiz join, answer
    grids, parsed-data categories), scoped to every student (or one
    teacher's) in ONE round trip instead of one per student. Exists
    specifically for get_student_improvement_summary — a school with a few
    hundred students would otherwise mean hundreds of sequential per-student
    DB round trips on an admin analytics page load; this makes it 3 total,
    regardless of N.
    """
    async with get_sessionmaker()() as db:
        stmt = (
            select(
                StudentResult.student_id,
                StudentResult.quiz_id,
                Quiz.title,
                Quiz.quiz_date,
                Quiz.created_at,
                Student.name,
            )
            .join(Quiz, StudentResult.quiz_id == Quiz.id)
            .join(Student, StudentResult.student_id == Student.id)
            .where(Quiz.teacher_id == Student.teacher_id)
            .where(Quiz.status.in_(["saved", "practice"]))
        )
        if teacher_id is not None:
            stmt = stmt.where(Student.teacher_id == teacher_id)
        stmt = stmt.order_by(
            StudentResult.student_id,
            func.coalesce(Quiz.quiz_date, cast(Quiz.created_at, Date)),
            Quiz.created_at,
        )
        res = await db.execute(stmt)
        rows = res.all()
        if not rows:
            return {}
        quiz_ids = list({r.quiz_id for r 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()}

        parsed_res = await db.execute(
            select(
                ParsedData.quiz_id, ParsedData.question_num, ParsedData.main_category
            )
            .where(ParsedData.quiz_id.in_(quiz_ids))
            .where(ParsedData.main_category != "")
        )
        categories_by_quiz: dict[int, dict[int, str]] = {}
        for qid, qnum, cat in parsed_res.all():
            categories_by_quiz.setdefault(qid, {})[qnum] = cat

    out: dict[int, dict[str, list[dict]]] = {}
    for student_id, quiz_id, title, quiz_date, created_at, student_name in rows:
        cat_map = categories_by_quiz.get(quiz_id)
        grid_json = grids.get(quiz_id)
        if not cat_map or not grid_json:
            continue
        official: list = grid_json.get("official_answers") or []
        norm = _normalize_name_simple(student_name)
        student_row = next(
            (
                gs
                for gs in grid_json.get("students") or []
                if _normalize_name_simple(gs.get("name", "")) == norm
            ),
            None,
        )
        if not student_row:
            continue
        raw_answers = student_row.get("answers") or []

        tally: dict[str, list[int]] = {}
        for qnum, category in cat_map.items():
            idx = qnum - 1
            if idx < 0 or idx >= len(official) or not official[idx]:
                continue
            cell = raw_answers[idx] if idx < len(raw_answers) else None
            effective = official[idx] if cell == "=" else cell
            counts = tally.setdefault(category, [0, 0])
            counts[1] += 1
            if effective == official[idx]:
                counts[0] += 1

        date_str = str(quiz_date) if quiz_date else created_at.date().isoformat()
        student_out = out.setdefault(student_id, {})
        for category, (correct, total) in tally.items():
            if total == 0:
                continue
            student_out.setdefault(category, []).append(
                {
                    "quiz_id": quiz_id,
                    "quiz_title": title,
                    "quiz_date": date_str,
                    "correct": correct,
                    "total": total,
                    "accuracy": round(correct / total * 100, 1),
                }
            )
    return out


async def get_student_improvement_summary(teacher_id: Optional[int] = None) -> dict:
    """For each student with enough history in a category, compare their
    first up-to-3 assessments in that category against their most recent
    up-to-3 — a student "improved" in a category if recent accuracy exceeds
    baseline. Requires >=4 data points in a category (fewer than that and
    the baseline/recent windows would just be the same quizzes, always
    showing zero change).
    """
    async with get_sessionmaker()() as db:
        stmt = select(Student.id)
        if teacher_id is not None:
            stmt = stmt.where(Student.teacher_id == teacher_id)
        res = await db.execute(stmt)
        student_ids = [sid for (sid,) in res.all()]

    all_history = await get_all_students_category_history(teacher_id)

    deltas: list[float] = []
    improved_students = 0
    eligible_students = 0
    for sid in student_ids:
        history = all_history.get(sid, {})
        student_deltas = []
        for entries in history.values():
            n = len(entries)
            if n < 4:
                continue
            baseline = entries[: min(3, n)]
            recent = entries[-min(3, n) :]
            baseline_acc = sum(e["accuracy"] for e in baseline) / len(baseline)
            recent_acc = sum(e["accuracy"] for e in recent) / len(recent)
            student_deltas.append(recent_acc - baseline_acc)
        if student_deltas:
            eligible_students += 1
            deltas.extend(student_deltas)
            if max(student_deltas) > 0:
                improved_students += 1

    # Distribution buckets (percentage-point change), for the histogram.
    bucket_edges = [-100, -20, -10, 0, 10, 20, 100]
    bucket_labels = ["<-20", "-20~-10", "-10~0", "0~10", "10~20", ">20"]
    bucket_counts = [0] * len(bucket_labels)
    for d in deltas:
        pct = d * 100
        for i in range(len(bucket_labels)):
            if bucket_edges[i] <= pct < bucket_edges[i + 1] or (
                i == len(bucket_labels) - 1 and pct >= bucket_edges[i]
            ):
                bucket_counts[i] += 1
                break

    return {
        "eligible_students": eligible_students,
        "improved_students": improved_students,
        "total_students": len(student_ids),
        "distribution": [
            {"label": label, "count": c}
            for label, c in zip(bucket_labels, bucket_counts)
        ],
    }


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 = {qid: count for qid, count in 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()

    # Keep practice_quizzes from growing unbounded — each student's dashboard
    # score history lives permanently in StudentResult regardless (see
    # _feed_practice_quiz_to_dashboard), so trimming old practice-quiz rows
    # only drops the ability to revisit that specific quiz's exact questions/
    # review via its old share link, never a student's actual score history.
    await _trim_old_practice_quizzes(student_id)
    return row


PRACTICE_QUIZ_KEEP_PER_STUDENT = 50


async def _trim_old_practice_quizzes(
    student_id: int, keep: Optional[int] = None
) -> int:
    """Delete all but the `keep` most-recent practice quizzes for a student.
    Returns the number of rows deleted.
    `keep` is looked up from the module global at call time (rather than
    bound as a default argument value) so tests can monkeypatch
    PRACTICE_QUIZ_KEEP_PER_STUDENT and have it actually take effect —
    default-argument values are bound once at function-definition time, not
    per call.
    """
    if keep is None:
        keep = PRACTICE_QUIZ_KEEP_PER_STUDENT
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(PracticeQuiz.id)
            .where(PracticeQuiz.student_id == student_id)
            # id as a secondary key: created_at alone ties whenever multiple
            # quizzes are created within the same timestamp resolution
            # window, which would otherwise make "most recent" ambiguous.
            .order_by(PracticeQuiz.created_at.desc(), PracticeQuiz.id.desc())
            .offset(keep)
        )
        old_ids = [row[0] for row in res.all()]
        if not old_ids:
            return 0
        await db.execute(delete(PracticeQuiz).where(PracticeQuiz.id.in_(old_ids)))
        await db.commit()
        return len(old_ids)


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, Student.name)
            .join(Student, PracticeQuiz.student_id == Student.id)
            .where(PracticeQuiz.id == practice_quiz_id)
            .where(PracticeQuiz.teacher_id == teacher_id)
        )
        row = res.first()
        if not row:
            return None
        quiz, student_name = row
        d = _row(quiz)
        d["student_name"] = student_name
        return d


async def delete_practice_quiz(practice_quiz_id: int, teacher_id: int) -> bool:
    """Delete a practice quiz (invalidates its share link immediately).
    Scoped to the owning teacher. Returns False if not found/not owned.
    """
    async with get_sessionmaker()() as db:
        res = await db.execute(
            delete(PracticeQuiz)
            .where(PracticeQuiz.id == practice_quiz_id)
            .where(PracticeQuiz.teacher_id == teacher_id)
        )
        await db.commit()
        return res.rowcount > 0  # type: ignore[attr-defined]


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_for_student(student_id: int) -> list[dict]:
    """A student's own assigned practice quizzes — self-scoped by student_id
    alone (no teacher_id needed, the caller already only knows their own id
    from their JWT). Powers the "assigned quizzes" banner on the student
    dashboard. Trimmed (no `questions`/`answers` payload) since the list view
    only needs enough to show what's pending and link into it.
    """
    async with get_sessionmaker()() as db:
        res = await db.execute(
            select(PracticeQuiz)
            .where(PracticeQuiz.student_id == student_id)
            .order_by(PracticeQuiz.created_at.desc(), PracticeQuiz.id.desc())
        )
        return [
            {
                "id": q.id,
                "title": q.title,
                "share_token": q.share_token,
                "status": q.status,
                "score": q.score,
                "created_at": q.created_at,
                "completed_at": q.completed_at,
            }
            for q in res.scalars().all()
        ]


async def list_practice_quizzes(
    teacher_id: int, student_id: Optional[int] = None
) -> list[dict]:
    async with get_sessionmaker()() as db:
        stmt = (
            select(PracticeQuiz, Student.name)
            .join(Student, PracticeQuiz.student_id == Student.id)
            .where(PracticeQuiz.teacher_id == teacher_id)
            .order_by(PracticeQuiz.created_at.desc(), PracticeQuiz.id.desc())
        )
        if student_id is not None:
            stmt = stmt.where(PracticeQuiz.student_id == student_id)
        res = await db.execute(stmt)
        out = []
        for quiz, student_name in res.all():
            d = _row(quiz)
            d["student_name"] = student_name
            out.append(d)
        return out


async def submit_practice_quiz(
    share_token: str,
    answers: dict[str, str],
    analyze_fn: Optional[
        Callable[[list[dict], dict[str, str]], Awaitable[tuple[str, str]]]
    ] = None,
) -> 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.
    `analyze_fn`, if given, is awaited with (questions, answers) right after
    grading and before feeding the dashboard, so the resulting StudentResult
    carries weaknesses/study_recommendations the same way a full quiz report
    does. Kept as an injected callback (rather than importing report_generator
    directly here) so this data-access module doesn't depend on the LLM layer.
    """
    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"):
        weaknesses, study_recommendations = ("", "")
        if analyze_fn is not None:
            try:
                weaknesses, study_recommendations = await analyze_fn(questions, answers)
            except Exception:
                pass  # non-fatal — the score still gets recorded below
        await _feed_practice_quiz_to_dashboard(
            result, weaknesses, study_recommendations
        )

    return result


async def _feed_practice_quiz_to_dashboard(
    pq: dict, weaknesses: str = "", study_recommendations: str = ""
) -> 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(),
            source_practice_quiz_id=pq["id"],
        )
        db.add(quiz)
        await db.flush()
        db.add(
            StudentResult(
                quiz_id=quiz.id,
                student_id=pq["student_id"],
                score=pq["score"],
                weaknesses=weaknesses or None,
                study_recommendations=study_recommendations or None,
            )
        )
        await db.commit()