114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
from ..models import Dataset, Entry, User
|
|
from ..tools.data import validate_json
|
|
from ..tools.logs import write
|
|
from ..tools.test import evaluate_answers, generate_questions
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
from flask.helpers import abort, flash, url_for
|
|
from flask_login import login_required
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from datetime import datetime, timedelta
|
|
from json import loads
|
|
|
|
api = Blueprint(
|
|
name='api',
|
|
import_name=__name__
|
|
)
|
|
|
|
@api.route('/questions/', methods=['POST'])
|
|
def _fetch_questions():
|
|
id = request.get_json()['id']
|
|
try: entry = Entry.query.filter_by(id=id).first()
|
|
except SQLAlchemyError as exception:
|
|
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
|
|
return abort(500)
|
|
if not entry: return jsonify({'error': 'Invalid entry ID.'}), 400
|
|
test = entry.test
|
|
user_code = entry.user_code
|
|
time_limit = test.time_limit
|
|
time_adjustment = 0
|
|
if time_limit:
|
|
_time_limit = int(time_limit)
|
|
if user_code:
|
|
time_adjustment = test.adjustments[user_code]
|
|
_time_limit += time_adjustment
|
|
end_delta = timedelta(minutes=_time_limit)
|
|
end_time = datetime.now() + end_delta
|
|
else:
|
|
end_time = None
|
|
entry.start()
|
|
dataset = test.dataset
|
|
success, message = dataset.check_file()
|
|
if not success: return jsonify({'error': message}), 500
|
|
data_path = dataset.get_file()
|
|
with open(data_path, 'r') as data_file:
|
|
data = loads(data_file.read())
|
|
questions = generate_questions(data)
|
|
return jsonify({
|
|
'time_limit': end_time,
|
|
'questions': questions,
|
|
'start_time': entry.start_time,
|
|
'time_adjustment': time_adjustment
|
|
}), 200
|
|
|
|
@api.route('/submit/', methods=['POST'])
|
|
def _submit_quiz():
|
|
id = request.get_json()['id']
|
|
answers = request.get_json()['answers']
|
|
try: entry = Entry.query.filter_by(id=id).first()
|
|
except SQLAlchemyError as exception:
|
|
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
|
|
return abort(500)
|
|
if not entry: return jsonify({'error': 'Unrecognised Entry.'}), 400
|
|
test = entry.test
|
|
dataset = test.dataset
|
|
success, message = dataset.check_file()
|
|
if not success: return jsonify({'error': message}), 500
|
|
data_path = dataset.get_file()
|
|
with open(data_path, 'r') as data_file:
|
|
data = loads(data_file.read())
|
|
result = evaluate_answers(answers=answers, key=data)
|
|
entry.complete(answers=answers, result=result)
|
|
return jsonify({
|
|
'success': 'Your submission has been processed. Redirecting you to receive your results.',
|
|
'id': id
|
|
}), 200
|
|
|
|
@api.route('/editor/', methods=['POST'])
|
|
@login_required
|
|
def _editor(id:str=None):
|
|
request_data = request.get_json()
|
|
id = request_data['id']
|
|
try:
|
|
dataset = Dataset.query.filter_by(id=id).first()
|
|
user = User.query.filter_by(id=creator).first()
|
|
except SQLAlchemyError as exception:
|
|
write('system.log', f'Database error when processing request \'{request.url}\': {exception}')
|
|
return abort(500)
|
|
if not dataset: return jsonify({'error': 'Invalid request. Dataset not found.'}), 404
|
|
data_path = dataset.get_file()
|
|
if request_data['action'] == 'fetch':
|
|
with open(data_path, 'r') as data_file:
|
|
data = loads(data_file.read())
|
|
return jsonify({'success': 'Successfully downloaded dataset', 'data': data}), 200
|
|
default = request_data['default']
|
|
creator = request_data['creator']
|
|
name = request_data['name']
|
|
data = request_data['data']
|
|
if not validate_json(data): return jsonify({'error': 'The data you submitted was invalid.'}), 400
|
|
dataset.set_name(name)
|
|
dataset.creator = user
|
|
success, message = dataset.update(data=data, default=default)
|
|
if not success: return jsonify({'error': message}), 400
|
|
return jsonify({'success': message}), 200
|
|
|
|
@api.route('/editor/new/', methods=['POST'])
|
|
@login_required
|
|
def _editor_new():
|
|
new_dataset = Dataset()
|
|
new_dataset.set_name('New Dataset')
|
|
success, message = new_dataset.create(data=[], default=False)
|
|
if not success: return jsonify({'error':message}), 400
|
|
flash(message, 'success')
|
|
return jsonify({'success': message, 'redirect_to': url_for('editor._editor_console', id=new_dataset.id)}), 200 |