from ..forms.admin import CreateUser, DeleteUser, Login, Register, ResetPassword, UpdatePassword, UpdateUser, UploadData from ..models import Dataset, User from ..tools.auth import disable_if_logged_in, require_account_creation from ..tools.data import check_is_json, validate_json from flask import Blueprint, flash, jsonify, render_template, redirect, request, session from flask.helpers import url_for from flask_login import current_user, login_required import secrets admin = Blueprint( name='admin', import_name=__name__, template_folder='templates', static_folder='static' ) @admin.route('/') @admin.route('/home/') @admin.route('/dashboard/') @login_required def _home(): return 'Home Page' # TODO Dashboard @admin.route('/settings/') @login_required def _settings(): users = User.query.all() datasets = Dataset.query.all() return render_template('/admin/settings/index.html', users=users, datasets=datasets) @admin.route('/login/', methods=['GET','POST']) @disable_if_logged_in @require_account_creation def _login(): form = Login() if request.method == 'POST': if form.validate_on_submit(): users = User.query.all() user = None for _user in users: if _user.get_username() == request.form.get('username').lower(): user = _user break if user: if user.verify_password(request.form.get('password')): user.login(remember=request.form.get('remember')) return jsonify({'success': f'Successfully logged in.'}), 200 return jsonify({'error': f'The password you entered is incorrect.'}), 401 return jsonify({'error': f'The username you entered does not exist.'}), 401 errors = [*form.username.errors, *form.password.errors] return jsonify({ 'error': errors}), 400 if 'remembered_username' in session: form.username.data = session.pop('remembered_username') next = request.args.get('next') return render_template('/admin/auth/login.html', form=form, next=next) @admin.route('/logout/') @login_required def _logout(): current_user.logout() return redirect(url_for('admin._login')) @admin.route('/register/', methods=['GET','POST']) @disable_if_logged_in def _register(): from ..models.user import User form = Register() if request.method == 'POST': if form.validate_on_submit(): new_user = User() new_user.set_username = request.form.get('username').lower() new_user.set_email = request.form.get('email').lower() new_user.set_password = request.form.get('password').lower() success, message = new_user.register() if success: flash(message=f'{message} Please log in to continue.', category='success') session['remembered_username'] = request.form.get('username').lower() return jsonify({'success': message}), 200 flash(message=message, category='error') return jsonify({'error': message}), 401 errors = [*form.username.errors, *form.email.errors, *form.password.errors, *form.password_reenter.errors] return jsonify({ 'error': errors}), 400 return render_template('admin/auth/register.html') @admin.route('/reset/') def _reset(): form = ResetPassword() if request.method == 'POST': if form.validate_on_submit(): user = None users = User.query.all() for _user in users: if _user.get_username() == request.form.get('username'): user = _user break if not user: return jsonify({'error': 'The user account does not exist.'}), 400 if not user.get_email() == request.form.get('email'): return jsonify({'error': 'The email address does not match the user account.'}), 400 return user.reset_password() errors = [*form.username.errors, *form.email.errors] return jsonify({ 'error': errors}), 400 token = request.args.get('token') if token: user = User.query.filter_by(reset_token=token).first() if not user: return redirect(url_for('admin._reset')) verification_token = user.verification_token user.clear_reset_tokens() if request.args.get('verification') == verification_token: form = UpdatePassword() return render_template('/auth/update_password.html', form=form, user=user.id) flash('The verification of your password reset request failed and the token has been invalidated. Please make a new reset password request.', 'error') return render_template('/admin/auth/reset.html', form=form) @admin.route('/update_password/', methods=['POST']) def _update_password(): form = UpdatePassword() if form.validate_on_submit(): user = request.form.get('user') user = User.query.filter_by(id=user).first() user.update(password=request.form.get('password')) session['remembered_username'] = user.get_username() flash('Your password has been reset.', 'success') return jsonify({'success':'Your password has been reset'}), 200 errors = [*form.password.errors, *form.password_reenter.errors] return jsonify({ 'error': errors}), 401 @admin.route('/settings/users/', methods=['GET', 'POST']) @login_required def _users(): form = CreateUser() users = User.query.all() if request.method == 'POST': if form.validate_on_submit(): password = request.form.get('password') new_user = User() new_user.set_username = request.form.get('username').lower() new_user.set_password = secrets.token_hex(12) if not password else password new_user.set_email = request.form.get('email') success, message = new_user.register(notify=request.form.get('notify')) if success: return jsonify({'success': message}), 200 return jsonify({'error': message}), 401 errors = [*form.username.errors, *form.email.errors, *form.password.errors] return jsonify({ 'error': errors}), 401 return render_template('/admin/settings/users.html', form = form, users = users) @admin.route('/settings/users/delete/', methods=['GET', 'POST']) @login_required def _delete_user(id:str): user = User.query.filter_by(id=id).first() form = DeleteUser() if request.method == 'POST': if not user: return jsonify({'error': 'User does not exist.'}), 400 if id == current_user.id: return jsonify({'error': 'Cannot delete your own account.'}), 400 if form.validate_on_submit(): password = request.form.get('password') if not current_user.verify_password(password): return jsonify({'error': 'The password you entered is incorrect.'}), 401 success, message = user.delete(notify=request.form.get('notify')) if success: return jsonify({'success': message}), 200 return jsonify({'error': message}), 400 errors = form.password.errors return jsonify({ 'error': errors}), 400 if id == current_user.id: flash('Cannot delete your own user account.', 'error') return redirect(url_for('admin._users')) if not user: flash('User not found.', 'error') return redirect(url_for('admin._users')) return render_template('/admin/settings/delete_user.html', form=form, id = id, user = user) @admin.route('/settings/users/update/', methods=['GET', 'POST']) @login_required def _update_user(id:str): user = User.query.filter_by(id=id).first() form = UpdateUser() if request.method == 'POST': if not user: return jsonify({'error': 'User does not exist.'}), 400 if form.validate_on_submit(): success, message = user.update( password = request.form.get('password'), email = request.form.get('email'), notify = request.form.get('notify') ) if success: return jsonify({'success': message}), 200 return jsonify({'error': message}), 400 errors = [*form.confirm_password.errors, *form.email.errors, *form.password.errors, *form.password_reenter.errors] return jsonify({ 'error': errors}), 400 if not user: flash('User not found.', 'error') return redirect(url_for('admin._users')) return render_template('/admin/settings/delete_user.html', form=form, id = id, user = user) @admin.route('/settings/questions/', methods=['GET', 'POST']) @login_required def _quesitons(): form = UploadData() if request.method == 'POST': if form.validate_on_submit(): upload = form.data_file.data if not check_is_json(upload): return jsonify({'error': 'Invalid file. Please upload a JSON file.'}), 400 if not validate_json(upload): return jsonify({'error': 'The data in the file is invalid.'}), 400 # TODO Perhaps make a more complex validation script new_dataset = Dataset() success, message = new_dataset.create( upload = upload, default = request.form.get('default') ) if success: return jsonify({'success': message}), 200 return jsonify({'error': message}), 400 errors = form.data_file.errors return jsonify({ 'error': errors}), 400 data = Dataset.query.all() return render_template('/admin/settings/questions.html', data=data) @admin.route('/settings/questions/edit/', methods=['POST']) @login_required def _delete_questions(): id = request.get_json()['id'] action = request.get_json()['action'] dataset = Dataset.query.filter_by(id=id).first() if action == 'delete': success, message = dataset.delete() elif action == 'default': success, message = dataset.make_default() if success: return jsonify({'success': message}), 200 return jsonify({'error': message}), 400 # TODO Test views # TODO Result views