Casu Al Snek
6 months ago
parent
716e5cffb4
commit
607902fd48
@ -1,3 +1,48 @@ |
||||
from flask import Blueprint |
||||
from flask import Blueprint, url_for, jsonify, g |
||||
from utils.auth import auth_required |
||||
from db.model import db, Badge, UserBadge |
||||
from sqlalchemy import select |
||||
badge_route = Blueprint('badge', __name__) |
||||
|
||||
badge = Blueprint('badge', __name__) |
||||
@badge_route.route('/listAllBadges') |
||||
def all_badges(): |
||||
badges: list[Badge] = db.session.execute(select(Badge)).scalars() |
||||
data: list = [] |
||||
for bgd in badges: |
||||
data.append({ |
||||
'id': bgd.id, |
||||
'name': bgd.name, |
||||
'description': bgd.description, |
||||
'createDate': bgd.createDate, |
||||
'icon': url_for('send_file', filename=bgd.icon) |
||||
'canClaim': bgd.canClaim |
||||
}) |
||||
return jsonify({ |
||||
'count': len(data), |
||||
'data': data |
||||
}) |
||||
|
||||
@badge_route.route('/myBadges') |
||||
@auth_required() |
||||
def my_badges(): |
||||
user_badges: list[UserBadge] = db.session.execute(select(UserBadge).where( |
||||
UserBadge.userID == g.current_user.id |
||||
)).scalars() |
||||
data: list = [] |
||||
for ub in user_badges: |
||||
bgd = ub.badge |
||||
data.append({ |
||||
'id': ub.id, |
||||
'badgeID': bgd.id, |
||||
'userID': ub.userID, |
||||
'name': bgd.name, |
||||
'description': bgd.description, |
||||
'createDate': bgd.createDate, |
||||
'icon': url_for('send_file', filename=bgd.icon) |
||||
'canClaim': bgd.canClaim, |
||||
'claimedDate': ub.claimedDate, |
||||
}) |
||||
return jsonify({ |
||||
'count': len(data), |
||||
'data': data |
||||
}) |
||||
@ -0,0 +1,226 @@ |
||||
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('chat', __name__) |
||||
|
||||
|
||||
@quiz.route('/generate') |
||||
@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.args['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_incomplete_quiz(): |
||||
try: |
||||
course_id: uuid.UUID = uuid.UUID(request.args['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') |
||||
@auth_required() |
||||
def get_incomplete_quiz(): |
||||
try: |
||||
answer: str = request.form['answer'].strip() |
||||
quiz_id: uuid.UUID = uuid.UUID(request.form['course_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.creatorUser.id == g.current_user.id: |
||||
Quiz.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 |
||||
Loading…
Reference in new issue