from fastapi import FastAPI, Request, HTTPException from fastapi.responses import HTMLResponse, FileResponse from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from typing import List, Optional import os import json import uuid import logging from pydantic import BaseModel, validator app = FastAPI() # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Set up static files app.mount("/static", StaticFiles(directory="static"), name="static") app.mount("/assets", StaticFiles(directory="templates/assets"), name="assets") # Constants QUESTIONS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "quiz_questions.json") SCOREBOARD_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "quiz_scoreboard.json") # Pydantic Models for Quiz Questions class QuizQuestionBase(BaseModel): question: str options: List[str] answer: str @validator('options') def check_options_count(cls, v): if not 2 <= len(v) <= 8: raise ValueError('Number of options must be between 2 and 8') return v @validator('answer') def check_answer_in_options(cls, v, values): if 'options' in values and v not in values['options']: raise ValueError('Answer must be one of the provided options') return v class QuizQuestionCreate(QuizQuestionBase): pass class QuizQuestionInDB(QuizQuestionBase): id: str class QuizQuestionUpdate(BaseModel): question: Optional[str] = None options: Optional[List[str]] = None answer: Optional[str] = None @validator('options') def check_options_count_optional(cls, v): if v is not None and not 2 <= len(v) <= 8: raise ValueError('Number of options must be between 2 and 8') return v @validator('answer') def check_answer_in_options_optional(cls, v, values): if v is not None and 'options' in values and values['options'] is not None and v not in values['options']: raise ValueError('Answer must be one of the provided options if options are also being updated') return v # Scoreboard Models class ScoreboardEntry(BaseModel): initials: str time: int @validator('initials') def validate_initials(cls, v): v = v.strip().upper() if len(v) != 3: raise ValueError('Initials must be exactly 3 characters') # Allow both letters and numbers if not all(c.isalnum() for c in v): raise ValueError('Initials must contain only letters and numbers') return v @validator('time') def validate_time(cls, v): if v < 0: raise ValueError('Time cannot be negative') return v class Config: json_schema_extra = { "example": { "initials": "A1B", "time": 120 } } # Helper functions for quiz questions JSON I/O def _load_quiz_questions() -> List[QuizQuestionInDB]: if not os.path.exists(QUESTIONS_FILE): return [] try: with open(QUESTIONS_FILE, "r") as f: data = json.load(f) return [QuizQuestionInDB(**q) for q in data] except (json.JSONDecodeError, TypeError): # Handle empty or malformed file return [] def _save_quiz_questions(questions: List[QuizQuestionInDB]): with open(QUESTIONS_FILE, "w") as f: json.dump([q.dict() for q in questions], f, indent=4) def _generate_question_id() -> str: return str(uuid.uuid4()) # Helper functions for scoreboard JSON I/O def _load_scoreboard() -> List[ScoreboardEntry]: if not os.path.exists(SCOREBOARD_FILE): logging.info(f"Scoreboard file {SCOREBOARD_FILE} does not exist, creating empty file") with open(SCOREBOARD_FILE, "w") as f: json.dump([], f) return [] try: with open(SCOREBOARD_FILE, "r") as f: data = json.load(f) return [ScoreboardEntry(**entry) for entry in data] except (json.JSONDecodeError, TypeError) as e: logging.error(f"Error loading scoreboard: {str(e)}") return [] except Exception as e: logging.error(f"Unexpected error loading scoreboard: {str(e)}") return [] def _save_scoreboard(entries: List[ScoreboardEntry]): try: logging.info(f"Writing {len(entries)} entries to {SCOREBOARD_FILE}") with open(SCOREBOARD_FILE, "w") as f: json.dump([entry.dict() for entry in entries], f, indent=4) logging.info("Scoreboard file written successfully") except Exception as e: logging.error(f"Error saving scoreboard: {str(e)}") raise HTTPException(status_code=500, detail="Failed to save scoreboard") # Routes @app.get("/favicon.ico") async def favicon(): return FileResponse("templates/assets/favicon.ico") @app.get("/oat_quiz", response_class=HTMLResponse) async def get_quiz_page(request: Request): quiz_html_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "templates", "quiz.html") with open(quiz_html_file_path, 'r') as html_file: return HTMLResponse(content=html_file.read(), status_code=200) # API Endpoints for Quiz @app.get("/api/quiz/questions", response_model=List[QuizQuestionInDB]) async def get_quiz_questions(): return _load_quiz_questions() @app.post("/api/admin/quiz/questions", response_model=QuizQuestionInDB, status_code=201) async def create_quiz_question(question_data: QuizQuestionCreate): questions = _load_quiz_questions() new_question = QuizQuestionInDB(id=_generate_question_id(), **question_data.dict()) questions.append(new_question) _save_quiz_questions(questions) return new_question @app.put("/api/admin/quiz/questions/{question_id}", response_model=QuizQuestionInDB) async def update_quiz_question(question_id: str, question_update: QuizQuestionUpdate): questions = _load_quiz_questions() question_index = -1 for i, q in enumerate(questions): if q.id == question_id: question_index = i break if question_index == -1: raise HTTPException(status_code=404, detail="Question not found") current_question = questions[question_index] update_data = question_update.dict(exclude_unset=True) updated_question_data = current_question.dict() updated_question_data.update(update_data) # Validate that if options are updated, the answer is still valid, or if only answer is updated, it's valid with existing options. temp_options = updated_question_data.get('options') temp_answer = updated_question_data.get('answer') if temp_options and temp_answer and temp_answer not in temp_options: raise HTTPException(status_code=400, detail="Answer must be one of the provided options.") if temp_options and not (2 <= len(temp_options) <= 8): raise HTTPException(status_code=400, detail="Number of options must be between 2 and 8.") questions[question_index] = QuizQuestionInDB(**updated_question_data) _save_quiz_questions(questions) return questions[question_index] @app.delete("/api/admin/quiz/questions/{question_id}", status_code=204) async def delete_quiz_question(question_id: str): questions = _load_quiz_questions() question_found = False for i, q in enumerate(questions): if q.id == question_id: questions.pop(i) question_found = True break if not question_found: raise HTTPException(status_code=404, detail="Question not found") _save_quiz_questions(questions) return # No content for 204 # API Endpoints for Scoreboard @app.get("/api/quiz/scoreboard", response_model=List[ScoreboardEntry]) async def get_scoreboard(): try: logging.info("Fetching scoreboard entries") entries = _load_scoreboard() # Sort by time (ascending) and return all entries entries.sort(key=lambda x: x.time) logging.info(f"Returning {len(entries)} scoreboard entries") return entries # Return all entries instead of just top 5 except Exception as e: logging.error(f"Error in get_scoreboard: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/quiz/scoreboard", response_model=ScoreboardEntry, status_code=201) async def add_scoreboard_entry(entry: ScoreboardEntry): try: logging.info(f"Attempting to save scoreboard entry: {entry.dict()}") entries = _load_scoreboard() logging.info(f"Current scoreboard entries: {entries}") entries.append(entry) # Sort by time (ascending) and keep all entries entries.sort(key=lambda x: x.time) logging.info(f"Saving {len(entries)} entries to scoreboard") _save_scoreboard(entries) # Save all entries instead of just top 5 logging.info("Scoreboard entry saved successfully") return entry except Exception as e: logging.error(f"Error adding scoreboard entry: {str(e)}") raise HTTPException(status_code=500, detail=str(e))