import mysql.connector from flask_bcrypt import Bcrypt from flask import request, jsonify import logging import json import traceback from datetime import datetime from question_verify import verify_question from flask_jwt_extended import get_jwt_identity import uuid import random import re import json from generate_question import generate_ques import os import pymysql logging.basicConfig( filename=os.path.join('/tmp', 'app.log'), level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s' ) bcrypt = Bcrypt() # CA_CERT_PATH = os.path.join(os.getcwd(), 'ca.pem') # assert os.path.exists(CA_CERT_PATH), f"Missing: {CA_CERT_PATH}" def get_db_connection(): try: connection = pymysql.connect( host="quamble-quamble.c.aivencloud.com", port=17939, user="avnadmin", password=os.getenv("DB_PASSWORD"), database="quamble", ssl={'ca': '/app/ca.pem'}, cursorclass=pymysql.cursors.DictCursor ) print("Database connection established.") logging.info("Database connection established.") return connection except mysql.connector.Error as err : print(f"Error: {err}") logging.error("Error connecting to the database: %s", err) return None def add_user(username, email, password, role, preference_1=None, preference_2=None, preference_3=None, preference_4=None): try: user_id = str(uuid.uuid4()) connection = get_db_connection() if connection: cursor = connection.cursor() hashed_password = bcrypt.generate_password_hash(password).decode('utf-8') cursor.execute(""" INSERT INTO users (id, username, email, password, role, preference_1, preference_2, preference_3, preference_4) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, (user_id, username, email, hashed_password, role, preference_1, preference_2, preference_3, preference_4)) connection.commit() logging.info("User added: %s", username) except mysql.connector.Error as err: logging.error("Error adding user %s: %s", username, err) finally: #cursor.close() #connection.close() logging.info("Database connection closed after adding user.") def get_user_by_email(email): connection = get_db_connection() if connection: try: cursor = connection.cursor() cursor.execute("SELECT * FROM users WHERE email=%s", (email,)) user = cursor.fetchone() logging.info("User retrieved: %s", user) return user except mysql.connector.Error as err: logging.error("Error retrieving user by email %s: %s", email, err) return None finally: cursor.close() connection.close() logging.info("Database connection closed after retrieving user.") def add_quiz(user_id, theme, questions, correct_options): connection = get_db_connection() if not connection: logging.error("Database connection failed.") return None try: cursor = connection.cursor() cursor.execute("SELECT theme_quiz_table FROM Themes WHERE theme = %s", (theme,)) theme_data = cursor.fetchone() if not theme_data: logging.error("Invalid theme provided: %s", theme) return None theme_table = theme_data[0] cursor.execute("SELECT COUNT(*) FROM users WHERE id = %s", (user_id,)) user_exists = cursor.fetchone()[0] if user_exists == 0: logging.error("User ID does not exist: %s", user_id) return None cursor.execute("INSERT INTO quiz (user_id_creator, theme) VALUES (%s, %s)", (user_id, theme)) quiz_id = cursor.lastrowid questions_json = json.dumps(questions) correct_options_json = json.dumps(correct_options) query = f""" INSERT INTO {theme_table} (quiz_id, user_id_creator, questions_by_llm, correct_options_llm) VALUES (%s, %s, %s, %s) """ cursor.execute(query, (quiz_id, user_id, questions_json, correct_options_json)) connection.commit() logging.info("Quiz added for user_id %s with theme %s in table %s", user_id, theme, theme_table) return quiz_id except mysql.connector.Error as err: logging.error("Error adding quiz: %s", err) return None finally: if cursor: cursor.close() if connection: connection.close() logging.info("Database connection closed after adding quiz.") def add_question_from_master(): if request.method == 'POST': logging.info("Received a POST request to add questions to quiz.") theme = request.form.get('theme') if not theme: logging.error("Theme not provided in the request.") return jsonify({"error": "Please provide the theme."}), 400 question = request.form.get('question') correct_option = request.form.get('correct_option') if not question or not correct_option: logging.error("Missing question or correct option in the request.") return jsonify({"error": "Please provide both question and correct option."}), 400 question = question.replace("\t", "") validation, feedback, difficulty = verify_question(question, correct_option) if validation != 'valid': logging.warning(f"Verification failed: {feedback}") return jsonify({"error": f"Verification failed: {feedback}"}), 400 logging.info(f"Verified question: {question}") logging.info(f"Verified correct option: {correct_option}") logging.info(f"Difficulty level: {difficulty}") user_id_creator = get_jwt_identity() logging.info(f"Authenticated user ID: {user_id_creator}") connection = get_db_connection() if connection: try: cursor = connection.cursor() cursor.execute("SELECT theme_bank FROM themes WHERE theme = %s", (theme,)) result = cursor.fetchone() if not result: logging.error("Invalid theme provided: %s", theme) return jsonify({"error": "Invalid theme provided."}), 400 theme_bank = result[0] question_json = json.dumps({"question": question}) check_query = f""" SELECT COUNT(*) FROM {theme_bank} WHERE question_by_master = %s AND user_id_creator = %s """ cursor.execute(check_query, (question_json, user_id_creator)) is_duplicate = cursor.fetchone()[0] > 0 if is_duplicate: logging.warning("Duplicate question detected for user %s: %s", user_id_creator, question) return jsonify({"error": "Duplicate question detected."}), 400 insert_query = f""" INSERT INTO {theme_bank} (user_id_creator, question_by_master, correct_option_master, Difficulty) VALUES (%s, %s, %s, %s) """ cursor.execute(insert_query, (user_id_creator, question_json, correct_option, difficulty)) connection.commit() logging.info("Question added for theme %s", theme) return 'True' except mysql.connector.Error as err: logging.error("Error adding question: %s", err) return jsonify({"error": "Failed to add question to the database."}), 500 finally: #cursor.close() #connection.close() logging.info("Database connection closed.") else: logging.error("Database connection failed.") return jsonify({"error": "Database connection failed."}), 500 def add_ques_llm(theme, question, options, correct_option, difficulty): connection = get_db_connection() if connection: try: cursor = connection.cursor() cursor.execute("SELECT theme_bank FROM themes WHERE theme = %s", (theme,)) theme_bank_result = cursor.fetchone() if not theme_bank_result: logging.error("Invalid theme provided: %s", theme) return None theme_bank = theme_bank_result["theme_bank"] user_id_creator = get_jwt_identity() question_json = json.dumps({ "question": question, "options": options }) check_query = f"SELECT COUNT(*) FROM {theme_bank} WHERE question_by_llm = %s" cursor.execute(check_query, (question_json,)) is_duplicate = cursor.fetchone()["COUNT(*)"] > 0 if is_duplicate: logging.warning("Duplicate question detected for theme %s: %s", theme, question) return "Duplicate" insert_query = f""" INSERT INTO {theme_bank} (user_id_creator, question_by_llm, correct_option_llm, Difficulty) VALUES (%s, %s, %s, %s) """ cursor.execute(insert_query, (user_id_creator, question_json, correct_option, difficulty)) connection.commit() logging.info("Question added for theme %s", theme) return 'True' except mysql.connector.Error as err: logging.error("Error adding question: %s", err) return None finally: #cursor.close() #connection.close() logging.info("Database connection closed after adding question.") else: logging.error("Database connection failed.") return None def add_question_to_db(theme, question, correct_option, difficulty): connection = get_db_connection() try: cursor = connection.cursor() theme_check_query = "SELECT COUNT(*) FROM themes WHERE theme = %s" cursor.execute(theme_check_query, (theme,)) theme_exists = cursor.fetchone()["COUNT(*)"] if theme_exists == 0: insert_query = "INSERT INTO themes (theme, theme_bank, theme_quiz_table) VALUES (%s, %s, %s)" cursor.execute(insert_query, (theme, f"{theme}_bank", f"theme_{theme}")) safe_theme = re.sub(r'\W+', '_', theme) table_creation_query = f""" CREATE TABLE IF NOT EXISTS {safe_theme}_bank ( ques_id INT AUTO_INCREMENT PRIMARY KEY, user_id_creator VARCHAR(64), question_by_llm MEDIUMTEXT, question_hash VARCHAR(640), correct_option_llm MEDIUMTEXT, question_by_master MEDIUMTEXT, correct_option_master MEDIUMTEXT, Difficulty VARCHAR(64), is_used VARCHAR(64), Issue VARCHAR(6400) ) """ cursor.execute(table_creation_query) theme_table_creation_query = f""" CREATE TABLE IF NOT EXISTS theme_{safe_theme} ( quiz_id INT(8) PRIMARY KEY AUTO_INCREMENT, user_id_creator VARCHAR(64), questions_by_llm MEDIUMTEXT, correct_options_llm MEDIUMTEXT, questions_by_master MEDIUMTEXT, correct_options_master MEDIUMTEXT, num_questions INT(11), user_id_attempt MEDIUMTEXT, user_responses MEDIUMTEXT ) """ cursor.execute(theme_table_creation_query) check_query = f"SELECT COUNT(*) FROM `{safe_theme}_bank` WHERE question_by_llm = %s" cursor.execute(check_query, (question,)) is_duplicate = cursor.fetchone()["COUNT(*)"] > 0 if is_duplicate: logging.warning("Duplicate question detected for theme %s: %s", safe_theme, question) return "Duplicate" query = f"INSERT INTO {safe_theme}_bank (question_by_llm, correct_option_llm, Difficulty, is_used) VALUES (%s, %s, %s, 'unused')" cursor.execute(query, (question, correct_option, difficulty)) connection.commit() return 'True' except mysql.connector.Error as err: logging.error("Error adding question: %s", err) return None finally: #cursor.close() #connection.close() logging.info("Database connection closed") def create_quiz(theme, num_questions): connection = get_db_connection() if connection: try: cursor = connection.cursor() cursor.execute("SELECT theme_bank, theme_quiz_table FROM themes WHERE theme = %s", (theme,)) theme_data = cursor.fetchone() if not theme_data: logging.error("Invalid theme provided: %s", theme) return jsonify({"error": "Invalid theme provided."}), 400 theme_bank, theme_quiz_table = theme_data query_available_questions = f""" SELECT COUNT(*) FROM {theme_bank} WHERE question_by_llm IS NOT NULL AND question_by_llm != '' AND (is_used IS NULL OR is_used = '') """ cursor.execute(query_available_questions) available_count = cursor.fetchone()[0] if available_count < int(num_questions): logging.warning("Not enough questions available for theme %s: Available = %d, Requested = %d", theme, available_count, num_questions) return jsonify({"error": "Not enough questions available."}), 400 query_random_questions = f""" SELECT ques_id, question_by_llm, correct_option_llm FROM {theme_bank} WHERE question_by_llm IS NOT NULL AND question_by_llm != '' AND (is_used IS NULL OR is_used = '') ORDER BY RAND() LIMIT {int(num_questions)} """ cursor.execute(query_random_questions) selected_questions = cursor.fetchall() questions = [] correct_options = [] selected_question_ids = [] for question_id, question, correct_option in selected_questions: questions.append(question) correct_options.append(correct_option) selected_question_ids.append(question_id) mark_as_used_query = f""" UPDATE {theme_bank} SET is_used = 'used' WHERE ques_id IN ({', '.join(map(str, selected_question_ids))}) """ cursor.execute(mark_as_used_query) user_id_creator = get_jwt_identity() cursor.execute("INSERT INTO quiz (user_id_creator, theme) VALUES (%s, %s)", (user_id_creator, theme)) quiz_id = cursor.lastrowid questions_json = json.dumps(questions) correct_options_json = json.dumps(correct_options) insert_quiz_query = f""" INSERT INTO {theme_quiz_table} (quiz_id, user_id_creator, questions_by_llm, correct_options_llm, num_questions) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(insert_quiz_query, (quiz_id, user_id_creator, questions_json, correct_options_json, num_questions)) connection.commit() logging.info("Quiz created successfully for theme %s with %d questions.", theme, num_questions) questions_reflect = json.loads(questions_json) correct_options_reflect = json.loads(correct_options_json) return jsonify({ "quiz_id": quiz_id, "questions": questions_reflect, "correct_options": correct_options_reflect }), 201 except mysql.connector.Error as err: logging.error("Error creating quiz: %s", err) return jsonify({"error": "Failed to create quiz."}), 500 except Exception as e: logging.error("An unexpected error occurred while creating the quiz: %s", str(e)) return jsonify({"error": "An unexpected error occurred."}), 500 finally: if cursor: cursor.close() if connection: connection.close() logging.info("Database connection closed.") else: logging.error("Database connection failed.") return jsonify({"error": "Database connection failed."}), 500 def create_quiz_master(user_id_creator, theme, num_questions): connection = get_db_connection() if connection: try: cursor = connection.cursor() cursor.execute("SELECT theme_bank, theme_quiz_table FROM themes WHERE theme = %s", (theme,)) theme_data = cursor.fetchone() if not theme_data: logging.error("Invalid theme provided: %s", theme) return jsonify({"error": "Invalid theme provided."}), 400 theme_bank, theme_quiz_table = theme_data query_available_questions = f""" SELECT COUNT(*) FROM {theme_bank} WHERE question_by_master IS NOT NULL AND is_used IS NULL OR is_used = "unused" """ cursor.execute(query_available_questions) available_count = cursor.fetchone()[0] if available_count < int(num_questions): logging.warning("Not enough questions available for theme %s: Available = %d, Requested = %d", theme, available_count, num_questions) return jsonify({"error": "Not enough questions available."}), 400 query_random_questions = f""" SELECT ques_id, question_by_master, correct_option_master FROM {theme_bank} WHERE question_by_master IS NOT NULL AND question_by_master != '' ORDER BY RAND() LIMIT {int(num_questions)} """ cursor.execute(query_random_questions) selected_questions = cursor.fetchall() questions, correct_options, selected_question_ids = [], [], [] for question_id, question, correct_option in selected_questions: questions.append(question) correct_options.append(correct_option) selected_question_ids.append(question_id) mark_as_used_query = f""" UPDATE {theme_bank} SET is_used = 'used' WHERE ques_id IN ({', '.join(map(str, selected_question_ids))}) """ cursor.execute(mark_as_used_query) cursor.execute("INSERT INTO quiz (user_id_creator, theme) VALUES (%s, %s)", (user_id_creator, theme)) quiz_id = cursor.lastrowid insert_quiz_query = f""" INSERT INTO {theme_quiz_table} (quiz_id, user_id_creator, questions_by_master, correct_options_master, num_questions) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(insert_quiz_query, (quiz_id, user_id_creator, json.dumps(questions), json.dumps(correct_options), num_questions)) connection.commit() return jsonify({ "quiz_id": quiz_id, "ques_id": selected_question_ids, "correct_options": correct_options }), 201 except mysql.connector.Error as err: logging.error("Error creating quiz: %s", err) return jsonify({"error": "Failed to create quiz."}), 500 except Exception as e: logging.error("An unexpected error occurred while creating the quiz: %s", str(e)) return jsonify({"error": "An unexpected error occurred."}), 500 finally: if cursor: cursor.close() if connection: connection.close() logging.info("Database connection closed.") else: logging.error("Database connection failed.") return jsonify({"error": "Database connection failed."}), 500 def recording_issue(theme, ques_id, issue_description): connection = get_db_connection() if connection: try: cursor = connection.cursor() cursor.execute("SELECT theme_bank FROM themes WHERE theme = %s", (theme,)) theme_data = cursor.fetchone() if not theme_data: logging.error("Invalid theme provided: %s", theme) return jsonify({"error": "Invalid theme provided."}), 400 theme_bank = theme_data[0] select_query = f"SELECT COUNT(*) FROM {theme_bank} WHERE ques_id = %s" cursor.execute(select_query, (ques_id,)) count = cursor.fetchone()[0] if count == 0: return jsonify({"error": f"Question ID {ques_id} does not exist in the database"}), 400 update_query = f""" UPDATE {theme_bank} SET Issue = %s WHERE ques_id = %s """ cursor.execute(update_query, (issue_description, ques_id)) connection.commit() logging.info("Issue recorded for theme %s and question_id %s.", theme, ques_id) return jsonify({"message": "Issue reported successfully"}), 201 except mysql.connector.Error as err: return jsonify({"error": "An error occurred while reporting the issue", "details": str(err)}), 500 finally: cursor.close() connection.close() logging.info("Database connection closed.") else: logging.error("Database connection failed.") return jsonify({"error": "Database connection failed."}), 500 def record_feedback(user_id, rating, comments): connection = get_db_connection() if connection: try: cursor = connection.cursor() insert_query= """ INSERT INTO feedbacks (user_id, rating, comments) VALUES (%s, %s, %s) """ cursor.execute(insert_query, (user_id, rating, comments)) # Commit the transaction and close the connection connection.commit() cursor.close() connection.close() return jsonify({"message": "Feedback submitted successfully"}), 201 except mysql.connector.Error as err: print(f"Error: {err}") return jsonify({"error": "Failed to submit feedback"}), 500 def create_quiz_by_id(theme, num_questions): connection = get_db_connection() if not connection: logging.error("Database connection failed.") return jsonify({"error": "Database connection failed."}), 500 try: cursor = connection.cursor() cursor.execute("SELECT theme_bank FROM themes WHERE theme = %s", (theme,)) result = cursor.fetchone() if not result: logging.error("Invalid theme provided: %s", theme) return jsonify({"error": "Invalid theme provided."}), 400 theme_bank = result[0] cursor.execute(f""" SELECT COUNT(*) FROM {theme_bank} WHERE question_by_llm IS NOT NULL AND is_used IS NULL OR is_used = "unused" """) available_count = cursor.fetchone()[0] shortfall = int(num_questions) - available_count logging.info("Available questions: %d; Required: %d; Shortfall: %d", available_count, int(num_questions), shortfall) added_count = 0 attempts = 0 max_attempts = shortfall * 10 while added_count < shortfall and attempts < max_attempts: attempts += 1 ques = generate_ques(theme) question_match = re.search(r'Question:\s*(.*)', ques) options_matches = re.findall(r'([A-D])\)\s*(.*)', ques) correct_answer_match = re.search(r'Correct answer:\s*([A-D])', ques) difficulty_match = re.search(r'Difficulty level:\s*(.*)', ques) if question_match and options_matches and correct_answer_match: question = question_match.group(1).strip() options = [f"{opt[0]}) {opt[1].strip()}" for opt in options_matches] correct_option = correct_answer_match.group(1).strip().upper() difficulty = difficulty_match.group(1).strip() if difficulty_match else "medium" if correct_option not in ['A', 'B', 'C', 'D']: continue status = add_ques_llm(theme, question, options, correct_option, difficulty) if status == "True": added_count += 1 elif status == "Duplicate": continue # retry else: logging.warning("Failed to insert generated question.") else: logging.warning("Malformed question generated; skipping.") if added_count < shortfall: return jsonify({"error": f"Could not generate enough unique questions. Needed {shortfall}, added {added_count}."}), 400 cursor.execute(f""" SELECT COUNT(*) FROM {theme_bank} WHERE question_by_llm IS NOT NULL AND is_used IS NULL OR is_used = "unused" """) available_count = cursor.fetchone()[0] if available_count < int(num_questions): logging.warning("Not enough questions available for theme %s: Available = %d, Requested = %d", theme, available_count, num_questions) return jsonify({"error": "Not enough questions available."}), 400 cursor.execute(f""" SELECT ques_id, correct_option_llm FROM {theme_bank} WHERE question_by_llm IS NOT NULL AND question_by_llm != '' AND (is_used IS NULL OR is_used = '') ORDER BY RAND() LIMIT {int(num_questions)} """) selected_questions = cursor.fetchall() questions = [q[0] for q in selected_questions] correct_options = [q[1] for q in selected_questions] cursor.execute(f""" UPDATE {theme_bank} SET is_used = 'used' WHERE ques_id IN ({', '.join(map(str, questions))}) """) user_id_creator = get_jwt_identity() cursor.execute("INSERT INTO quiz (user_id_creator, theme) VALUES (%s, %s)", (user_id_creator, theme)) quiz_id = cursor.lastrowid cursor.execute("SELECT theme_quiz_table FROM themes WHERE theme = %s", (theme,)) theme_table = cursor.fetchone()[0] cursor.execute(f""" INSERT INTO {theme_table} (quiz_id, user_id_creator, questions_by_llm, correct_options_llm, num_questions) VALUES (%s, %s, %s, %s, %s) """, (quiz_id, user_id_creator, json.dumps(questions), json.dumps(correct_options), num_questions)) connection.commit() logging.info("Quiz created successfully for theme %s with %d questions.", theme, num_questions) return jsonify({"quiz_id": quiz_id, "questions": questions, "correct_options": correct_options}), 201 except mysql.connector.Error as err: logging.error("Error creating quiz: %s", err) return jsonify({"error": "Failed to create quiz."}), 500 except Exception as e: logging.error("An unexpected error occurred while creating the quiz: %s", str(e)) return jsonify({"error": "An unexpected error occurred."}), 500 finally: if cursor: cursor.close() if connection: connection.close() logging.info("Database connection closed.") def add_theme_if_not_exists(theme): connection = get_db_connection() try: cursor = connection.cursor() bank_table = f"{theme}_bank" quiz_table = f"theme_{theme}" theme_check_query = "SELECT COUNT(*) FROM themes WHERE theme = %s" cursor.execute(theme_check_query, (theme,)) theme_exists = cursor.fetchone()["COUNT(*)"] if theme_exists == 0: insert_query = "INSERT INTO themes (theme, theme_bank, theme_quiz_table) VALUES (%s, %s, %s)" cursor.execute(insert_query, (theme, f"{theme}_bank", f"theme_{theme}")) connection.commit() create_bank_table = f""" CREATE TABLE IF NOT EXISTS `{bank_table}` ( ques_id INT PRIMARY KEY AUTO_INCREMENT, user_id_creator VARCHAR(64), question_by_llm MEDIUMTEXT, question_hash VARCHAR(640), correct_option_llm MEDIUMTEXT, question_by_master MEDIUMTEXT, correct_option_master MEDIUMTEXT, Difficulty VARCHAR(64), is_used VARCHAR(64), Issue VARCHAR(6400) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; """ cursor.execute(create_bank_table) logging.info("Created table: %s", bank_table) # Create theme_ table create_quiz_table = f""" CREATE TABLE IF NOT EXISTS `{quiz_table}` ( quiz_id INT PRIMARY KEY AUTO_INCREMENT, user_id_creator VARCHAR(64), questions_by_llm MEDIUMTEXT, correct_options_llm MEDIUMTEXT, questions_by_master MEDIUMTEXT, correct_options_master MEDIUMTEXT, num_questions INT, user_id_attempt MEDIUMTEXT, user_responses MEDIUMTEXT ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; """ cursor.execute(create_quiz_table) logging.info("Created table: %s", quiz_table) except mysql.connector.Error as err: logging.error("Error adding theme: %s", err) return jsonify({"error": "Failed to add theme."}), 500 finally: cursor.close() connection.close() logging.info("Database connection is closed") return not theme_exists def profile_added_db(first_name, last_name, organisation, industry, bio): connection = get_db_connection() if connection: try: cursor = connection.cursor() user_id_creator = get_jwt_identity() logging.info(f"Authenticated user ID: {user_id_creator}") query_check_user = "SELECT COUNT(*) AS user_count FROM users WHERE id = %s" cursor.execute(query_check_user, (user_id_creator,)) user_exists = cursor.fetchone()['user_count'] if user_exists: query_update_user = """ UPDATE users SET first_name = %s, last_name = %s, organisation = %s, industry = %s, bio = %s WHERE id = %s """ cursor.execute(query_update_user, (first_name, last_name, organisation, industry, bio, user_id_creator)) connection.commit() logging.info(f"Profile for user {user_id_creator} updated successfully.") return jsonify({"message": "Profile updated successfully!"}), 200 else: logging.warning(f"User with ID {user_id_creator} does not exist.") return jsonify({"error": "User does not exist."}), 404 except Exception as e: logging.error(f"Unexpected error: {e}") return jsonify({"error":"Failed to update profile"}), 500 finally: cursor.close() connection.close() logging.info("Database connection closed.") def view_profile_db(): try: user_id_creator = get_jwt_identity() logging.info(f"Authenticated user ID: {user_id_creator}") connection = get_db_connection() if connection: cursor = connection.cursor() query_check_user = "SELECT COUNT(*) AS user_count FROM users WHERE id = %s" cursor.execute(query_check_user, (user_id_creator,)) user_exists = cursor.fetchone()['user_count'] if user_exists: query_get_profile = """ SELECT first_name, last_name, role, organisation, industry, bio FROM users WHERE id = %s """ cursor.execute(query_get_profile, (user_id_creator,)) profile = cursor.fetchone() if profile: logging.info(f"User {user_id_creator}'s profile fetched successfully.") return jsonify({ "first_name": profile["first_name"], "last_name": profile["last_name"], "role": profile["role"], "organisation": profile["organisation"], "industry": profile["industry"], "bio": profile["bio"] }), 200 else: logging.warning(f"Profile for user {user_id_creator} not found.") return jsonify({"error": "Profile not found"}), 404 else: logging.warning(f"User with ID {user_id_creator} does not exist.") return jsonify({"error": "User does not exist."}), 404 except Exception as e: logging.error(f"Unexpected error: {e}") return jsonify({"error": f"Unexpected error: {str(e)}"}), 500 finally: cursor.close() connection.close() logging.info("Database connection closed.") def view_quiz_score_db(user_id, theme_table, theme, quiz_id): try: connection = get_db_connection() if connection: cursor = connection.cursor() query = f""" SELECT num_questions FROM {theme_table} WHERE quiz_id = %s """ cursor.execute(query, (quiz_id,)) theme_data = cursor.fetchone() num_questions = theme_data['num_questions'] if theme_data else None if num_questions is None: logging.warning(f"Theme '{theme}' not found in theme_table.") return jsonify({"error": "Theme not found"}), 404 query = """ SELECT score, time_taken, user_response FROM quiz_response WHERE quiz_id = %s AND user_id_attempt = %s """ cursor.execute(query, (quiz_id, user_id)) quiz_response = cursor.fetchone() if quiz_response: user_response = quiz_response['user_response'] score = quiz_response['score'] time = quiz_response['time_taken'] accuracy = (score / num_questions) * 100 if num_questions > 0 else 0 response_data = { "quiz_id": quiz_id, "theme": theme, "user_response": user_response, "score": score, "accuracy": round(accuracy, 2), "time_taken": time } logging.info(f"Score and accuracy for user {user_id}, quiz {quiz_id} fetched successfully.") return jsonify(response_data), 200 logging.warning(f"No quiz response found for user {user_id} and quiz {quiz_id}.") return jsonify({"error": "Quiz response not found"}), 404 except Exception as e: logging.error(f"Error fetching quiz score: {e}") return jsonify({"error": "Unable to fetch quiz score"}), 500 finally: cursor.close() connection.close() def get_recent_quizzes_db(user_id, limit=5): try: connection = get_db_connection() if not connection: return {"error": "Database connection failed."} cursor = connection.cursor() cursor.execute(""" SELECT qr.quiz_id, qr.theme, qr.user_response, qr.score, qr.time_taken, qr.submitted_on, t.theme_quiz_table FROM quiz_response qr JOIN themes t ON qr.theme = t.theme WHERE qr.user_id_attempt = %s ORDER BY qr.submitted_on DESC LIMIT %s """, (user_id, limit)) recent_quizzes = cursor.fetchall() if not recent_quizzes: return {"message": "No quizzes found for the user."} results = [] for quiz in recent_quizzes: quiz_id = quiz['quiz_id'] theme = quiz['theme'] theme_quiz_table = quiz['theme_quiz_table'] if not theme_quiz_table: logging.warning(f"No quiz table found for theme: {theme}") continue cursor.execute(f""" SELECT questions_by_llm, correct_options_llm, questions_by_master, correct_options_master FROM {theme_quiz_table} WHERE quiz_id = %s """, (quiz_id,)) quiz_details = cursor.fetchone() if not quiz_details: logging.warning(f"Quiz '{quiz_id}' not found in theme table '{theme_quiz_table}'.") continue questions = quiz_details['questions_by_llm'] or quiz_details['questions_by_master'] correct_answers = quiz_details['correct_options_llm'] or quiz_details['correct_options_master'] if not questions or not correct_answers: logging.warning(f"Quiz '{quiz_id}' has no questions or answers.") continue results.append({ "quiz_id": quiz_id, "theme": theme, "questions": questions, "correct_answers": correct_answers, "user_responses": quiz['user_response'], "score": quiz["score"], "time_taken": quiz['time_taken'], "date_attempted": quiz['submitted_on'] }) return results except mysql.connector.Error as err: logging.error(f"MySQL Error: {err}") return {"error": "Database operation failed."} except Exception as e: logging.error(f"Unexpected error: {str(e)}") return {"error": "An unexpected error occurred."} finally: if 'cursor' in locals(): cursor.close() if 'connection' in locals(): connection.close() logging.info("Database connection closed.") def get_quiz_details_db(user_id, quiz_id): try: connection = get_db_connection() if not connection: return {"error": "Database connection failed."} cursor = connection.cursor() logging.info(f"Checking for user_id_attempt: {user_id}, quiz_id: {quiz_id}") cursor.execute(""" SELECT qr.quiz_id, qr.theme, qr.user_response, qr.score, qr.time_taken, qr.submitted_on, t.theme_quiz_table FROM quiz_response qr JOIN themes t ON qr.theme = t.theme WHERE qr.user_id_attempt = %s AND qr.quiz_id = %s """, (user_id, quiz_id)) quiz = cursor.fetchone() if not quiz: return {"message": "Quiz not found."} theme_quiz_table = quiz['theme_quiz_table'] if not theme_quiz_table: logging.warning(f"No quiz table found for theme: {quiz['theme']}") return {"error": "Invalid theme data."} cursor.execute(f""" SELECT questions_by_llm, correct_options_llm, questions_by_master, correct_options_master FROM {theme_quiz_table} WHERE quiz_id = %s """, (quiz_id,)) quiz_details = cursor.fetchone() if not quiz_details: logging.warning(f"Quiz '{quiz_id}' not found in theme table '{theme_quiz_table}'.") return {"error": "Quiz questions not found."} questions = quiz_details['questions_by_llm'] or quiz_details['questions_by_master'] correct_answers = quiz_details['correct_options_llm'] or quiz_details['correct_options_master'] if not questions or not correct_answers: logging.warning(f"Quiz '{quiz_id}' has incomplete data.") return {"error": "Incomplete quiz data."} return { "quiz_id": quiz["quiz_id"], "theme": quiz["theme"], "questions": questions, "correct_answers": correct_answers, "user_responses": quiz["user_response"], "score": quiz["score"], "time_taken": quiz["time_taken"], "date_attempted": quiz["submitted_on"] } except mysql.connector.Error as err: logging.error(f"MySQL Error: {err}") return {"error": "Database operation failed."} except Exception as e: logging.error(f"Unexpected error: {str(e)}") return {"error": "An unexpected error occurred."} finally: if 'cursor' in locals(): cursor.close() if 'connection' in locals(): connection.close() logging.info("Database connection closed.") def fetch_quiz_for_theme_db(user_id, theme): try: connection = get_db_connection() if not connection: return {"error": "Database connection failed."} cursor = connection.cursor() cursor.execute("SELECT theme_quiz_table, theme_bank FROM themes WHERE theme = %s", (theme,)) theme_data = cursor.fetchone() if not theme_data or not theme_data['theme_quiz_table'] or not theme_data['theme_bank']: return {"error": "No quiz table found for this theme."} theme_quiz_table = theme_data['theme_quiz_table'] theme_bank_table = theme_data['theme_bank'] cursor.execute("SELECT quiz_id FROM quiz_response WHERE user_id_attempt = %s", (user_id,)) attempted_quizzes = {row['quiz_id'] for row in cursor.fetchall()} cursor.execute(f""" SELECT quiz_id, num_questions, questions_by_llm, correct_options_llm, questions_by_master, correct_options_master FROM {theme_quiz_table} """) quizzes = cursor.fetchall() if not quizzes: return {"message": "No new quizzes available for this theme."} results = [] for quiz in quizzes: quiz_id = quiz['quiz_id'] if quiz_id in attempted_quizzes: continue ques_id_list = None if quiz['questions_by_llm']: try: ques_id_list = json.loads(quiz['questions_by_llm']) except json.JSONDecodeError: logging.error(f"Invalid JSON in questions_by_llm for quiz_id {quiz_id}: {quiz['questions_by_llm']}") continue elif quiz['questions_by_master']: try: ques_id_list = json.loads(quiz['questions_by_master']) except json.JSONDecodeError: logging.error(f"Invalid JSON in questions_by_master for quiz_id {quiz_id}: {quiz['questions_by_master']}") continue logging.info(f"Quiz ID {quiz_id} - Extracted ques_id_list: {ques_id_list} (Type: {type(ques_id_list)})") if isinstance(ques_id_list, dict): ques_id_list = list(ques_id_list.values()) elif not isinstance(ques_id_list, list): logging.error(f"Invalid ques_id_list type for quiz_id {quiz_id}: {type(ques_id_list)}") continue if not ques_id_list: logging.warning(f"No questions found for quiz_id {quiz_id}") continue number_of_questions = quiz['num_questions'] format_strings = ",".join(["%s"] * len(ques_id_list)) logging.info(f"Executing SQL query with ques_id_list: {ques_id_list}") cursor.execute(f""" SELECT ques_id, question_by_llm, correct_option_llm, question_by_master, correct_option_master FROM {theme_bank_table} WHERE ques_id IN ({format_strings}) """, tuple(ques_id_list)) question_details = cursor.fetchall() final_questions = [] final_correct_options = [] final_ques_ids = [] for q in question_details: question_text = q['question_by_llm'] or q['question_by_master'] correct_option = q['correct_option_llm'] or q['correct_option_master'] if question_text and correct_option: final_questions.append(question_text) final_correct_options.append(correct_option) final_ques_ids.append(q['ques_id']) results.append({ "quiz_id": quiz_id, "num_questions": number_of_questions, "ques_id": final_ques_ids, "questions": final_questions, "correct_options": final_correct_options }) return {"quizzes": results} except mysql.connector.Error as err: logging.error(f"MySQL Error: {err}") return {"error": "Database operation failed."} except Exception as e: logging.error(f"Unexpected error: {str(e)}") return {"error": "An unexpected error occurred."} finally: if 'cursor' in locals(): cursor.close() if 'connection' in locals(): connection.close() logging.info("Database connection closed.")