| """Durable answer scans. RPC transactions own state; the server owns image access. |
| |
| No fallback to process memory is allowed. The adapter uses the same Supabase |
| project as sign-in; a verified user ID is passed on every student operation. |
| |
| Images: with MATHLY_ANSWER_BUCKET set (a private Hugging Face storage bucket, written with HF_TOKEN), new answer |
| images are kept there instead of Supabase Storage, so reading them back (to grade and to show them) is not Supabase |
| egress. Their file names start with "hf-"; images saved before keep their names and stay in Supabase, where they |
| are still read from. |
| """ |
| from __future__ import annotations |
|
|
| import io |
| import json |
| import logging |
| import os |
| import tempfile |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
| import uuid |
| import warnings |
| from dataclasses import dataclass |
| from datetime import datetime, timedelta, timezone |
| from pathlib import Path |
|
|
| from PIL import Image, ImageOps, UnidentifiedImageError |
|
|
| MAX_IMAGE = 5_000_000 |
| Image.MAX_IMAGE_PIXELS = 20_000_000 |
| BUCKET = "mathly-answers" |
| HUB_MARK = "hf-" |
| ORPHAN_AGE = timedelta(hours=24) |
| log = logging.getLogger("mathly.answers") |
|
|
|
|
| @dataclass |
| class ScanError(Exception): |
| status: int |
| code: str |
| reason: str = "" |
| detail: str = "" |
|
|
|
|
| def clean_image(data: bytes, content_type: str) -> bytes: |
| if not data: |
| raise ScanError(400, "empty_image") |
| if len(data) > MAX_IMAGE: |
| raise ScanError(413, "image_too_large") |
| if content_type not in ("image/jpeg", "image/png", "image/webp"): |
| raise ScanError(415, "unsupported_image") |
| try: |
| with warnings.catch_warnings(): |
| warnings.simplefilter("error", Image.DecompressionBombWarning) |
| with Image.open(io.BytesIO(data)) as source: |
| if source.format not in ("JPEG", "PNG", "WEBP") or source.width * source.height > 20_000_000: |
| raise ValueError("image dimensions or format") |
| image = ImageOps.exif_transpose(source).convert("RGB") |
| image.thumbnail((2400, 2400)) |
| output = io.BytesIO() |
| image.save(output, "JPEG", quality=88) |
| result = output.getvalue() |
| if len(result) > MAX_IMAGE: |
| raise ScanError(413, "image_too_large") |
| return result |
| except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError, |
| Image.DecompressionBombWarning) as exc: |
| raise ScanError(415, "invalid_image") from exc |
|
|
|
|
| class HubImages: |
| """Answer images in a private Hugging Face storage bucket, under answers/<the object key>.""" |
| PREFIX = "answers/" |
|
|
| def __init__(self, bucket: str, token: str | None = None): |
| from huggingface_hub import HfApi |
| from huggingface_hub.utils import disable_progress_bars |
| disable_progress_bars() |
| self.bucket, self.api = bucket, HfApi(token=token or None) |
|
|
| @classmethod |
| def from_env(cls): |
| bucket = os.environ.get("MATHLY_ANSWER_BUCKET", "").strip() |
| return cls(bucket, os.environ.get("HF_TOKEN", "").strip()) if bucket else None |
|
|
| def _call(self, what, *args, **kwargs): |
| from huggingface_hub.errors import EntryNotFoundError |
| try: |
| return what(self.bucket, *args, **kwargs) |
| except EntryNotFoundError as exc: |
| raise ScanError(404, "not_found") from exc |
| except Exception as exc: |
| log.warning("answer bucket failed: %s", type(exc).__name__) |
| raise ScanError(503, "answer_storage_unavailable") from exc |
|
|
| def put(self, object_key: str, data: bytes): |
| self._call(self.api.batch_bucket_files, add=[(data, self.PREFIX + object_key)]) |
|
|
| def read(self, object_key: str) -> bytes: |
| with tempfile.TemporaryDirectory() as tmp: |
| target = Path(tmp) / "image.jpg" |
| self._call(self.api.download_bucket_files, [(self.PREFIX + object_key, str(target))], raise_on_missing_files=True) |
| return target.read_bytes() |
|
|
| def delete(self, object_key: str): |
| self._call(self.api.batch_bucket_files, delete=[self.PREFIX + object_key]) |
|
|
| def older_than(self, age: timedelta) -> list[str]: |
| """The object keys uploaded before now - age.""" |
| cutoff = datetime.now(timezone.utc) - age |
| files = self._call(self.api.list_bucket_tree, prefix=self.PREFIX.rstrip("/"), recursive=True) |
| return [f.path[len(self.PREFIX):] for f in files |
| if getattr(f, "type", "") == "file" and f.path.startswith(self.PREFIX) |
| and (getattr(f, "uploaded_at", None) or getattr(f, "mtime", None) or cutoff) < cutoff] |
|
|
|
|
| def on_hub(object_key: str) -> bool: |
| return object_key.rsplit("/", 1)[-1].startswith(HUB_MARK) |
|
|
|
|
| class AnswerScans: |
| def __init__(self, auth, *, enabled=None, images="env"): |
| self.auth = auth |
| self.enabled = (os.environ.get("MATHLY_ANSWER_SCANS_ENABLED") == "1") if enabled is None else enabled |
| self.images = HubImages.from_env() if images == "env" else images |
|
|
| def new_key(self, user_id: str, scan_id: str, lease_id: str) -> str: |
| """A new image's object key; the database checks its user/scan/lease start.""" |
| return f"{user_id}/{scan_id}/{lease_id}/{HUB_MARK if self.images else ''}{uuid.uuid4()}.jpg" |
|
|
| @property |
| def configured(self): |
| return bool(self.enabled and self.auth and self.auth.secret) |
|
|
| def action(self, user_id: str | None, action: str, **data): |
| if not self.configured: |
| raise ScanError(503, "answer_storage_unavailable") |
| body = {"p_user": user_id, "p_action": action, "p_data": data} |
| raw = self._request("/rest/v1/rpc/math_answer_action", "POST", json.dumps(body).encode(), "application/json") |
| return json.loads(raw) |
|
|
| def grade(self, user_id: str, action: str, **data): |
| if not self.configured: |
| raise ScanError(503, "answer_storage_unavailable") |
| body = {"p_user": user_id, "p_action": action, "p_data": data} |
| raw = self._request("/rest/v1/rpc/math_grade_action", "POST", json.dumps(body).encode(), "application/json") |
| return json.loads(raw) |
|
|
| def quiz(self, user_id: str | None, action: str, **data): |
| """Unit-quiz sittings (app/quiz_grades.py): submit, list a student's own, or `report` for a set of accounts.""" |
| if not self.configured: |
| raise ScanError(503, "answer_storage_unavailable") |
| body = {"p_user": user_id, "p_action": action, "p_data": data} |
| try: |
| raw = self._request("/rest/v1/rpc/math_quiz_action", "POST", json.dumps(body).encode(), "application/json") |
| except ScanError as exc: |
| if exc.code == "answer_storage_unavailable": |
| raise ScanError(503, "quiz_sittings_unavailable") from exc |
| raise |
| return json.loads(raw) |
|
|
| def review(self, user_id: str | None, action: str, **data): |
| """The teacher's reviews of answers and the students' requests to look again (app/teacher_review.py). The |
| student's id is `user_id` (a teacher's route finds it first with the `owner` action).""" |
| if not self.configured: |
| raise ScanError(503, "answer_storage_unavailable") |
| body = {"p_user": user_id, "p_action": action, "p_data": data} |
| try: |
| raw = self._request("/rest/v1/rpc/math_review_action", "POST", json.dumps(body).encode(), "application/json") |
| except ScanError as exc: |
| if exc.code == "answer_storage_unavailable": |
| raise ScanError(503, "review_unavailable") from exc |
| raise |
| return json.loads(raw) |
|
|
| def _request(self, path, method="GET", data=None, content_type=None): |
| headers = {"Authorization": "Bearer " + self.auth.secret, "apikey": self.auth.secret, |
| "User-Agent": "mathly/1.0"} |
| if content_type: |
| headers["Content-Type"] = content_type |
| req = urllib.request.Request(self.auth.url + path, data=data, method=method, headers=headers) |
| try: |
| with urllib.request.urlopen(req, timeout=30) as response: |
| return response.read() |
| except urllib.error.HTTPError as exc: |
| |
| try: |
| error = json.loads(exc.read()) |
| except (ValueError, OSError): |
| error = {} |
| code, message = error.get("code"), error.get("message", "") |
| if code == "P0002": |
| raise ScanError(404, "not_found") from exc |
| if code == "P0001" and message in {"grading_limit", "review_limit"}: |
| raise ScanError(429, message) from exc |
| if code == "P0001" and message in {"superseded", "invalid_upload", "upload_in_progress", "confirmation_required", |
| "answer_required", "grading_busy", "grading_interrupted", "invalid_sitting"}: |
| raise ScanError(409, message) from exc |
| raise ScanError(503, "answer_storage_unavailable") from exc |
| except (urllib.error.URLError, TimeoutError, OSError) as exc: |
| raise ScanError(503, "answer_storage_unavailable") from exc |
|
|
| def _hub(self, object_key): |
| if not on_hub(object_key): |
| return None |
| if not self.images: |
| raise ScanError(503, "answer_storage_unavailable") |
| return self.images |
|
|
| def put(self, object_key, data): |
| if hub := self._hub(object_key): |
| return hub.put(object_key, data) |
| self._request(f"/storage/v1/object/{BUCKET}/{urllib.parse.quote(object_key, safe='/')}", |
| "POST", data, "image/jpeg") |
|
|
| def read(self, object_key): |
| if hub := self._hub(object_key): |
| return hub.read(object_key) |
| return self._request(f"/storage/v1/object/authenticated/{BUCKET}/{urllib.parse.quote(object_key, safe='/')}") |
|
|
| def delete(self, object_key): |
| if hub := self._hub(object_key): |
| return hub.delete(object_key) |
| self._request(f"/storage/v1/object/{BUCKET}", "DELETE", |
| json.dumps({"prefixes": [object_key]}).encode(), "application/json") |
|
|
| def referenced(self, keys: list[str]) -> set[str]: |
| """Which of these object keys an answer still uses.""" |
| found = set() |
| for i in range(0, len(keys), 40): |
| chunk = keys[i:i + 40] |
| names = ",".join('"' + k.replace('"', "") + '"' for k in chunk) |
| raw = self._request("/rest/v1/math_answer_attachments?select=object_key&object_key=in." |
| + urllib.parse.quote(f"({names})", safe="(),")) |
| found |= {row["object_key"] for row in json.loads(raw)} |
| return found |
|
|
| def cleanup(self): |
| """Removes images no answer uses, a day after they were uploaded (an upload's answer is saved right after it).""" |
| candidates = self.action(None, "cleanup_candidates") |
| for key in candidates: |
| self.delete(key) |
| removed = len(candidates) |
| if self.images: |
| old = self.images.older_than(ORPHAN_AGE) |
| used = self.referenced(old) if old else set() |
| for key in old: |
| if key not in used: |
| self.images.delete(key) |
| removed += 1 |
| return removed |
|
|
|
|
| def public_result(result): |
| """Keep storage paths and upload coordination secrets out of public responses.""" |
| if isinstance(result, list): |
| return [public_result(item) for item in result] |
| result = {k: dict(v) if isinstance(v, dict) else v for k, v in result.items()} |
| if result.get("scan"): |
| for key in ("idempotency_key", "lease_id", "lease_until"): |
| result["scan"].pop(key, None) |
| result["scan"]["code"] = scan_code(result["scan"].get("id")) |
| if result.get("attachment"): |
| result["attachment"].pop("object_key", None) |
| if result.get("grading"): |
| result["grading"].pop("lease_id", None) |
| result["grading"].pop("lease_until", None) |
| return result |
|
|
|
|
| def scan_code(scan_id): |
| """The 4 digits shown on both the computer and the phone, so the student can match the request.""" |
| try: |
| return f"{uuid.UUID(str(scan_id)).int % 10000:04d}" |
| except (ValueError, TypeError, AttributeError): |
| return "" |
|
|
|
|
| def uuid_text(value): |
| try: |
| return str(uuid.UUID(str(value))) |
| except (ValueError, TypeError, AttributeError) as exc: |
| raise ScanError(404, "not_found") from exc |
|
|
