Spaces:
Sleeping
Sleeping
| import json | |
| import logging | |
| from flask import jsonify, request | |
| from flask_jwt_extended import get_jwt_identity | |
| from models import get_db_connection | |
| import mysql.connector | |
| from datetime import datetime | |
| import traceback | |
| def submit_quiz(): | |
| if request.method == 'POST': | |
| logging.info("Received a POST request to submit quiz.") | |
| quiz_id = request.form.get('quiz_id') | |
| user_response = request.form.getlist('user_response') | |
| theme = request.form.get('theme') | |
| start_time = request.form.get('start_time') | |
| end_time = request.form.get('end_time') | |
| logging.info(f"Received data: quiz_id={quiz_id}, user_response={user_response}, theme={theme}") | |
| if not quiz_id or not user_response or not theme or not start_time or not end_time: | |
| return jsonify({"error": "Please provide all required fields."}), 400 | |
| user_id_attempt = get_jwt_identity() | |
| logging.info(f"Authenticated user ID: {user_id_attempt}") | |
| connection = get_db_connection() | |
| if not connection: | |
| return jsonify({"error": "Database connection failed."}), 500 | |
| cursor = connection.cursor() | |
| logging.info("Database connection established.") | |
| try: | |
| cursor.execute("SELECT theme_quiz_table FROM themes WHERE theme = %s", (theme,)) | |
| theme_entry = cursor.fetchone() | |
| if not theme_entry: | |
| return jsonify({"error": "Invalid theme or theme not supported."}), 400 | |
| theme_table = theme_entry['theme_quiz_table'] | |
| cursor.execute(f"SELECT * FROM {theme_table} WHERE quiz_id = %s AND FIND_IN_SET(%s, user_id_attempt) > 0", (quiz_id, user_id_attempt)) | |
| if cursor.fetchone(): | |
| return jsonify({"error": "You have already submitted this quiz."}), 403 | |
| cursor.execute(f"SELECT * FROM {theme_table} WHERE quiz_id = %s", (quiz_id,)) | |
| quiz = cursor.fetchone() | |
| if not quiz: | |
| return jsonify({"error": "Quiz not found."}), 404 | |
| questions_data = json.loads(quiz.get('questions_by_llm') or quiz.get('questions_by_master', '[]')) | |
| correct_options = json.loads(quiz.get('correct_options_llm') or quiz.get('correct_options_master', '[]')) | |
| if len(user_response) != len(correct_options): | |
| return jsonify({"error": "Number of responses does not match the number of questions."}), 400 | |
| score = sum(1 for i in range(len(correct_options)) if user_response[i] == correct_options[i]) | |
| time_taken = str(datetime.strptime(end_time, "%H:%M:%S") - datetime.strptime(start_time, "%H:%M:%S")) | |
| cursor.execute(""" | |
| INSERT INTO quiz_response (quiz_id, theme, user_id_attempt, user_response, score, time_taken) | |
| VALUES (%s, %s, %s, %s, %s, %s) | |
| """, (quiz_id, theme, user_id_attempt, json.dumps(user_response), score, time_taken)) | |
| cursor.execute(f""" | |
| UPDATE {theme_table} | |
| SET user_id_attempt = CONCAT(COALESCE(user_id_attempt, ''), %s, ',') | |
| WHERE quiz_id = %s | |
| """, (user_id_attempt, quiz_id)) | |
| cursor.execute(""" | |
| UPDATE users | |
| SET total_score = total_score + %s | |
| WHERE id = %s | |
| """, (score, user_id_attempt)) | |
| connection.commit() | |
| logging.info(f"User {user_id_attempt} submitted quiz {quiz_id} with score {score} and time taken {time_taken}.") | |
| return jsonify({"quiz_id": quiz_id, "score": score, "time_taken": time_taken}), 200 | |
| except mysql.connector.Error as err: | |
| logging.error(f"MySQL Error: {err}") | |
| return jsonify({"error": "Database operation failed."}), 500 | |
| except Exception as e: | |
| logging.error(f"Unexpected error: {str(e)}") | |
| return jsonify({"error": "An unexpected error occurred."}), 500 | |
| finally: | |
| cursor.close() | |
| connection.close() | |
| logging.info("Database connection closed.") | |