244 lines
8.3 KiB
Python
244 lines
8.3 KiB
Python
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel, EmailStr
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.models.attempt import AttemptAnswer, QuizAttempt
|
|
from app.models.email_verification import EmailVerification
|
|
from app.models.quiz import Quiz
|
|
from app.models.user import User
|
|
from app.schemas.auth import Token
|
|
from app.utils.auth import create_access_token, get_current_user, verify_password
|
|
from app.utils.quiz_questions import get_quiz_questions
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
class MobileLoginRequest(BaseModel):
|
|
email: EmailStr
|
|
password: str
|
|
|
|
|
|
class MobileAnswerSubmit(BaseModel):
|
|
question_id: int
|
|
user_answer: str
|
|
transcript: str | None = None
|
|
selected_letter: str | None = None
|
|
confidence: float | None = None
|
|
|
|
|
|
class MobileAttemptSubmit(BaseModel):
|
|
quiz_id: int
|
|
started_at: datetime | None = None
|
|
completed_at: datetime | None = None
|
|
selected_question_ids: list[int] | None = None
|
|
answers: list[MobileAnswerSubmit]
|
|
|
|
|
|
ANDROID_MODEL_MANIFEST = {
|
|
"version": 1,
|
|
"default_stack": {
|
|
"stt": "android-speech-offline",
|
|
"mapper": "deterministic-option-matcher",
|
|
"tts": "android-system-tts",
|
|
},
|
|
"models": [
|
|
{
|
|
"id": "android-speech-offline",
|
|
"role": "stt",
|
|
"runtime": "android.speech.SpeechRecognizer",
|
|
"label": "Android offline speech recognizer",
|
|
"download_url": None,
|
|
"required": False,
|
|
"notes": "Uses the device speech service with offline preference while bundled STT artifacts are added.",
|
|
},
|
|
{
|
|
"id": "deterministic-option-matcher",
|
|
"role": "mapper",
|
|
"runtime": "in-app",
|
|
"label": "Exact option/letter matcher",
|
|
"download_url": None,
|
|
"required": True,
|
|
"notes": "Grades locally against synced answer keys; no cloud model decides correctness.",
|
|
},
|
|
{
|
|
"id": "android-system-tts",
|
|
"role": "tts",
|
|
"runtime": "android.speech.tts.TextToSpeech",
|
|
"label": "Android system TTS",
|
|
"download_url": None,
|
|
"required": False,
|
|
"notes": "Fast local read-aloud fallback before packaged neural TTS is shipped.",
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def _mobile_login_rate_limit(client_ip: str):
|
|
try:
|
|
import redis as redis_lib
|
|
|
|
r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True, socket_connect_timeout=1)
|
|
key = f"mobile_login_attempts:{client_ip}"
|
|
count = r.incr(key)
|
|
if count == 1:
|
|
r.expire(key, 15 * 60)
|
|
if count > 20:
|
|
raise HTTPException(status_code=429, detail="Too many login attempts. Try again in 15 minutes.")
|
|
except HTTPException:
|
|
raise
|
|
except Exception:
|
|
# Mobile login should still work if Redis is briefly unavailable.
|
|
return
|
|
|
|
|
|
def _visible_quizzes_query(db: Session, current_user: User):
|
|
query = db.query(Quiz).filter(Quiz.deleted_at.is_(None))
|
|
if not current_user.is_moderator:
|
|
query = query.filter(Quiz.is_published == 1, Quiz.course_id.is_(None))
|
|
return query
|
|
|
|
|
|
def _question_payload(question):
|
|
return {
|
|
"id": question.id,
|
|
"question_text": question.question_text,
|
|
"question_type": question.question_type,
|
|
"options": question.options,
|
|
"correct_answer": question.correct_answer,
|
|
"explanation": question.explanation,
|
|
"page_reference": question.page_reference,
|
|
"image_path": question.image_path,
|
|
}
|
|
|
|
|
|
def _quiz_payload(db: Session, quiz: Quiz, include_questions: bool = True):
|
|
payload = {
|
|
"id": quiz.id,
|
|
"title": quiz.title,
|
|
"mode": quiz.mode,
|
|
"questions_count": quiz.questions_count,
|
|
"time_limit_minutes": quiz.time_limit_minutes,
|
|
"questions_per_attempt": quiz.questions_per_attempt,
|
|
"category_id": quiz.category_id,
|
|
"course_id": quiz.course_id,
|
|
"is_published": quiz.is_published,
|
|
"is_shared": quiz.is_shared,
|
|
"created_at": quiz.created_at.isoformat() if quiz.created_at else None,
|
|
}
|
|
if include_questions:
|
|
payload["questions"] = [_question_payload(question) for question in get_quiz_questions(db, quiz.id)]
|
|
return payload
|
|
|
|
|
|
@router.post("/auth/login")
|
|
def mobile_login(data: MobileLoginRequest, request: Request, db: Session = Depends(get_db)):
|
|
"""Password login for native apps. Uses rate limiting instead of browser Turnstile."""
|
|
client_ip = request.client.host if request and request.client else "unknown"
|
|
_mobile_login_rate_limit(client_ip)
|
|
|
|
email = data.email.lower().strip()
|
|
user = db.query(User).filter(User.email == email).first()
|
|
if not user or not verify_password(data.password, user.hashed_password):
|
|
raise HTTPException(status_code=401, detail="Invalid email or password")
|
|
|
|
verification = db.query(EmailVerification).filter(EmailVerification.user_id == user.id).first()
|
|
if verification and verification.verified_at is None:
|
|
raise HTTPException(status_code=403, detail="Email not verified")
|
|
|
|
token = create_access_token(data={"sub": user.email})
|
|
return {
|
|
**Token(access_token=token).model_dump(),
|
|
"user": {"id": user.id, "email": user.email, "name": user.name, "role": user.role},
|
|
"server_time": datetime.utcnow().isoformat(),
|
|
}
|
|
|
|
|
|
@router.get("/sync")
|
|
def mobile_sync(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Bulk sync all visible quizzes with answer keys for offline study."""
|
|
quizzes = _visible_quizzes_query(db, current_user).order_by(Quiz.created_at.desc()).all()
|
|
return {
|
|
"server_time": datetime.utcnow().isoformat(),
|
|
"user": {"id": current_user.id, "email": current_user.email, "name": current_user.name, "role": current_user.role},
|
|
"quizzes": [_quiz_payload(db, quiz) for quiz in quizzes],
|
|
"deleted_quiz_ids": [],
|
|
}
|
|
|
|
|
|
@router.get("/quizzes/{quiz_id}")
|
|
def mobile_quiz_detail(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
quiz = _visible_quizzes_query(db, current_user).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
return _quiz_payload(db, quiz)
|
|
|
|
|
|
@router.get("/models")
|
|
def mobile_models(current_user: User = Depends(get_current_user)):
|
|
_ = current_user
|
|
return ANDROID_MODEL_MANIFEST
|
|
|
|
|
|
@router.post("/attempts")
|
|
def upload_mobile_attempt(
|
|
data: MobileAttemptSubmit,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
quiz = _visible_quizzes_query(db, current_user).filter(Quiz.id == data.quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
question_map = {question.id: question for question in get_quiz_questions(db, quiz.id)}
|
|
selected_ids = data.selected_question_ids or [answer.question_id for answer in data.answers]
|
|
total_questions = len(selected_ids) if selected_ids else len(question_map)
|
|
score = 0
|
|
|
|
attempt = QuizAttempt(
|
|
quiz_id=quiz.id,
|
|
user_id=current_user.id,
|
|
total_questions=total_questions,
|
|
started_at=data.started_at or datetime.utcnow(),
|
|
completed_at=data.completed_at or datetime.utcnow(),
|
|
selected_question_ids=selected_ids or None,
|
|
)
|
|
db.add(attempt)
|
|
db.flush()
|
|
|
|
for answer in data.answers:
|
|
question = question_map.get(answer.question_id)
|
|
if not question:
|
|
continue
|
|
is_correct = bool(answer.user_answer) and answer.user_answer.strip().lower() == question.correct_answer.strip().lower()
|
|
if is_correct:
|
|
score += 1
|
|
db.add(AttemptAnswer(
|
|
attempt_id=attempt.id,
|
|
question_id=question.id,
|
|
user_answer=answer.user_answer,
|
|
is_correct=is_correct,
|
|
))
|
|
|
|
attempt.score = score
|
|
db.commit()
|
|
percentage = round((score / total_questions * 100) if total_questions else 0, 1)
|
|
return {
|
|
"id": attempt.id,
|
|
"quiz_id": quiz.id,
|
|
"score": score,
|
|
"total_questions": total_questions,
|
|
"percentage": percentage,
|
|
"completed_at": attempt.completed_at.isoformat() if attempt.completed_at else None,
|
|
}
|