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, }