import json import os import uuid import requests from flask import Blueprint, request, jsonify, g, url_for from uuid import UUID from db.model import db, User, Course, Enrollment,Chat, Quiz, QuizAttempt from utils.auth import auth_required import requests from config import SPAM_SCORE_THRESHOLD, AI_SPAM_SERVICES_MICROSERVICE, USER_UPLOADS_DIR, AI_QUIZ_SERVICES_MICROSERVICE from sqlalchemy import desc, select, and_ quiz = Blueprint('quiz', __name__) @quiz.route('/generate', methods=['POST']) @auth_required() def generate_quiz(): try: course_id: uuid.UUID = uuid.UUID(request.form['course_id']) current_page: int = int(request.form['page']) except KeyError: return jsonify({'message': 'course_id and page must be specified'}), 401 enrollment_record: Enrollment = db.session.execute( select(Enrollment).where(and_( Enrollment.courseID == course_id, Enrollment.userID == g.current_user.id) ) ).scalar() if not enrollment_record: return jsonify({"error": "You are not enrolled in this course."}), 403 if current_page > enrollment_record.course.totalPages or current_page < 1: return jsonify({ 'message': 'Page range out of bound. No such page' }), 404 # Everything is alright, now get the text in current page and generate quiz current_page_text: str = '' with open( os.path.join( USER_UPLOADS_DIR, enrollment_record.course.serverFilename+"_parts", f"{current_page}.txt") ) as f: current_page_text = f.read() quiz_data_resp = requests.post(AI_QUIZ_SERVICES_MICROSERVICE, json={"string_message": current_page_text}) if quiz_data_resp.status_code != 200: return jsonify({"error": "Failed to make quiz request."}), 500 quiz_data = quiz_data_resp.json() # Insert the quiz into table rows: list[Quiz] = [] for quiz_item in quiz_data['questions']: rows.append( Quiz( creatorUserID=g.current_user.id, creatorUser=g.current_user, quiz_attempts=[], courseID=course_id, course=enrollment_record.course, quizQuestion=quiz_item['question'], quizAnswers=json.dumps(quiz_item['options']), quizCorrectAnswer=quiz_item['correct_answer'] ) ) db.session.add_all(rows) db.session.commit() return jsonify({'message': 'quizzes were generated for the current page'}) @quiz.route('/get/personalIncomplete') @auth_required() def get_incomplete_quiz(): try: course_id: uuid.UUID = uuid.UUID(request.form['course_id']) except KeyError: return jsonify({'message': 'course_id must be specified'}), 401 quiz_rows: list[Quiz] = db.session.execute(select(Quiz).where( and_(Quiz.creatorUserID == g.current_user.id, Quiz.creatorHasAttempted == False) )).scalars() data: list = [] for quiz_row in quiz_rows: data.append( { 'id': quiz_row.id, 'isActive': quiz_row.isActive, 'creationDate': quiz_row.creationDate, 'quizAnswers': quiz_row.quizAnswers, 'quizQuestion': quiz_row.quizQuestion, 'course': { 'id': quiz_row.course.id, 'name': quiz_row.course.name, 'description': quiz_row.course.description }, 'creator': { 'id': quiz_row.creatorUserID, 'firstName': quiz_row.creatorUser.firstName, 'lastName': quiz_row.creatorUser.lastName, 'username': quiz_row.creatorUser.username, 'pfpFilename': url_for('send_file', filename=quiz_row.creatorUser.pfpFilename) } } ) return jsonify({ 'count': len(data), 'data': data }), 200 @quiz.route('/get/allComplete') @auth_required() def get_complete_quiz(): try: course_id: uuid.UUID = uuid.UUID(request.form['course_id']) except KeyError: return jsonify({'message': 'course_id must be specified'}), 401 quiz_attempts: list[QuizAttempt] = db.session.execute( select(QuizAttempt).where(and_( QuizAttempt.userID == g.current_user.id, Course.id == course_id ))).scalars() # IF THIS DOES NOT WORK, ADD COURSE IF TO QUIZ_ATTEMPT TABLE ITSELF completes: list = [] for attempt in quiz_attempts: quiz_row: Quiz = attempt.quiz completes.append( { 'id': attempt.id, 'quizID': quiz_row.id, 'isActive': quiz_row.isActive, 'creationDate': quiz_row.creationDate, 'quizAnswers': quiz_row.quizAnswers, 'quizQuestion': quiz_row.quizQuestion, 'userAnswer': attempt.userAnswer, 'quizCorrectAnswer': quiz_row.quizCorrectAnswer, 'isCorrect': attempt.isCorrect, 'course': { 'id': quiz_row.course.id, 'name': quiz_row.course.name, 'description': quiz_row.course.description }, 'creator': { 'id': quiz_row.creatorUserID, 'firstName': quiz_row.creatorUser.firstName, 'lastName': quiz_row.creatorUser.lastName, 'username': quiz_row.creatorUser.username, 'pfpFilename': url_for('send_file', filename=quiz_row.creatorUser.pfpFilename) } } ) return jsonify({ 'count': len(completes), 'data': completes }), 200 @quiz.route('/submit',methods=['POST']) @auth_required() def submit_quiz(): try: answer: str = request.form['answer'].strip() quiz_id: uuid.UUID = uuid.UUID(request.form['quiz_id']) except KeyError: return jsonify({'message': 'course_id and answer must be specified'}), 401 quiz_already_attempted: QuizAttempt = db.session.execute(select(QuizAttempt).where( and_(QuizAttempt.quizID == quiz_id, QuizAttempt.userID == g.current_user.id ) )).scalar() if quiz_already_attempted: return jsonify({'message': 'Already attempted this quiz'}), 401 quiz_row: Quiz = db.session.execute(select(Quiz).where(Quiz.id == quiz_id)).scalar() if not quiz_row: return jsonify({'message': 'Quiz does not exist'}), 404 valid_answers: list = json.loads(quiz_row.quizAnswers) is_correct: bool = False if answer not in valid_answers: return jsonify({'message': 'No such choice of answer given'}), 404 if answer == quiz_row.quizCorrectAnswer: is_correct = True new_attempt: QuizAttempt = QuizAttempt( userID=g.current_user.id, user=g.current_user, quizID=quiz_id, quiz=quiz_row, userAnswer=answer, isCorrect=int(is_correct) ) db.session.add(new_attempt) if quiz_row.creatorUser.id == g.current_user.id: quiz_row.creatorHasAttempted = True db.session.commit() return jsonify({ 'message': 'Answer submitted', 'isCorrect': is_correct, 'attemptID': new_attempt.id, 'quizID': quiz_row.id, 'quizAnswers': quiz_row.quizAnswers, 'quizQuestion': quiz_row.quizQuestion, 'quizCorrectAnswer': quiz_row.quizCorrectAnswer, 'userAnswer': answer }) @quiz.route('/quizData') @auth_required() def get_quiz_info(): try: quiz_id: uuid.UUID = uuid.UUID(request.args['quiz_id']) except KeyError: return jsonify({'message': 'quiz_id must be specified'}), 401 quiz_row: Quiz = db.session.execute(select(Quiz).where(Quiz.id == quiz_id)).scalar() if not quiz_row: return jsonify({'message': 'Quiz does not exist'}), 404 return jsonify({ 'id': quiz_row.id, 'isActive': quiz_row.isActive, 'creationDate': quiz_row.creationDate, 'quizAnswers': quiz_row.quizAnswers, 'quizQuestion': quiz_row.quizQuestion, 'course': { 'id': quiz_row.course.id, 'name': quiz_row.course.name, 'description': quiz_row.course.description }, 'creator': { 'id': quiz_row.creatorUserID, 'firstName': quiz_row.creatorUser.firstName, 'lastName': quiz_row.creatorUser.lastName, 'username': quiz_row.creatorUser.username, 'pfpFilename': url_for('send_file', filename=quiz_row.creatorUser.pfpFilename) } }), 200