| """The work on the board (ืืืจื): every answer keeps a picture of the student's board for its question, so a teacher can grade the way and not |
| only the final answer. |
| |
| The browser sends the picture when the student answers (practice, a lesson round) or chooses (a quiz, an exercise test, the simulation), and |
| again before handing in when the board changed. One picture is kept per student, attempt (or "practice") and question; the one it replaces is |
| left with nothing referring to it, and the hourly cleanup removes it. In a graded attempt (a lesson quiz, an exercise test, the simulation) the |
| teacher of the student's class sees each question with the student's choice and board, and can give it 0, half or full points for the work; |
| the attempt's grade is recomputed.""" |
| from __future__ import annotations |
|
|
| from datetime import timedelta |
|
|
| from fastapi import APIRouter, Depends, HTTPException, Query, Request |
| from fastapi.concurrency import run_in_threadpool |
| from pydantic import BaseModel |
|
|
| from . import attempts, clock, pictures |
| from .access import allowed, learner, need |
| from .written import picture_error |
|
|
| router = APIRouter(prefix="/api") |
|
|
| PRACTICE = "practice" |
| GRADED = ("quiz", "exercise_test", "simulation") |
| LATE_UPLOAD = timedelta(minutes=10) |
| WORK_REQUIRED = GRADED |
|
|
|
|
| def _skill_of_key(bank, key: str) -> str | None: |
| """The skill a bank question key belongs to (a key starts with its skill's id).""" |
| for skill in bank.skills: |
| if key.startswith(skill["id"]) and key in bank.question_keys(skill["id"]): |
| return skill["id"] |
| return None |
|
|
|
|
| @router.post("/work", status_code=201) |
| async def upload_work(request: Request, ref: str = Query(..., max_length=64), key: str = Query(..., max_length=200), |
| user: dict = Depends(learner)) -> dict: |
| """The student's board for one answer, as a picture (raw image bytes). `ref`: the attempt's id, or "practice".""" |
| store, bank = request.app.state.store, request.app.state.bank |
| if ref == PRACTICE: |
| skill_id = _skill_of_key(bank, key) |
| else: |
| row = store.get_attempt(ref) |
| if row is None or row["user_id"] != user["id"]: |
| raise HTTPException(404, "ืืืืื ืื ื ืืฆื") |
| if not attempts.is_open(row) and clock.now() - row["submitted_at"] > LATE_UPLOAD: |
| raise HTTPException(409, "ืืืืื ืืืจ ืืืืฉ") |
| skill_id = next((q["skill_id"] for q in row["questions"] if q["key"] == key), None) |
| if skill_id is None: |
| raise HTTPException(404, "ืืฉืืื ืื ื ืืฆืื") |
| declared = request.headers.get("content-length") |
| if declared and declared.isdigit() and int(declared) > pictures.MAX_IMAGE + 1024: |
| raise picture_error(pictures.PictureError(413, "image_too_large")) |
| data, kind = await request.body(), request.headers.get("content-type", "") |
|
|
| def work() -> dict: |
| jpeg = pictures.clean_image(data, kind) |
| saved = pictures.store_picture(store, request.app.state.pictures, user["id"], "work", jpeg) |
| store.save_work(user["id"], ref, key, skill_id, saved["id"], clock.now()) |
| return saved |
| try: |
| row = await run_in_threadpool(work) |
| except pictures.PictureError as exc: |
| raise picture_error(exc) from exc |
| return {"id": row["id"]} |
|
|
|
|
| |
|
|
| def _taught(request: Request, user: dict) -> list[dict]: |
| store = request.app.state.store |
| return store.classes() if allowed(user["roles"], "*") else store.classes_taught(user["email"]) |
|
|
|
|
| def teaches_student(request: Request, user: dict, student_id: str) -> bool: |
| """Is the caller a teacher of a class this student is in (or a developer)?""" |
| if allowed(user["roles"], "*"): |
| return True |
| store = request.app.state.store |
| return any(m["user_id"] == student_id for c in store.classes_taught(user["email"]) for m in store.members(c["id"])) |
|
|
|
|
| def _title(request: Request, row: dict) -> str: |
| bank, store = request.app.state.bank, request.app.state.store |
| if row["kind"] == "quiz": |
| return "ืืืื: " + next((l["name"] for l in bank.lessons if l["id"] == row["scope"]), row["scope"]) |
| test_id, _, number = row["scope"].partition(":") |
| test = store.get_test(int(test_id)) if test_id.isdigit() else None |
| when = f" ({test['date'].day}.{test['date'].month})" if test else "" |
| from .exam import TITLES, mode_of_scope |
| name = TITLES[mode_of_scope(row["scope"])] if row["kind"] == "exercise_test" else "ืืืื ืกืืืืืฆืื" |
| return name + when |
|
|
|
|
| def _iso(value): |
| return value.isoformat() if value else None |
|
|
|
|
| @router.get("/teaching/work") |
| def work_list(request: Request, user: dict = Depends(need("class.teach"))) -> list[dict]: |
| """The handed-in quizzes and tests of the teacher's students, the ones not checked yet first, then the newest.""" |
| store, items, seen = request.app.state.store, [], set() |
| for klass in _taught(request, user): |
| emails = {m["user_id"]: m["email"] for m in store.members(klass["id"]) if m["user_id"]} |
| for kind in GRADED: |
| for uid, rows in store.attempts_of_users(list(emails), kind).items(): |
| for row in rows: |
| if attempts.is_open(row) or row["id"] in seen: |
| continue |
| seen.add(row["id"]) |
| items.append({"id": row["id"], "kind": kind, "title": _title(request, row), "student": emails[uid], |
| "class": {"id": klass["id"], "name": klass["name"]}, "grade": row["grade"], |
| "submitted_at": _iso(row["submitted_at"]), "checked": bool(row.get("checked_at"))}) |
| items.sort(key=lambda i: i["submitted_at"] or "", reverse=True) |
| return sorted(items, key=lambda i: i["checked"]) |
|
|
|
|
| def _graded_attempt(request: Request, user: dict, attempt_id: str) -> dict: |
| row = request.app.state.store.get_attempt(attempt_id) |
| if row is None or row["kind"] not in GRADED or attempts.is_open(row) or not teaches_student(request, user, row["user_id"]): |
| raise HTTPException(404, "ืืืืื ืื ื ืืฆื") |
| return row |
|
|
|
|
| def teacher_view(request: Request, row: dict) -> dict: |
| """Every question with its choices (the right one marked: this is the teacher), the student's choice, the automatic points, the |
| teacher's credit and the board's picture.""" |
| bank, store = request.app.state.bank, request.app.state.store |
| work = store.work_of(row["user_id"], row["id"]) |
| questions = [] |
| for q in row["questions"]: |
| served = bank.practice_question(q["skill_id"], q["key"]) |
| if served is None: |
| continue |
| choices = [{**c, "correct": bool((bank.grade(q["skill_id"], q["key"], c["index"]) or {}).get("correct"))} for c in served["choices"]] |
| picked = row["answers"].get(q["key"]) |
| questions.append({"key": q["key"], "skill_name": (bank.skill(q["skill_id"]) or {}).get("name", ""), |
| "question": served["question"], **{k: served[k] for k in ("figure", "graph") if k in served}, |
| "choices": choices, "picked": picked if isinstance(picked, int) else None, |
| "auto": 1 if any(c["correct"] and c["index"] == picked for c in choices) else 0, |
| "credit": row["credit"].get(q["key"]), "work": work.get(q["key"]), |
| "ai": ((row.get("review") or {}).get("questions") or {}).get(q["key"])}) |
| student = next((m["email"] for c in store.classes() for m in store.members(c["id"]) if m["user_id"] == row["user_id"]), "") |
| return {"id": row["id"], "kind": row["kind"], "title": _title(request, row), "student": student, "grade": row["grade"], |
| "submitted_at": _iso(row["submitted_at"]), "checked_at": _iso(row.get("checked_at")), "checked_by": row.get("checked_by"), |
| "questions": questions, "ai_summary": (row.get("review") or {}).get("summary"), "ai": _ai_state(request, row)} |
|
|
|
|
| def _ai_state(request: Request, row: dict) -> dict: |
| """The AI review as the teacher sees it: can it run here (a key is set), its state, and why it failed.""" |
| review = row.get("review") or {} |
| return {"available": getattr(request.app.state, "reviewer", None) is not None, "state": review.get("state"), |
| "reason": review.get("reason"), "summary": review.get("summary"), "model": review.get("model"), |
| "at": review.get("at"), "events": review.get("events", []), "missing": review.get("missing", 0)} |
|
|
|
|
| @router.get("/teaching/work/{attempt_id}") |
| def work_open(attempt_id: str, request: Request, user: dict = Depends(need("class.teach"))) -> dict: |
| from . import boardreview |
| return teacher_view(request, boardreview.ensure(request.app, _graded_attempt(request, user, attempt_id))) |
|
|
|
|
| @router.post("/teaching/work/{attempt_id}/review") |
| def work_review(attempt_id: str, request: Request, key: str | None = Query(None, max_length=200), |
| user: dict = Depends(need("class.teach"))) -> dict: |
| """The teacher asks for the AI review (again): of the whole attempt, or of one question (`key`).""" |
| from . import boardreview |
| if getattr(request.app.state, "reviewer", None) is None: |
| raise HTTPException(409, "ืืืืงื ืขื AI ืื ืืืืืจืช ืืฉืจืช: ืืกืจ GEMINI_API_KEY") |
| row = _graded_attempt(request, user, attempt_id) |
| if key is not None and key not in {q["key"] for q in row["questions"]}: |
| raise HTTPException(404, "ืืฉืืื ืื ื ืืฆืื") |
| if key is not None and key not in request.app.state.store.work_of(row["user_id"], row["id"]): |
| raise HTTPException(409, "ืืื ืืื ืืฉืืื ืืื") |
| return teacher_view(request, boardreview.ensure(request.app, row, force=True, keys=[key] if key else None)) |
|
|
|
|
| class Credit(BaseModel): |
| credit: dict[str, float | None] |
|
|
|
|
| @router.post("/teaching/work/{attempt_id}") |
| def work_check(attempt_id: str, body: Credit, request: Request, user: dict = Depends(need("class.teach"))) -> dict: |
| """Points for the way: per question 0, 0.5 or 1 (None: back to the automatic points). The grade is recomputed from all of them.""" |
| store, bank = request.app.state.store, request.app.state.bank |
| row = _graded_attempt(request, user, attempt_id) |
| keys = {q["key"] for q in row["questions"]} |
| credit = dict(row["credit"]) |
| for key, value in body.credit.items(): |
| if key not in keys or (value is not None and value not in attempts.CREDITS): |
| raise HTTPException(422, "ื ืืงืื ืื ืชืงืื") |
| if value is None: |
| credit.pop(key, None) |
| else: |
| credit[key] = value |
| value, _ = attempts.grade(bank, row["questions"], row["answers"], credit) |
| saved = store.save_attempt(row["id"], credit=credit, grade=value, checked_at=clock.now(), checked_by=user["email"]) |
| return teacher_view(request, saved) |
|
|
