"""Running class markbook / gradebook API (/api/markbook). A markbook is a term-long teacher-editable ledger: roster rows × arbitrary assessment columns. It deliberately does not depend on exam-marker batches. All user-facing reads and writes use the as-user Supabase client so class/markbook RLS gates are enforced. """ from __future__ import annotations import csv import io import os from datetime import date as Date from typing import Any, Dict, List, Optional from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import Response from pydantic import BaseModel, Field from modules.logger_tool import initialise_logger from routers.exam.dependencies import ExamContext, get_exam_context, resolve_student_names logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), "default", True) router = APIRouter() class CreateAssessmentRequest(BaseModel): title: str = Field(..., min_length=1, max_length=160) date: Optional[Date] = None max_marks: float = Field(default=100, gt=0) class MarkUpsertRequest(BaseModel): mark: Optional[float] = Field(default=None, ge=0) def _rows(result: Any) -> List[Dict[str, Any]]: data = getattr(result, "data", None) if not data: return [] return data if isinstance(data, list) else [data] def _first(result: Any) -> Optional[Dict[str, Any]]: rows = _rows(result) return rows[0] if rows else None def _round(value: Optional[float]) -> Optional[float]: return None if value is None else round(float(value), 1) def _fetch_class_or_404(ctx: ExamContext, class_id: str) -> Dict[str, Any]: row = _first(ctx.supabase.table("classes").select("id, name, institute_id").eq("id", class_id).limit(1).execute()) if not row: raise HTTPException(status_code=404, detail="Class not found") return row def _active_roster(ctx: ExamContext, class_id: str) -> List[Dict[str, Any]]: roster = _rows( ctx.supabase.table("class_students") .select("student_id, status, enrolled_at") .eq("class_id", class_id) .eq("status", "active") .execute() ) names = resolve_student_names([r["student_id"] for r in roster if r.get("student_id")]) return [ { "student_id": r["student_id"], "student_name": names.get(r["student_id"]) or r["student_id"], "status": r.get("status"), "enrolled_at": r.get("enrolled_at"), } for r in roster if r.get("student_id") ] def _assessments(ctx: ExamContext, class_id: str) -> List[Dict[str, Any]]: return _rows( ctx.supabase.table("class_assessments") .select("id, class_id, tenant_id, title, date, max_marks, created_at, updated_at") .eq("class_id", class_id) .order("date") .order("created_at") .execute() ) def _marks(ctx: ExamContext, assessment_ids: List[str]) -> List[Dict[str, Any]]: if not assessment_ids: return [] return _rows( ctx.supabase.table("assessment_marks") .select("assessment_id, student_id, mark, updated_at, updated_by") .in_("assessment_id", assessment_ids) .execute() ) def _assemble_grid(ctx: ExamContext, class_id: str) -> Dict[str, Any]: cls = _fetch_class_or_404(ctx, class_id) roster = _active_roster(ctx, class_id) assessments = _assessments(ctx, class_id) assessment_ids = [a["id"] for a in assessments] marks = _marks(ctx, assessment_ids) marks_by_student: Dict[str, Dict[str, Optional[float]]] = {r["student_id"]: {} for r in roster} for m in marks: sid = m.get("student_id") aid = m.get("assessment_id") if isinstance(sid, str) and isinstance(aid, str) and sid in marks_by_student and aid in assessment_ids: marks_by_student[sid][aid] = m.get("mark") max_total = sum(float(a.get("max_marks") or 0) for a in assessments) students = [] all_entered: List[float] = [] for idx, student in enumerate(roster, start=1): entered = [ float(v) for v in (marks_by_student.get(student["student_id"], {}).get(aid) for aid in assessment_ids) if v is not None ] total = sum(entered) if entered else None students.append( { **student, "row_number": idx, "marks": {aid: marks_by_student.get(student["student_id"], {}).get(aid) for aid in assessment_ids}, "total": total, "percentage": _round((total / max_total) * 100) if total is not None and max_total > 0 else None, } ) all_entered.extend(entered) assessment_summaries = [] for a in assessments: vals = [] for s in roster: maybe_mark = marks_by_student.get(s["student_id"], {}).get(a["id"]) if maybe_mark is not None: vals.append(float(maybe_mark)) max_marks = float(a.get("max_marks") or 0) assessment_summaries.append( { "assessment_id": a["id"], "entered_count": len(vals), "average_mark": _round(sum(vals) / len(vals)) if vals else None, "average_percentage": _round((sum(vals) / len(vals) / max_marks) * 100) if vals and max_marks > 0 else None, } ) return { "class": cls, "students": students, "assessments": assessments, "assessment_summaries": assessment_summaries, "summary": { "student_count": len(roster), "assessment_count": len(assessments), "entered_mark_count": len(all_entered), "class_average_mark": _round(sum(all_entered) / len(all_entered)) if all_entered else None, }, } @router.get("/classes/{class_id}/assessments") async def list_assessments(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]: _fetch_class_or_404(ctx, class_id) return {"assessments": _assessments(ctx, class_id)} @router.post("/classes/{class_id}/assessments") async def create_assessment(class_id: str, body: CreateAssessmentRequest, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]: cls = _fetch_class_or_404(ctx, class_id) row = { "class_id": class_id, "tenant_id": cls["institute_id"], "title": body.title.strip(), "date": body.date.isoformat() if body.date else None, "max_marks": body.max_marks, } created = _first(ctx.supabase.table("class_assessments").insert(row).execute()) if not created: raise HTTPException(status_code=500, detail="Failed to create assessment") logger.info(f"Markbook assessment {created.get('id')} created for class {class_id} by {ctx.user_id}") return created @router.get("/classes/{class_id}/grid") async def get_grid(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Dict[str, Any]: return _assemble_grid(ctx, class_id) @router.put("/classes/{class_id}/assessments/{assessment_id}/marks/{student_id}") async def upsert_mark( class_id: str, assessment_id: str, student_id: str, body: MarkUpsertRequest, ctx: ExamContext = Depends(get_exam_context), ) -> Dict[str, Any]: cls = _fetch_class_or_404(ctx, class_id) assessment = _first( ctx.supabase.table("class_assessments") .select("id, class_id, tenant_id, max_marks") .eq("id", assessment_id) .eq("class_id", class_id) .limit(1) .execute() ) if not assessment: raise HTTPException(status_code=404, detail="Assessment not found") roster_row = _first( ctx.supabase.table("class_students") .select("student_id") .eq("class_id", class_id) .eq("student_id", student_id) .eq("status", "active") .limit(1) .execute() ) if not roster_row: raise HTTPException(status_code=404, detail="Student is not active in this class") if body.mark is not None and body.mark > float(assessment.get("max_marks") or 0): raise HTTPException(status_code=422, detail="mark exceeds assessment max_marks") row = { "assessment_id": assessment_id, "student_id": student_id, "tenant_id": assessment.get("tenant_id") or cls["institute_id"], "mark": body.mark, "updated_by": ctx.user_id, } upserted = _first(ctx.supabase.table("assessment_marks").upsert(row, on_conflict="assessment_id,student_id").execute()) if not upserted: raise HTTPException(status_code=500, detail="Failed to upsert mark") return {"status": "ok", "mark": upserted} @router.get("/classes/{class_id}/csv") async def export_csv(class_id: str, ctx: ExamContext = Depends(get_exam_context)) -> Response: data = _assemble_grid(ctx, class_id) assessments = data["assessments"] buf = io.StringIO() writer = csv.writer(buf) writer.writerow(["row", "student_name", "student_id"] + [a["title"] for a in assessments] + ["total", "percentage"]) for student in data["students"]: writer.writerow( [student["row_number"], student.get("student_name") or "", student.get("student_id") or ""] + ["" if student["marks"].get(a["id"]) is None else student["marks"].get(a["id"]) for a in assessments] + ["" if student["total"] is None else student["total"], "" if student["percentage"] is None else student["percentage"]] ) filename = f"markbook-{class_id}.csv" return Response(content=buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{filename}"'})