from bson import ObjectId from mongoengine.errors import ValidationError,NotUniqueError from flask import Blueprint, request, render_template, redirect, url_for, session, make_response,jsonify,flash,current_app from models.admin import Admin from models.slider import Slider from models.exam import Exam from werkzeug.security import check_password_hash from werkzeug.utils import secure_filename from utils.sms_service import send_sms from datetime import datetime, date, timedelta, time from models.sub_exam import SubExam import random import os from models.Coupon import Coupon from models.notification_model import Notification from bson.dbref import DBRef from models.old_papers_model import OldPapers from models.mock_test_model import MockTest, MockTestQuestion from models.ebooks_model import Ebook from models.course import Course from models.transaction_history import TransactionHistory from models.user import User from mongoengine.queryset.visitor import Q import calendar from io import BytesIO, StringIO import pandas as pd import json import uuid EBOOK_ALLOWED_EXTENSIONS = {'pdf'} MAX_EBOOK_SIZE = 10 * 1024 * 1024 # 10 MB def allowed_ebook(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in EBOOK_ALLOWED_EXTENSIONS OLD_PAPER_ALLOWED_EXTENSIONS = {'pdf'} MAX_OLD_PAPER_SIZE = 10 * 1024 * 1024 # 10 MB def allowed_old_paper(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in OLD_PAPER_ALLOWED_EXTENSIONS ALLOWED_EXTENSIONS = {'jpg', 'jpeg', 'png', 'webp'} # Set your maximum file size (500 KB) MAX_FILE_SIZE = 500 * 1024 # 500 KB def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS user_bp = Blueprint('user', __name__) # Login Function Start @user_bp.route('/user', methods=['GET', 'POST']) def user_login(): if 'user' in session: return redirect(url_for('user.user_dashboard')) username_error = None password_error = None general_error = None if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') remember_me = request.form.get('remember_me') == 'on' session.permanent = remember_me # Sets session lifetime accordingly if not username: username_error = "Enter username" if not password: password_error = "Enter password" if username_error or password_error: return render_template('user/login.html', username_error=username_error, password_error=password_error, request=request) user = Admin.objects(username=username).first() if not user: general_error = "Username does not exist" return render_template('user/login.html', general_error=general_error) if not check_password_hash(user.password, password): general_error = "Password is incorrect" return render_template('user/login.html', general_error=general_error) if user.status != 1: general_error = "Your account is deactivated" return render_template('user/login.html', general_error=general_error) # Save OTP and prepare response otp = str(random.randint(10000, 99999)) session['user_temp_id'] = str(user.id) session['login_otp'] = otp user.update(set__login_otp=otp, set__otp_created_at=datetime.utcnow()) user.save() sms_response = send_sms(user.phone, otp) if not sms_response['status']: return render_template('user/login.html', general_error='Failed to send OTP due to server error') resp = make_response(redirect(url_for('user.user_otp_verification'))) # Handle remember me cookie if remember_me: resp.set_cookie('remembered_username', username, max_age=60*60*24*30) # 30 days else: resp.set_cookie('remembered_username', '', expires=0) return resp # For GET requests remembered_username = request.cookies.get('remembered_username', '') return render_template('user/login.html', remembered_username=remembered_username) # Login Function End # Alias: explicit /user/login path @user_bp.route('/user/login', methods=['GET', 'POST']) def user_login_alias(): return user_login() # Otp Verification After Login Function Start @user_bp.route('/user/otp-verification', methods=['GET', 'POST']) def user_otp_verification(): # Check if the user is already logged in if 'user' in session: return redirect(url_for('user.user_dashboard')) # Redirect to the dashboard if already logged in temp_user_id = session.get('user_temp_id') if not temp_user_id: return redirect(url_for('user.user_login')) user = Admin.objects(id=temp_user_id).first() if not user: return redirect(url_for('user.user_login')) # Initialize error variables otp_error = None general_error = None if request.method == 'POST': input_otp = request.form.get('otp') # Check if OTP is empty if not input_otp: otp_error = "Enter OTP" return render_template('user/otp-verification.html', otp_error=otp_error) # Check if OTP has expired (1 minute expiration time) if user.otp_created_at and datetime.utcnow() > user.otp_created_at + timedelta(minutes=3): general_error = "OTP expired try to login again" return render_template('user/otp-verification.html', general_error=general_error) # Validate OTP if input_otp != user.login_otp: general_error = "Invalid OTP" return render_template('user/otp-verification.html', general_error=general_error) # OTP is correct, complete the login process session.pop('user_temp_id', None) # Clear temporary session session['user'] = str(user.id) # Store user ID in session for a permanent login # Reset OTP after successful login user.login_otp = None user.save() return redirect(url_for('user.user_dashboard')) # Redirect to the user dashboard return render_template('user/otp-verification.html', otp_error=otp_error, general_error=general_error) # Otp Verification After Login Function End # Resend Otp After Login Function Start @user_bp.route('/user/resend-otp', methods=['POST']) def resend_otp_ajax(): temp_user_id = session.get('user_temp_id') if not temp_user_id: return jsonify({'status': False, 'message': 'Session expired try to login again'}), 401 user = Admin.objects(id=temp_user_id).first() if not user: return jsonify({'status': False, 'message': 'User not found. Please login again'}), 401 # Generate new OTP otp = str(random.randint(10000, 99999)) session['login_otp'] = otp # Save OTP and timestamp user.update(set__login_otp=otp, set__otp_created_at=datetime.utcnow()) user.save() # Send OTP SMS sms_response = send_sms(user.phone, otp) if not sms_response.get('status'): return jsonify({'status': False, 'message': 'Failed to send OTP. Try again later'}) return jsonify({'status': True, 'message': 'OTP resent successfully'}) # Resend Otp After Login Function End # User DashBoard @user_bp.route('/user/dashboard') def user_dashboard(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/index.html') # User DashBoard # Alias: /user/index -> dashboard @user_bp.route('/user/index') def user_index_alias(): return redirect(url_for('user.user_dashboard')) # View Users List Function start @user_bp.route('/user/view-users', methods=['GET']) def view_users(): if 'user' not in session: return redirect(url_for('user.user_login')) try: users = User.objects.order_by('-created_date') except Exception: current_app.logger.exception("Failed to load users") users = [] return render_template('user/view-users.html', users=users) # View Users List Function start # View User Uploaded Files List Function start @user_bp.route('/user/view-uploaded-files', methods=['GET']) def view_uploaded_files(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/view-uploaded-files.html') # View User Uploaded Files List Function start # View User Quiz Played History List Function start @user_bp.route('/user/view-quiz-history', methods=['GET']) def view_quiz_history(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/view-quiz-history.html') # View User Quiz Played History List Function start # View User Quiz Played Detail History List Function start @user_bp.route('/user/quiz-detail-history', methods=['GET']) def quiz_detail_history(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/quiz-detail-history.html') # View User Quiz Played Detail History List Function start # Add Slider Function start @user_bp.route('/user/add-slider', methods=['GET', 'POST']) def add_slider(): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None if request.method == 'POST': slider_image = request.files.get('slider_image') slider_position = request.form.get('sliderposition') status = request.form.get('status') # === IMAGE VALIDATION === if not slider_image or slider_image.filename == '': errors['slider_image'] = "Select slider image" elif not allowed_file(slider_image.filename): errors['slider_image'] = "Only JPG, JPEG, PNG, and WEBP images are allowed" else: slider_image.seek(0, os.SEEK_END) file_size = slider_image.tell() slider_image.seek(0) if file_size > MAX_FILE_SIZE: errors['slider_image'] = "Image must be less than 500 KB" # === POSITION VALIDATION === if not slider_position: errors['sliderposition'] = "Enter slider position" else: try: slider_position_int = int(slider_position) if Slider.objects(slide_position=slider_position_int).first(): errors['sliderposition'] = "Slider position must be unique" except ValueError: errors['sliderposition'] = "Slider position must be a numeric value" # === STATUS VALIDATION === if not status: errors['status'] = "Select status" # === RETURN ON VALIDATION ERRORS === if errors: return render_template( 'user/add-slider.html', errors=errors, request=request ) # === TRY BLOCK FOR IMAGE SAVE + DB === try: filename = secure_filename(slider_image.filename) upload_folder = os.path.join('static', 'uploads', 'sliders') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) slider_image.save(save_path) # Build full image URL base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path image_url = f"{base_url}{relative_url}" # Save to DB Slider(image=image_url, slide_position=slider_position_int, status=int(status)).save() except Exception as e: general_error = "Something went wrong" return render_template('user/add-slider.html', general_error=general_error, request=request) flash("Slider added successfully", "success") return redirect(url_for('user.add_slider')) # GET request return render_template('user/add-slider.html') # Add Slider Function end # Delete Slider Function Start @user_bp.route('/user/delete-slider/', methods=['POST']) def delete_slider(slider_id): if 'user' not in session: return jsonify({'success': False, 'message': 'Unauthorized access'}), 401 try: slider = Slider.objects(id=slider_id).first() if not slider: return jsonify({'success': False, 'message': 'Slider not found'}), 404 slider.delete() return jsonify({'success': True, 'message': 'Slider deleted successfully'}) except Exception as e: return jsonify({'success': False, 'message': 'Failed to delete slider'}), 500 # Delete Slider Function End # Edit Slider Function Start @user_bp.route('/user/edit-slider/', methods=['GET', 'POST']) def edit_slider(slider_id): if 'user' not in session: return redirect(url_for('user.user_login')) slider = Slider.objects(id=slider_id).first() if not slider: flash('Slider not found.', 'danger') return redirect(url_for('user.view_sliders')) errors = {} general_error = None if request.method == 'POST': slider_position = request.form.get('sliderposition') status = request.form.get('status') slider_image = request.files.get('slider_image') # POSITION VALIDATION if not slider_position: errors['sliderposition'] = "Enter slider position" else: try: slider_position_int = int(slider_position) existing_slider = Slider.objects(slide_position=slider_position_int).first() if existing_slider and existing_slider.id != slider.id: errors['sliderposition'] = "Slider position must be unique" except ValueError: errors['sliderposition'] = "Slider position must be numeric" # STATUS VALIDATION if not status: errors['status'] = "Select status" # IMAGE VALIDATION (optional on edit) if slider_image and slider_image.filename != '': if not allowed_file(slider_image.filename): errors['slider_image'] = "Only JPG, JPEG, PNG, and WEBP images are allowed" else: slider_image.seek(0, os.SEEK_END) file_size = slider_image.tell() slider_image.seek(0) if file_size > MAX_FILE_SIZE: errors['slider_image'] = "Image must be less than 500 KB" if errors: return render_template( 'user/edit-slider.html', errors=errors, slider=slider, request=request ) try: if slider_image and slider_image.filename != '': filename = secure_filename(slider_image.filename) upload_folder = os.path.join('static', 'uploads', 'sliders') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) slider_image.save(save_path) base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path image_url = f"{base_url}{relative_url}" slider.image = image_url slider.slide_position = slider_position_int slider.status = int(status) slider.save() except Exception as e: general_error = "Something went wrong" return render_template('user/edit-slider.html', general_error=general_error, slider=slider) flash("Slider updated successfully", "success") return render_template('user/edit-slider.html', slider=slider) # GET request prefill form return render_template('user/edit-slider.html', slider=slider) # Edit Slider Function End # View Sliders List Function start @user_bp.route('/user/view-sliders') def view_sliders(): if 'user' not in session: return redirect(url_for('user.user_login')) sliders = Slider.objects().order_by('slide_position') # Fetch all sliders ordered by position return render_template('user/view-sliders.html', sliders=sliders) # View Sliders List Function start # Add Exam Function start @user_bp.route('/user/add-exam', methods=['GET', 'POST']) def add_exam(): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None if request.method == 'POST': title = request.form.get('title') description = request.form.get('description') exam_position = request.form.get('exam_position') exam_duration = request.form.get('exam_duration') status = request.form.get('status') icon = request.files.get('icon') icon_url = None # === VALIDATIONS === if not title: errors['title'] = "Enter exam title" # Exam Position validation - Required & Unique if not exam_position: errors['exam_position'] = "Enter exam position" else: try: exam_position_int = int(exam_position) # Check if position already exists in DB if Exam.objects(exam_position=exam_position_int).first(): errors['exam_position'] = "Exam position must be unique" except ValueError: errors['exam_position'] = "Exam position must be numeric" if exam_duration: try: exam_duration_int = int(exam_duration) if exam_duration_int <= 0: errors['exam_duration'] = "Exam duration must be a positive number" except ValueError: errors['exam_duration'] = "Exam duration must be numeric" else: exam_duration_int = None if not status: errors['status'] = "Select status" # === ICON VALIDATION === if icon and icon.filename != '': if not allowed_file(icon.filename): errors['icon'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: icon.seek(0, os.SEEK_END) file_size = icon.tell() icon.seek(0) if file_size > MAX_FILE_SIZE: errors['icon'] = "Exam icon must be less than 500 KB" # === RETURN IF ERRORS === if errors: return render_template( 'user/add-exam.html', errors=errors, request=request ) # === ICON SAVE === if icon and icon.filename != '': try: filename = secure_filename(icon.filename) upload_folder = os.path.join('static', 'uploads', 'exam_icons') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) icon.save(save_path) base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path icon_url = f"{base_url}{relative_url}" except Exception as e: general_error = "Exam icon upload failed" return render_template('user/add-exam.html', general_error=general_error, request=request) # === SAVE TO DB === try: Exam( exam_title=title, exam_description=description, exam_position=exam_position_int, #exam_duration=exam_duration_int, status=int(status), icon=icon_url ).save() except Exception as e: general_error = "Something went wrong" return render_template('user/add-exam.html', general_error=general_error, request=request) flash("Exam added successfully", "success") return redirect(url_for('user.add_exam')) return render_template('user/add-exam.html') # Add Exam Function start # View Exams List Function start @user_bp.route('/user/view-exams', methods=['GET']) def exams_list(): if 'user' not in session: return redirect(url_for('user.user_login')) # Fetch exams from the database exams = Exam.objects().order_by('exam_position') return render_template('user/view-exams.html', exams=exams) # View Exams List Function start # Delete Exam Function Start @user_bp.route('/user/delete-exam/', methods=['POST']) def delete_exam(exam_id): if 'user' not in session: return jsonify({'success': False, 'message': 'Unauthorized access'}), 401 try: # Find the exam by its ID exam = Exam.objects(id=exam_id).first() if not exam: return jsonify({'success': False, 'message': 'Exam not found'}), 404 # Delete the exam exam.delete() return jsonify({'success': True, 'message': 'Exam deleted successfully'}) except Exception as e: return jsonify({'success': False, 'message': 'Failed to delete exam'}), 500 # Delete Exam Function Start # Edit Exam Function Start @user_bp.route('/user/edit-exam/', methods=['GET', 'POST']) def edit_exam(exam_id): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None exam = Exam.objects(id=exam_id).first() if not exam: flash('Exam not found.', 'danger') return redirect(url_for('user.exams_list')) # Redirect if exam not found if request.method == 'POST': title = request.form.get('title') description = request.form.get('description') exam_position = request.form.get('exam_position') exam_duration = request.form.get('exam_duration') status = request.form.get('status') icon = request.files.get('icon') icon_url = exam.icon # Retain the existing icon if no new one is uploaded # === VALIDATIONS === if not title: errors['title'] = "Enter exam title" if not exam_position: errors['exam_position'] = "Enter exam position" else: try: exam_position_int = int(exam_position) existing_exam = Exam.objects(exam_position=exam_position_int).first() if existing_exam and existing_exam.id != exam.id: errors['sliderposition'] = "Exam position must be unique" except ValueError: errors['sliderposition'] = "Exam position must be numeric" if not status: errors['status'] = "Select status" # === ICON VALIDATION === if icon and icon.filename != '': if not allowed_file(icon.filename): errors['icon'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: icon.seek(0, os.SEEK_END) file_size = icon.tell() icon.seek(0) if file_size > MAX_FILE_SIZE: errors['icon'] = "Exam icon must be less than 500 KB" # === RETURN IF ERRORS === if errors: return render_template( 'user/edit-exam.html', errors=errors, exam=exam, general_error=general_error, request=request ) # === ICON SAVE === if icon and icon.filename != '': try: filename = secure_filename(icon.filename) upload_folder = os.path.join('static', 'uploads', 'exam_icons') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) icon.save(save_path) base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path icon_url = f"{base_url}{relative_url}" except Exception as e: general_error = "Exam icon upload failed" return render_template('user/edit-exam.html', general_error=general_error, exam=exam, request=request) # === UPDATE EXAM IN DB === try: exam.update( exam_title=title, exam_description=description, exam_position=int(exam_position_int), #exam_duration=exam_duration, status=int(status), icon=icon_url ) flash("Exam updated successfully", "success") return render_template('user/edit-exam.html', exam=exam) except Exception as e: general_error = "Something went wrong" return render_template('user/edit-exam.html', general_error=general_error, exam=exam, request=request) return render_template('user/edit-exam.html', exam=exam) # Edit Exam Function End @user_bp.route('/user/add-subexam', methods=['GET', 'POST']) def add_subexam(): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None exams = Exam.objects(status=1).order_by('exam_title') if request.method == 'POST': exam_id = request.form.get('exam_id') title = request.form.get('title') description = request.form.get('description') subexam_position = request.form.get('subexam_position') status = request.form.get('status') icon = request.files.get('icon') icon_url = None parent_exam = None if not exam_id: errors['exam_id'] = "Select exam" else: parent_exam = Exam.objects(id=exam_id).first() if not parent_exam: errors['exam_id'] = "Invalid exam selected" if not title or not title.strip(): errors['title'] = "Enter subexam title" if not subexam_position: errors['subexam_position'] = "Enter subexam position" else: try: position_int = int(subexam_position) except ValueError: errors['subexam_position'] = "Subexam position must be numeric" if not status: errors['status'] = "Select status" if icon and icon.filename: if not allowed_file(icon.filename): errors['icon'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: icon.seek(0, os.SEEK_END) if icon.tell() > MAX_FILE_SIZE: errors['icon'] = "Icon must be less than 500 KB" icon.seek(0) if errors: return render_template('user/add-subexam.html', errors=errors, general_error=general_error, request=request, exams=exams) if icon and icon.filename: try: filename = secure_filename(icon.filename) upload_folder = os.path.join('static', 'uploads', 'subexam_icons') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) icon.save(save_path) # Get the full URL for the icon base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path icon_url = f"{base_url}{relative_url}" except Exception: return render_template('user/add-subexam.html', general_error="Subexam icon upload failed", request=request, exams=exams) try: SubExam( exam_id=parent_exam, icon=icon_url, sub_exam_title=title.strip(), sub_exam_description=(description or '').strip() or None, status=int(status), position=int(subexam_position) ).save() except Exception: return render_template('user/add-subexam.html', general_error="Something went wrong while saving", request=request, exams=exams) flash("Subexam added successfully", "success") return redirect(url_for('user.add_subexam')) return render_template('user/add-subexam.html', exams=exams) # List Subexams @user_bp.route('/user/view-subexams', methods=['GET']) def subexams_list(): if 'user' not in session: return redirect(url_for('user.user_login')) subexams = SubExam.objects().order_by('position') return render_template('user/view-subexams.html', subexams=subexams) # Edit Subexam @user_bp.route('/user/edit-subexam/', methods=['GET', 'POST']) def edit_subexam(subexam_id): if 'user' not in session: return redirect(url_for('user.user_login')) subexam = SubExam.objects(id=subexam_id).first() if not subexam: flash("Subexam not found", "danger") return redirect(url_for('user.subexams_list')) exams = Exam.objects(status=1).order_by('exam_title') errors, general_error = {}, None if request.method == 'POST': exam_id = request.form.get('exam_id') title = request.form.get('title') description = request.form.get('description') subexam_position = request.form.get('subexam_position') status = request.form.get('status') icon = request.files.get('icon') parent_exam = None if not exam_id: errors['exam_id'] = "Select exam" else: parent_exam = Exam.objects(id=exam_id).first() if not parent_exam: errors['exam_id'] = "Invalid exam selected" if not title or not title.strip(): errors['title'] = "Enter subexam title" try: position_int = int(subexam_position) except (TypeError, ValueError): errors['subexam_position'] = "Subexam position must be numeric" if not status: errors['status'] = "Select status" if icon and icon.filename: if not allowed_file(icon.filename): errors['icon'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: icon.seek(0, os.SEEK_END) if icon.tell() > MAX_FILE_SIZE: errors['icon'] = "Icon must be less than 500 KB" icon.seek(0) if errors: return render_template('user/edit-subexam.html', errors=errors, general_error=general_error, request=request, subexam=subexam, exams=exams) icon_url = subexam.icon if icon and icon.filename: try: filename = secure_filename(icon.filename) upload_folder = os.path.join('static', 'uploads', 'subexam_icons') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) icon.save(save_path) # Get the full URL for the icon base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path icon_url = f"{base_url}{relative_url}" except Exception: return render_template('user/edit-subexam.html', general_error="Subexam icon upload failed", request=request, subexam=subexam, exams=exams) subexam.update( set__exam_id=parent_exam, set__icon=icon_url, set__sub_exam_title=title.strip(), set__sub_exam_description=(description or '').strip() or None, set__status=int(status), set__position=position_int ) subexam.reload() flash("Subexam updated successfully", "success") return redirect(url_for('user.edit_subexam', subexam_id=subexam.id)) return render_template('user/edit-subexam.html', subexam=subexam, exams=exams) # Delete Subexam (AJAX) @user_bp.route('/user/delete-subexam/', methods=['POST']) def delete_subexam(subexam_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 subexam = SubExam.objects(id=subexam_id).first() if not subexam: return jsonify(success=False, message="Subexam not found"), 404 try: subexam.delete() return jsonify(success=True, message="Subexam deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 # ==================== COUPON MANAGEMENT ROUTES ==================== @user_bp.route("/user/add-coupon", methods=["GET", "POST"]) def add_coupon_html_direct(): print("HANDLER CALLED:", request.method) if "user" not in session: return redirect(url_for("user.user_login")) errors=dict() if request.method == "POST": data = request.form.to_dict() try: discount_value = int(data.get("discount_value", 0)) if discount_value < 1 or discount_value > 100: errors["discount_value"] = "Discount value must be between 1 and 100." except ValueError: errors["discount_value"] = "Discount value must be a number." coupon = Coupon( coupon_title=data.get("title"), coupon_code=data.get("code"), coupon_type=data.get("discount_type") if data.get("discount_type")=="flat" else "percentage", coupon_value=int(data.get("discount_value")) if data.get("discount_value") else None, coupon_start_date=data.get("start_date"), coupon_end_date=data.get("end_date"), status=int(data.get("status", 1)) ) try: coupon.save() return redirect(url_for("user.view_coupon_html_direct")) except NotUniqueError: errors["code"] = "Coupon code already exists." except ValidationError as ve: errors["_global"] = f"Validation failed: {ve}" except Exception as e: current_app.logger.exception("Failed to save coupon: %s", e) errors["_global"] = "Server error while creating coupon." return render_template("user/add-coupon.html", errors=errors, old=data) # return redirect(url_for("user_bp.add_coupon_html_direct")) return render_template("user/add-coupon.html",errors=errors) @user_bp.route("/user/view-coupon", methods=["GET", "POST"]) def view_coupon_html_direct(): print("VIew Coupon:", request.method) if "user" not in session: return redirect(url_for("user.user_login")) try: coupons = Coupon.objects.order_by("-created_date") # newest first except Exception as e: current_app.logger.exception("Failed to fetch coupons: %s", e) coupons = [] coupons_data = [c.to_json() for c in coupons] return render_template("user/view-coupons.html",coupons=coupons_data) @user_bp.route("/user/delete-coupon/", methods=["POST"]) def delete_coupon(coupon_id): print("delete") if "user" not in session: # If AJAX, return JSON 401; otherwise redirect to login page if request.headers.get("X-Requested-With") == "XMLHttpRequest" or request.is_json: return jsonify(success=False, message="Unauthorized"), 401 return redirect(url_for("user.user_login")) try: c = Coupon.objects(id=coupon_id).first() if not c: message = "Coupon not found." if request.headers.get("X-Requested-With") == "XMLHttpRequest" or request.is_json: return jsonify(success=False, message=message), 404 flash(message, "danger") else: c.delete() message = "Coupon deleted." if request.headers.get("X-Requested-With") == "XMLHttpRequest" or request.is_json: return jsonify(success=True, message=message), 200 flash(message, "success") except Exception as e: current_app.logger.exception("Failed to delete coupon: %s", e) message = "Error deleting coupon." if request.headers.get("X-Requested-With") == "XMLHttpRequest" or request.is_json: return jsonify(success=False, message=message), 500 flash(message, "danger") # Non-AJAX fallback: redirect to listing page return redirect(url_for("user_bp.view_coupon_html_direct")) @user_bp.route("/user/edit-coupon/", methods=["GET", "POST"]) def edit_coupon(coupon_id): if "user" not in session: return redirect(url_for("user.user_login")) coupon = Coupon.objects(id=coupon_id).first() print("form:",request.form) print("status data:",request.form.get("status"),coupon.status) if not coupon: flash("Coupon not found.", "danger") return redirect(url_for("user_bp.view_coupon_html_direct")) print("form data of status:",type(request.form.get("status"))) errors={} if request.method == "POST": try: value = int(request.form.get("coupon_value", 0)) if value < 1 or value > 100: errors["coupon_value"] = "Discount value must be between 1 and 100." else: coupon.coupon_value = value except ValueError: errors["coupon_value"] = "Discount value must be a number." coupon.coupon_title = request.form.get("coupon_title") or coupon.coupon_title coupon.coupon_code = request.form.get("coupon_code") or coupon.coupon_code coupon.coupon_type = request.form.get("coupon_type") or coupon.coupon_type coupon.coupon_value = request.form.get("coupon_value") or coupon.coupon_value start_date = request.form.get("coupon_start_date") end_date = request.form.get("coupon_end_date") if start_date: # Only replace if user selected a new date coupon.coupon_start_date = datetime.combine( datetime.strptime(start_date, "%Y-%m-%d").date(), coupon.coupon_start_date.time() if coupon.coupon_start_date else datetime.min.time() ) if end_date: coupon.coupon_end_date = datetime.combine( datetime.strptime(end_date, "%Y-%m-%d").date(), coupon.coupon_end_date.time() if coupon.coupon_end_date else datetime.min.time() ) coupon.status = int(request.form.get("status")) if errors: return render_template("user/edit-coupon.html", coupon=coupon.to_json(), errors=errors) print("inside post:",coupon.status) coupon.save() flash("Coupon updated.", "success") return redirect(url_for("user.view_coupon_html_direct")) return render_template("user/edit-coupon.html", coupon=coupon.to_json()) # ==================== COURSE MANAGEMENT ROUTES ==================== # Add Course Function start @user_bp.route('/user/add-course', methods=['GET', 'POST']) def add_course(): errors = {} form_data = {} image_url="" if request.method == 'POST': form_data = request.form.to_dict() print("form data:",form_data) exam_id = request.form.get('exam_id') or '' sub_exam_id = request.form.get('sub_exam_id') or '' course_title = request.form.get('title') or '' fees=request.form.get('fees') or '' status=request.form.get('status') or '' position=request.form.get('position') or '' discount_fees=request.form.get('discounted_fees') or '' duration=request.form.get('duration') or '' img_file=request.files.get("course_image") language=request.form.get('language') # validate required if not exam_id: errors['exam_id'] = 'Please select an exam.' if not sub_exam_id: errors['sub_exam_id'] = 'Please select a sub exam.' if not fees: errors['fees'] = 'Course title is required.' if not status: errors['status'] = 'Please select a status' if not position: errors['position'] = 'Position is required.' if int(discount_fees)>=int(fees): errors['discounted_fees'] = 'Discount fees must be less than fees.' if not duration: errors['duration'] = 'Duration is required.' exam_obj = None sub_obj = None if exam_id: try: exam_obj = Exam.objects.get(id=exam_id) except DoesNotExist: errors['exam_id'] = 'Selected exam not found.' if sub_exam_id: try: sub_obj = SubExam.objects.get(id=sub_exam_id) except DoesNotExist: errors['sub_exam_id'] = 'Selected sub exam not found.' print("error:",errors) if img_file and img_file.filename != '': if not allowed_file(img_file.filename): errors['image'] = "Only JPG, JPEG, PNG, GIF, WEBP files are allowed." else: # check file size try: img_file.stream.seek(0, os.SEEK_END) file_size = img_file.stream.tell() img_file.stream.seek(0) except Exception: file_size = None if file_size is not None and file_size > MAX_FILE_SIZE: errors['image'] = "Image must be less than 500 KB." if not errors: # SAVE IMAGE (if provided) if img_file and img_file.filename != '': try: filename = secure_filename(img_file.filename) upload_folder = os.path.join(current_app.static_folder or 'static', 'uploads', 'course_image') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) img_file.save(save_path) # build a URL relative to the host base_url = request.host_url.rstrip('/') # ensure path uses forward slashes relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find('static/') if static_index != -1: relative_url = '/' + relative_path[static_index:] else: # fallback if static not found relative_url = '/' + os.path.relpath(save_path, start=os.getcwd()).replace(os.sep, '/') image_url = f"{base_url}{relative_url}" except Exception: current_app.logger.exception("Course image upload failed") errors['image'] = "Image upload failed." # if not image_url: # errors['image']="Image is required" if not errors: try: course = Course( exam_id = exam_obj, sub_exam_id = sub_obj, course_title = course_title, course_fee=fees, status=status, position=position, discounted_fee=discount_fees, duration=duration, course_image=image_url, language=language ) course.save() flash('Course added successfully', 'success') return redirect(url_for('user.view_courses')) except ValidationError as e: errors['general'] = str(e) # GET or POST with errors -> fetch lists for the form exams = Exam.objects.filter(status=1).order_by('name') # or .order_by('title') subexams = [] # If form_data has exam_id (e.g. after failed POST) pre-load its subexams chosen_exam_id = form_data.get('exam_id') if chosen_exam_id: subexams = SubExam.objects.filter(exam_id=chosen_exam_id, status=1).order_by('name') else: # optional: preload subexams for first exam for better UX if exams: subexams = SubExam.objects.filter(exam_id=exams[0].id, status=1).order_by('name') return render_template('user/add-course.html', exams=exams, subexams=subexams, errors=errors, form_data=form_data) @user_bp.route('/user/view-courses', methods=['GET', 'POST']) def view_courses(): try: courses = Course.objects.order_by('-created_date').select_related() print("courses:",courses) except Exception: # fallback if select_related not available for your setup courses = Course.objects.order_by('-created_date') return render_template('user/view-courses.html', courses=courses) @user_bp.route('/user/edit-course/', methods=['GET', 'POST']) def edit_course(course_id): # image rules (same as add_notification/add_course) ALLOWED_EXT = {"png", "jpg", "jpeg", "gif", "webp"} MAX_FILE_SIZE = 500 * 1024 # 500 KB def allowed_file(fname): return "." in fname and fname.rsplit(".", 1)[1].lower() in ALLOWED_EXT if "user" not in session: return redirect(url_for("user.user_login")) # load course try: course = Course.objects.get(id=course_id) except DoesNotExist: flash("Course not found.", "danger") return redirect(url_for('user.view_courses')) except Exception: current_app.logger.exception("Error fetching course") flash("Unable to load course.", "danger") return redirect(url_for('user.view_courses')) errors = {} general_error = None form_data = {} # pre-fill form_data from existing course for GET form_data = { 'exam_id': getattr(course, 'exam_id').id if getattr(course, 'exam_id', None) else '', 'sub_exam_id': getattr(course, 'sub_exam_id').id if getattr(course, 'sub_exam_id', None) else '', 'title': getattr(course, 'course_title', '') or getattr(course, 'title', ''), 'fees': getattr(course, 'course_fee', '') or getattr(course, 'fees', ''), 'status': str(getattr(course, 'status', '1')), 'position': getattr(course, 'position', '') or '', 'discounted_fees': getattr(course, 'discounted_fee', '') or '', 'duration': getattr(course, 'duration', '') or '', 'language': getattr(course, 'language', '') or '', 'course_image': getattr(course, 'course_image', None) or getattr(course, 'image', None) or getattr(course, 'image_url', None) } if request.method == "POST": # gather posted form + files form_data = request.form.to_dict() title = (request.form.get('title') or '').strip() exam_id = request.form.get('exam_id') or '' sub_exam_id = request.form.get('sub_exam_id') or '' fees = request.form.get('fees') or '' status = request.form.get('status') or '' position = request.form.get('position') or '' discount_fees = request.form.get('discounted_fees') or '' duration = request.form.get('duration') or '' language = request.form.get('language') or '' img_file = request.files.get('course_image') # validations (same as add) if not exam_id: errors['exam_id'] = 'Please select an exam.' if not sub_exam_id: errors['sub_exam_id'] = 'Please select a sub exam.' if not title: errors['title'] = 'Course title is required.' if not fees: errors['fees'] = 'Course fees is required.' if not status: errors['status'] = 'Please select a status.' if not position: errors['position'] = 'Position is required.' if int(discount_fees)>=int(fees): errors['discounted_fees'] = 'Discount fees must be less than fees.' if not duration: errors['duration'] = 'Duration is required.' # check referenced exam/sub exist exam_obj = None sub_obj = None if exam_id: try: exam_obj = Exam.objects.get(id=exam_id) except DoesNotExist: errors['exam_id'] = 'Selected exam not found.' if sub_exam_id: try: sub_obj = SubExam.objects.get(id=sub_exam_id) except DoesNotExist: errors['sub_exam_id'] = 'Selected sub exam not found.' # image validation if img_file and img_file.filename != '': if not allowed_file(img_file.filename): errors['image'] = "Only JPG, JPEG, PNG, GIF, WEBP files are allowed." else: try: img_file.stream.seek(0, os.SEEK_END) file_size = img_file.stream.tell() img_file.stream.seek(0) except Exception: file_size = None if file_size is not None and file_size > MAX_FILE_SIZE: errors['image'] = "Image must be less than 500 KB." if not errors: # update fields try: course.course_title = title course.course_fee = fees course.language=language try: course.status = int(status) except (TypeError, ValueError): course.status = status course.position = position course.discounted_fee = discount_fees course.duration = duration if exam_obj: course.exam_id = exam_obj if sub_obj: course.sub_exam_id = sub_obj # save image if uploaded if img_file and img_file.filename != '': try: filename = secure_filename(img_file.filename) upload_folder = os.path.join(current_app.static_folder or 'static', 'uploads', 'course_image') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) img_file.save(save_path) # store a static path (so templates can use it directly) static_filename = f"uploads/course_image/{filename}" # store as relative static path '/static/...' or as just path depending on your preference relative_url = url_for('static', filename=static_filename) course.course_image = relative_url except Exception: current_app.logger.exception("Course image upload failed") errors['image'] = "Image upload failed." # save the course course.save() flash('Course updated successfully', 'success') return redirect(url_for('user.view_courses')) except ValidationError as e: current_app.logger.exception("Validation error saving course") errors['general'] = str(e) except Exception: current_app.logger.exception("Failed to update course") errors['general'] = "Failed to update course." # GET or POST with errors -> repopulate exams/subexams for form exams = Exam.objects.filter(status=1).order_by('name') subexams = [] chosen_exam_id = form_data.get('exam_id') if chosen_exam_id: subexams = SubExam.objects.filter(exam_id=chosen_exam_id, status=1).order_by('name') else: if exams: subexams = SubExam.objects.filter(exam_id=exams[0].id, status=1).order_by('name') return render_template('user/edit-course.html', exams=exams, subexams=subexams, errors=errors, form_data=form_data, course=course) @user_bp.route('/user/delete-course/', methods=['POST']) def delete_course(course_id): try: course = Course.objects.get(id=course_id) except DoesNotExist: return jsonify(success=False, message='Course not found'), 404 except Exception as e: return jsonify(success=False, message=str(e)), 500 try: course.delete() return jsonify(success=True, message='Course deleted successfully') except Exception as e: return jsonify(success=False, message=f'Could not delete course: {e}'), 500 # View Course Purchase History Function start @user_bp.route('/user/view-course-purchase-history', methods=['GET']) def view_course_purchase_history(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/view-course-purchase-history.html') # View Course Purchase History Function end # ==================== NOTIFICATION MANAGEMENT ROUTES ==================== @user_bp.route("/user/add-notification", methods=["GET", "POST"]) def add_notification(): # allow same image rules as add_exam ALLOWED_EXT = {"png", "jpg", "jpeg", "gif", "webp"} MAX_FILE_SIZE = 500 * 1024 # 500 KB def allowed_file(fname): return "." in fname and fname.rsplit(".", 1)[1].lower() in ALLOWED_EXT if "user" not in session: return redirect(url_for("user.user_login")) errors = {} general_error = None image_url = None if request.method == "POST": title = (request.form.get("title") or "").strip() description = (request.form.get("description") or "").strip() status = request.form.get("status") # support both possible input names: image_file or image (template uses image_file) img_file = request.files.get("image") # === VALIDATIONS === if not title: errors['title'] = "Title is required." if not status: errors['status'] = "Select status" # IMAGE VALIDATION if img_file and img_file.filename != '': if not allowed_file(img_file.filename): errors['image'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: # determine file size try: img_file.stream.seek(0, os.SEEK_END) file_size = img_file.stream.tell() img_file.stream.seek(0) except Exception: file_size = None if file_size is not None and file_size > MAX_FILE_SIZE: errors['image'] = "Image must be less than 500 KB" # If there are validation errors, re-render and show them if errors: return render_template( "user/add-notification.html", errors=errors, form_data=request.form ) # === SAVE IMAGE (if provided) === if img_file and img_file.filename != '': try: filename=secure_filename(img_file.filename) upload_folder = os.path.join('static', 'uploads', 'notification_image') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) img_file.save(save_path) base_url = request.host_url.replace("http://", "https://").rstrip('/') relative_path = save_path.replace(os.sep, '/') static_index = relative_path.find("static/") if static_index != -1: relative_url = '/' + relative_path[static_index:] else: relative_url = '/' + relative_path icon_url = f"{base_url}{relative_url}" except Exception: current_app.logger.exception("Notification image upload failed") general_error = "Image upload failed" return render_template("user/add-notification.html", general_error=general_error, form_data=request.form) # === SAVE NOTIFICATION === try: # status may need conversion to int try: status_int = int(status) except (TypeError, ValueError): status_int = 1 notification = Notification( title=title, description=description, image=icon_url if icon_url else None, status=status_int, created_date=datetime.utcnow() ) notification.save() flash("Notification added successfully.", "success") return redirect(url_for("user.view_notifications")) except Exception: current_app.logger.exception("Failed to save notification") general_error = "Failed to save notification" return render_template("user/add-notification.html", general_error=general_error, form_data=request.form) # GET return render_template("user/add-notification.html", errors=errors) # --- View Notifications --- @user_bp.route("/user/view-notifications") def view_notifications(): if "user" not in session: return redirect(url_for("user.user_login")) notifications = Notification.objects.order_by('-created_date') return render_template("user/view-notifications.html", notifications=notifications) # --- Edit Notification --- @user_bp.route("/user/edit-notification/", methods=["GET", "POST"]) def edit_notification(notification_id): if "user" not in session: return redirect(url_for("user.user_login")) notification = Notification.objects(id=notification_id).first() if not notification: flash("Notification not found.", "danger") return redirect(url_for("user.view_notifications")) errors = {} if request.method == "POST": title = (request.form.get("title") or "").strip() description = (request.form.get("description") or "").strip() img_file = request.files.get("image") # must match status = int(request.form.get("status", 1)) # keep old image if none uploaded image_url = notification.image print("request.files keys:", list(request.files.keys())) print("img_file:", img_file) if not title: errors["title"] = "Title is required." if not description: errors["description"] = "Description is required." if errors: return render_template("user/edit-notification.html", notification=notification, errors=errors) # handle new image upload if img_file and img_file.filename: try: filename = secure_filename(img_file.filename) upload_folder = os.path.join(current_app.static_folder, "uploads", "notifications") os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) img_file.save(save_path) # ✅ fixed # generate URL image_url = url_for("static", filename=f"uploads/notifications/{filename}", _external=True, _scheme="https") except Exception as e: current_app.logger.exception("Image upload failed") return render_template("user/edit-notification.html", notification=notification, general_error="Image upload failed") try: notification.title = title notification.description = description notification.image = image_url notification.status = status notification.save() flash("Notification updated successfully.", "success") return redirect(url_for("user.view_notifications")) except Exception as e: current_app.logger.exception("Failed to update notification: %s", e) errors["_global"] = "Server error while updating notification." return render_template("user/edit-notification.html", notification=notification, errors=errors) return render_template("user/edit-notification.html", notification=notification) # --- Delete Notification --- @user_bp.route("/user/delete-notification/", methods=["POST"]) def delete_notification(notification_id): if "user" not in session: # If AJAX, return JSON; otherwise redirect to login if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify({"success": False, "message": "Unauthorized"}), 401 return redirect(url_for("user.user_login")) notification = Notification.objects(id=notification_id).first() if not notification: if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify({"success": False, "message": "Notification not found"}), 404 flash("Notification not found.", "danger") return redirect(url_for("user.view_notifications")) try: notification.delete() if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify({"success": True, "message": "Notification deleted."}) flash("Notification deleted successfully.", "success") except Exception as e: current_app.logger.exception("Failed to delete notification: %s", e) if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return jsonify({"success": False, "message": "Failed to delete."}), 500 flash("Could not delete notification.", "danger") return redirect(url_for("user.view_notifications")) # ==================== VIDEO MANAGEMENT ROUTES ==================== # Add Videos Function start @user_bp.route('/user/add-videos', methods=['GET', 'POST']) def add_videos(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/add-videos.html') # Add Videos Function end # View Video Courses List Function start @user_bp.route('/user/view-video-courses', methods=['GET']) def view_video_courses(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/view-video-courses.html') # View Video Courses List Function end @user_bp.route('/user/add-ebook', methods=['GET', 'POST']) def add_ebook(): if 'user' not in session: return redirect(url_for('user.user_login')) errors, general_error = {}, None exams = Exam.objects(status=1).order_by('exam_title') subexams = SubExam.objects(status=1).order_by('sub_exam_title') # initial fill; UI filters via AJAX if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') title = request.form.get('title') author = request.form.get('author') status = request.form.get('status') ebook_file = request.files.get('ebook_file') exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None if not subexam: errors['sub_exam_id'] = "Select subexam" elif exam and subexam.exam_id != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" if not title or not title.strip(): errors['title'] = "Enter ebook title" if not author or not author.strip(): errors['author'] = "Enter author" if not status: errors['status'] = "Select status" if not ebook_file or ebook_file.filename == '': errors['ebook_file'] = "Select PDF" elif not allowed_ebook(ebook_file.filename): errors['ebook_file'] = "Only PDF files are allowed" else: ebook_file.seek(0, os.SEEK_END) if ebook_file.tell() > MAX_EBOOK_SIZE: errors['ebook_file'] = "File must be less than 10 MB" ebook_file.seek(0) if errors: return render_template('user/add-ebook.html', errors=errors, general_error=general_error, request=request, exams=exams, subexams=subexams) # save file try: filename = secure_filename(ebook_file.filename) uid = uuid.uuid4().hex upload_folder = os.path.join('static', 'uploads', 'ebooks') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, f"{uid}_{filename}") ebook_file.save(save_path) rp = save_path.replace(os.sep, '/') idx = rp.find("static/") file_url = '/' + rp[idx:] if idx != -1 else '/' + rp except Exception: return render_template('user/add-ebook.html', general_error="Ebook upload failed", request=request, exams=exams, subexams=subexams) try: Ebook( exam_id=exam, sub_exam_id=subexam, course_id=None, # as requested title=title.strip(), author=author.strip(), file_path=file_url, status=int(status) ).save() except Exception: return render_template('user/add-ebook.html', general_error="Something went wrong while saving", request=request, exams=exams, subexams=subexams) flash("Ebook added successfully", "success") return redirect(url_for('user.add_ebook')) return render_template('user/add-ebook.html', exams=exams, subexams=subexams) @user_bp.route('/user/view-ebooks', methods=['GET']) def ebooks_list(): if 'user' not in session: return redirect(url_for('user.user_login')) ebooks = Ebook.objects().order_by('-created_date') # Build safe, display-ready rows that work for both new and legacy docs. view_rows = [] for eb in ebooks: exam_title = '-' subexam_title = '-' # Prefer the new explicit fields try: if eb.exam_id: exam_title = getattr(eb.exam_id, 'exam_title', '-') or '-' except Exception: exam_title = '-' try: if eb.sub_exam_id: subexam_title = getattr(eb.sub_exam_id, 'sub_exam_title', '-') or '-' except Exception: subexam_title = '-' # Fallback for legacy rows that only have course_id populated if exam_title == '-' or subexam_title == '-': try: if not eb.exam_id and getattr(eb, 'course_id', None) and getattr(eb.course_id, 'exam_id', None): exam_title = getattr(eb.course_id.exam_id, 'exam_title', exam_title) or exam_title except Exception: pass try: if not eb.sub_exam_id and getattr(eb, 'course_id', None) and getattr(eb.course_id, 'sub_exam_id', None): subexam_title = getattr(eb.course_id.sub_exam_id, 'sub_exam_title', subexam_title) or subexam_title except Exception: pass view_rows.append({ 'id': str(eb.id), 'title': eb.title, 'author': eb.author, 'exam_title': exam_title, 'subexam_title': subexam_title, 'publication_date': eb.publication_date, 'status': eb.status, }) # NOTE: your template should read exam/subexam from view_rows[loop.index0] return render_template('user/view-ebooks.html', ebooks=ebooks, view_rows=view_rows) @user_bp.route('/user/edit-ebook/', methods=['GET', 'POST']) def edit_ebook(ebook_id): if 'user' not in session: return redirect(url_for('user.user_login')) ebook = Ebook.objects(id=ebook_id).first() if not ebook: flash("Ebook not found", "danger") return redirect(url_for('user.ebooks_list')) exams = Exam.objects(status=1).order_by('exam_title') subexams = SubExam.objects(status=1).order_by('sub_exam_title') errors, general_error = {}, None if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') title = request.form.get('title') author = request.form.get('author') status = request.form.get('status') ebook_file = request.files.get('ebook_file') exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None if not subexam: errors['sub_exam_id'] = "Select subexam" elif exam and subexam.exam_id != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" if not title or not title.strip(): errors['title'] = "Enter ebook title" if not author or not author.strip(): errors['author'] = "Enter author" if not status: errors['status'] = "Select status" new_file_url = None if ebook_file and ebook_file.filename != '': if not allowed_ebook(ebook_file.filename): errors['ebook_file'] = "Only PDF files are allowed" else: ebook_file.seek(0, os.SEEK_END) if ebook_file.tell() > MAX_EBOOK_SIZE: errors['ebook_file'] = "File must be less than 10 MB" ebook_file.seek(0) if errors: return render_template('user/edit-ebook.html', errors=errors, general_error=general_error, request=request, ebook=ebook, exams=exams, subexams=subexams) if ebook_file and ebook_file.filename != '': try: filename = secure_filename(ebook_file.filename) uid = uuid.uuid4().hex upload_folder = os.path.join('static', 'uploads', 'ebooks') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, f"{uid}_{filename}") ebook_file.save(save_path) rp = save_path.replace(os.sep, '/') idx = rp.find("static/") new_file_url = '/' + rp[idx:] if idx != -1 else '/' + rp if ebook.file_path and ebook.file_path.startswith('/static/'): old_abs = ebook.file_path.lstrip('/') if os.path.isfile(old_abs): try: os.remove(old_abs) except Exception: pass except Exception: return render_template('user/edit-ebook.html', general_error="Ebook upload failed", request=request, ebook=ebook, exams=exams, subexams=subexams) try: update_fields = { 'set__exam_id': exam, 'set__sub_exam_id': subexam, 'set__course_id': None, 'set__title': title.strip(), 'set__author': author.strip(), 'set__status': int(status) } if new_file_url: update_fields['set__file_path'] = new_file_url ebook.update(**update_fields) ebook.reload() flash("Ebook updated successfully", "success") return redirect(url_for('user.edit_ebook', ebook_id=ebook.id)) except Exception: return render_template('user/edit-ebook.html', general_error="Something went wrong while saving", request=request, ebook=ebook, exams=exams, subexams=subexams) return render_template('user/edit-ebook.html', ebook=ebook, exams=exams, subexams=subexams) @user_bp.route('/user/delete-ebook/', methods=['POST']) def delete_ebook(ebook_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 eb = Ebook.objects(id=ebook_id).first() if not eb: return jsonify(success=False, message="Ebook not found"), 404 try: if eb.file_path and eb.file_path.startswith('/static/'): abs_path = eb.file_path.lstrip('/') if os.path.isfile(abs_path): try: os.remove(abs_path) except Exception: pass eb.delete() return jsonify(success=True, message="Ebook deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 @user_bp.route('/user/add-old-paper', methods=['GET', 'POST']) def add_old_paper(): if 'user' not in session: return redirect(url_for('user.user_login')) errors, general_error = {}, None exams = Exam.objects(status=1).order_by('exam_title') subexams = SubExam.objects(status=1).order_by('sub_exam_title') # initial list; filtered via AJAX if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') paper_title = request.form.get('paper_title') duration = request.form.get('duration') number_of_questions = request.form.get('number_of_questions') status = request.form.get('status') content_file = request.files.get('content_file') exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None if not subexam: errors['sub_exam_id'] = "Select subexam" elif exam and subexam.exam_id != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" if not paper_title or not paper_title.strip(): errors['paper_title'] = "Enter paper title" try: duration_int = int(duration) if duration_int <= 0: errors['duration'] = "Duration must be positive" except (TypeError, ValueError): errors['duration'] = "Duration must be numeric" try: noq_int = int(number_of_questions) if noq_int <= 0: errors['number_of_questions'] = "Number must be positive" except (TypeError, ValueError): errors['number_of_questions'] = "Number must be numeric" if not status: errors['status'] = "Select status" if not content_file or content_file.filename == '': errors['content_file'] = "Select PDF" elif not allowed_old_paper(content_file.filename): errors['content_file'] = "Only PDF files are allowed" else: content_file.seek(0, os.SEEK_END) if content_file.tell() > MAX_OLD_PAPER_SIZE: errors['content_file'] = "File must be less than 10 MB" content_file.seek(0) if errors: return render_template('user/add-old-paper.html', errors=errors, general_error=general_error, request=request, exams=exams, subexams=subexams) # Save file try: filename = secure_filename(content_file.filename) uid = uuid.uuid4().hex upload_folder = os.path.join('static', 'uploads', 'old_papers') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, f"{uid}_{filename}") content_file.save(save_path) rp = save_path.replace(os.sep, '/') idx = rp.find("static/") file_url = '/' + rp[idx:] if idx != -1 else '/' + rp except Exception: return render_template('user/add-old-paper.html', general_error="PDF upload failed", request=request, exams=exams, subexams=subexams) try: OldPapers( exam_id=exam, sub_exam_id=subexam, course_id=None, # per requirements paper_title=paper_title.strip(), duration=duration_int, number_of_questions=noq_int, content_file_path=file_url, status=int(status) ).save() except Exception: return render_template('user/add-old-paper.html', general_error="Something went wrong while saving", request=request, exams=exams, subexams=subexams) flash("Old paper added successfully", "success") return redirect(url_for('user.add_old_paper')) return render_template('user/add-old-paper.html', exams=exams, subexams=subexams) @user_bp.route('/user/view-old-papers', methods=['GET']) def old_papers_list(): if 'user' not in session: return redirect(url_for('user.user_login')) old_papers = OldPapers.objects().order_by('-created_date') # Build safe, display-ready rows (handles legacy rows with only course_id) view_rows = [] for op in old_papers: exam_title = '-' subexam_title = '-' # Prefer explicit fields on OldPapers try: if op.exam_id: exam_title = getattr(op.exam_id, 'exam_title', '-') or '-' except Exception: exam_title = '-' try: if op.sub_exam_id: subexam_title = getattr(op.sub_exam_id, 'sub_exam_title', '-') or '-' except Exception: subexam_title = '-' # Fallback for legacy rows via course_id if exam_title == '-' or subexam_title == '-': try: if not op.exam_id and getattr(op, 'course_id', None) and getattr(op.course_id, 'exam_id', None): exam_title = getattr(op.course_id.exam_id, 'exam_title', exam_title) or exam_title except Exception: pass try: if not op.sub_exam_id and getattr(op, 'course_id', None) and getattr(op.course_id, 'sub_exam_id', None): subexam_title = getattr(op.course_id.sub_exam_id, 'sub_exam_title', subexam_title) or subexam_title except Exception: pass view_rows.append({ 'id': str(op.id), 'paper_title': op.paper_title, 'duration': op.duration, 'number_of_questions': op.number_of_questions, 'exam_title': exam_title, 'subexam_title': subexam_title, 'status': op.status, }) # Pass both the raw docs and the precomputed display rows return render_template('user/view-old-papers.html', old_papers=old_papers, view_rows=view_rows) @user_bp.route('/user/edit-old-paper/', methods=['GET', 'POST']) def edit_old_paper(old_paper_id): if 'user' not in session: return redirect(url_for('user.user_login')) old_paper = OldPapers.objects(id=old_paper_id).first() if not old_paper: flash("Old paper not found", "danger") return redirect(url_for('user.old_papers_list')) exams = Exam.objects(status=1).order_by('exam_title') subexams = SubExam.objects(status=1).order_by('sub_exam_title') errors, general_error = {}, None if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') paper_title = request.form.get('paper_title') duration = request.form.get('duration') number_of_questions = request.form.get('number_of_questions') status = request.form.get('status') content_file = request.files.get('content_file') exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None if not subexam: errors['sub_exam_id'] = "Select subexam" elif exam and subexam.exam_id != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" if not paper_title or not paper_title.strip(): errors['paper_title'] = "Enter paper title" try: duration_int = int(duration) if duration_int <= 0: errors['duration'] = "Duration must be positive" except (TypeError, ValueError): errors['duration'] = "Duration must be numeric" try: noq_int = int(number_of_questions) if noq_int <= 0: errors['number_of_questions'] = "Number must be positive" except (TypeError, ValueError): errors['number_of_questions'] = "Number must be numeric" if not status: errors['status'] = "Select status" new_file_url = None if content_file and content_file.filename != '': if not allowed_old_paper(content_file.filename): errors['content_file'] = "Only PDF files are allowed" else: content_file.seek(0, os.SEEK_END) if content_file.tell() > MAX_OLD_PAPER_SIZE: errors['content_file'] = "File must be less than 10 MB" content_file.seek(0) if errors: return render_template('user/edit-old-paper.html', errors=errors, general_error=general_error, request=request, old_paper=old_paper, exams=exams, subexams=subexams) if content_file and content_file.filename != '': try: filename = secure_filename(content_file.filename) uid = uuid.uuid4().hex upload_folder = os.path.join('static', 'uploads', 'old_papers') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, f"{uid}_{filename}") content_file.save(save_path) rp = save_path.replace(os.sep, '/') idx = rp.find("static/") new_file_url = '/' + rp[idx:] if idx != -1 else '/' + rp # cleanup old local file if applicable if old_paper.content_file_path and old_paper.content_file_path.startswith('/static/'): old_abs = old_paper.content_file_path.lstrip('/') if os.path.isfile(old_abs): try: os.remove(old_abs) except Exception: pass except Exception: return render_template('user/edit-old-paper.html', general_error="PDF upload failed", request=request, old_paper=old_paper, exams=exams, subexams=subexams) try: update_fields = { 'set__exam_id': exam, 'set__sub_exam_id': subexam, 'set__course_id': None, 'set__paper_title': paper_title.strip(), 'set__duration': duration_int, 'set__number_of_questions': noq_int, 'set__status': int(status) } if new_file_url: update_fields['set__content_file_path'] = new_file_url old_paper.update(**update_fields) old_paper.reload() flash("Old paper updated successfully", "success") return redirect(url_for('user.edit_old_paper', old_paper_id=old_paper.id)) except Exception: return render_template('user/edit-old-papers.html', general_error="Something went wrong while saving", request=request, old_paper=old_paper, exams=exams, subexams=subexams) return render_template('user/edit-old-paper.html', old_paper=old_paper, exams=exams, subexams=subexams) @user_bp.route('/user/delete-old-paper/', methods=['POST']) def delete_old_paper(old_paper_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 op = OldPapers.objects(id=old_paper_id).first() if not op: return jsonify(success=False, message="Old paper not found"), 404 try: if op.content_file_path and op.content_file_path.startswith('/static/'): abs_path = op.content_file_path.lstrip('/') if os.path.isfile(abs_path): try: os.remove(abs_path) except Exception: pass op.delete() return jsonify(success=True, message="Old paper deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 @user_bp.route('/user/add-mock-test', methods=['GET', 'POST']) def add_mock_test(): if 'user' not in session: return redirect(url_for('user.user_login')) errors, general_error = {}, None exams = Exam.objects(status=1).order_by('exam_title') subexams = SubExam.objects(status=1).order_by('sub_exam_title') # initial; filtered via AJAX if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') title = request.form.get('title') duration = request.form.get('duration') total_questions = request.form.get('total_questions') total_marks = request.form.get('total_marks') instructions = request.form.get('instructions') status = request.form.get('status') exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None if not subexam: errors['sub_exam_id'] = "Select subexam" elif exam and subexam.exam_id != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" if not title or not title.strip(): errors['title'] = "Enter mock test title" try: duration_int = int(duration) if duration_int <= 0: errors['duration'] = "Duration must be positive" except (TypeError, ValueError): errors['duration'] = "Duration must be numeric" try: tq_int = int(total_questions) if tq_int <= 0: errors['total_questions'] = "Total questions must be positive" except (TypeError, ValueError): errors['total_questions'] = "Total questions must be numeric" try: tm_int = int(total_marks) if tm_int <= 0: errors['total_marks'] = "Total marks must be positive" except (TypeError, ValueError): errors['total_marks'] = "Total marks must be numeric" if not instructions or not instructions.strip(): errors['instructions'] = "Enter instructions" if not status: errors['status'] = "Select status" if errors: return render_template('user/add-mock-test.html', errors=errors, general_error=general_error, request=request, exams=exams, subexams=subexams) try: new_mock=MockTest( exam_id=exam, sub_exam_id=subexam, course_id=None, # per requirement title=title.strip(), duration=duration_int, total_questions=tq_int, total_marks=tm_int, status=int(status), instructions=instructions.strip() ) new_mock.save() except Exception: return render_template('user/add-mock-test.html', general_error="Something went wrong while saving", request=request, exams=exams, subexams=subexams) uploaded = request.files.get('questions_file') if uploaded and uploaded.filename: # Basic allowed check and small size safeguard (optional) filename = secure_filename(uploaded.filename) ext = filename.rsplit('.', 1)[-1].lower() if '.' in filename else '' if ext not in ('csv', 'xlsx'): # don't alter existing errors flow — attach an error to show in UI and continue (prevents breaking) errors['questions_file'] = "Allowed file types: .csv, .xlsx" # you may choose to return render_template here; to preserve exact flow we simply flash & continue flash("Questions file ignored: invalid file type", "danger") else: imported_count = 0 import_errors = [] try: try: uploaded.stream.seek(0) except Exception: pass # CSV if ext == 'csv': print("inside if csv") try: df=pd.read_csv(uploaded) except: uploaded.stream.seek(0) content = uploaded.stream.read() text = content.decode('utf-8', errors='replace') df = pd.read_csv(StringIO(text)) else: uploaded.stream.seek(0) df = pd.read_excel(uploaded, engine='openpyxl') # Normalize headers to lowercase stripped strings df.columns = [str(c).strip().lower() if c is not None else '' for c in df.columns] for idx, row in df.iterrows(): row_num = idx + 2 if ext == 'xlsx' else idx + 2 # +2 to account for header row (makes sense in CSV too) try: # use .get-like access for DataFrame row values question_text = row.get('question') if hasattr(row, 'get') else row['question'] if 'question' in row else None # pandas may give numpy.nan for empty -> treat as empty string if isinstance(question_text, float) and pd.isna(question_text): question_text = '' question_text = (str(question_text).strip() if question_text is not None else '') if not question_text: import_errors.append({'row': row_num, 'error': 'Empty question'}) continue q_type = row.get('question_type') if hasattr(row, 'get') else row['question_type'] if 'question_type' in row else None q_type = (str(q_type).strip() if q_type not in (None, float) and not (isinstance(q_type, float) and pd.isna(q_type)) else 'mcq') if not q_type: q_type = 'mcq' raw_options = row.get('options') if hasattr(row, 'get') else row['options'] if 'options' in row else '' if isinstance(raw_options, float) and pd.isna(raw_options): raw_options = '' raw_options = raw_options if raw_options is not None else '' answer = row.get('answer') if hasattr(row, 'get') else row['answer'] if 'answer' in row else '' if isinstance(answer, float) and pd.isna(answer): answer = '' answer = str(answer).strip() if answer is not None else '' raw_marks = row.get('marks') if hasattr(row, 'get') else row['marks'] if 'marks' in row else 0 if isinstance(raw_marks, float) and pd.isna(raw_marks): raw_marks = 0 try: marks = int(float(raw_marks)) if raw_marks != '' else 0 except Exception: marks = 0 # parse options: JSON array preferred, else pipe-separated options_parsed = [] if raw_options is None: raw_options = '' if isinstance(raw_options, (list, tuple)): # already a list (pandas may convert Excel arrays), keep as-is options_parsed = list(raw_options) else: raw_options_str = str(raw_options).strip() if raw_options_str: try: parsed = json.loads(raw_options_str) if isinstance(parsed, list): options_parsed = parsed else: options_parsed = [str(parsed)] except Exception: # fallback to pipe-separated options_parsed = [p.strip() for p in raw_options_str.split('|') if p.strip()] # Create and save the question try: q = MockTestQuestion( mock_test=new_mock.id, question_type=q_type, question=question_text, options=options_parsed, answer=answer, marks=marks ) q.save() imported_count += 1 except Exception as e: current_app.logger.exception("Failed to save question row %s: %s", row_num, e) import_errors.append({'row': row_num, 'error': str(e)}) except Exception as e: current_app.logger.exception("Error processing row %s: %s", row_num, e) import_errors.append({'row': row_num, 'error': str(e)}) except Exception as e: import_errors.append({'row': 'file', 'error': str(e)}) # optional: flash summary, but DOES NOT prevent normal redirect/behavior if imported_count: flash(f"Imported {imported_count} questions from uploaded file.", "success") if import_errors: flash(f"{len(import_errors)} errors occurred while importing questions. See server logs.", "danger") flash("Mock test added successfully", "success") return redirect(url_for('user.add_mock_test')) return render_template('user/add-mock-test.html', exams=exams, subexams=subexams) # View Mock Tests (robust against missing Course refs) @user_bp.route('/user/view-mock-tests', methods=['GET']) def mock_tests_list(): if 'user' not in session: return redirect(url_for('user.user_login')) mock_tests = MockTest.objects().order_by('-created_date') # Build safe, display-ready rows (handles legacy docs with only course_id) view_rows = [] for mt in mock_tests: exam_title = '-' subexam_title = '-' # Prefer explicit fields on MockTest try: if mt.exam_id: exam_title = getattr(mt.exam_id, 'exam_title', '-') or '-' except Exception: exam_title = '-' try: if mt.sub_exam_id: subexam_title = getattr(mt.sub_exam_id, 'sub_exam_title', '-') or '-' except Exception: subexam_title = '-' # Fallback for legacy rows via course_id (if that course points to exam/subexam) if exam_title == '-' or subexam_title == '-': try: if not mt.exam_id and getattr(mt, 'course_id', None) and getattr(mt.course_id, 'exam_id', None): exam_title = getattr(mt.course_id.exam_id, 'exam_title', exam_title) or exam_title except Exception: pass try: if not mt.sub_exam_id and getattr(mt, 'course_id', None) and getattr(mt.course_id, 'sub_exam_id', None): subexam_title = getattr(mt.course_id.sub_exam_id, 'sub_exam_title', subexam_title) or subexam_title except Exception: pass view_rows.append({ 'id': str(mt.id), 'title': mt.title, 'duration': mt.duration, 'total_questions': mt.total_questions, 'total_marks': mt.total_marks, 'instructions': mt.instructions, 'exam_title': exam_title, 'subexam_title': subexam_title, 'status': mt.status, }) return render_template('user/view-mock-tests.html', mock_tests=mock_tests, view_rows=view_rows) @user_bp.route('/user/edit-mock-test/', methods=['GET', 'POST']) def edit_mock_test(mock_test_id): if 'user' not in session: return redirect(url_for('user.user_login')) mock_test = MockTest.objects(id=mock_test_id).first() if not mock_test: flash("Mock test not found", "danger") return redirect(url_for('user.mock_tests_list')) exams = Exam.objects(status=1).order_by('exam_title') subexams = SubExam.objects(status=1).order_by('sub_exam_title') errors, general_error = {}, None if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') title = request.form.get('title') duration = request.form.get('duration') total_questions = request.form.get('total_questions') total_marks = request.form.get('total_marks') instructions = request.form.get('instructions') status = request.form.get('status') exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None if not subexam: errors['sub_exam_id'] = "Select subexam" elif exam and subexam.exam_id != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" if not title or not title.strip(): errors['title'] = "Enter mock test title" try: duration_int = int(duration) if duration_int <= 0: errors['duration'] = "Duration must be positive" except (TypeError, ValueError): errors['duration'] = "Duration must be numeric" try: tq_int = int(total_questions) if tq_int <= 0: errors['total_questions'] = "Total questions must be positive" except (TypeError, ValueError): errors['total_questions'] = "Total questions must be numeric" try: tm_int = int(total_marks) if tm_int <= 0: errors['total_marks'] = "Total marks must be positive" except (TypeError, ValueError): errors['total_marks'] = "Total marks must be numeric" if not instructions or not instructions.strip(): errors['instructions'] = "Enter instructions" if not status: errors['status'] = "Select status" if errors: return render_template('user/edit-mock-test.html', errors=errors, general_error=general_error, request=request, mock_test=mock_test, exams=exams, subexams=subexams) try: mock_test.update( set__exam_id=exam, set__sub_exam_id=subexam, set__course_id=None, set__title=title.strip(), set__duration=duration_int, set__total_questions=tq_int, set__total_marks=tm_int, set__instructions=instructions.strip(), set__status=int(status) ) mock_test.reload() flash("Mock test updated successfully", "success") return redirect(url_for('user.edit_mock_test', mock_test_id=mock_test.id)) except Exception: return render_template('user/edit-mock-test.html', general_error="Something went wrong while saving", request=request, mock_test=mock_test, exams=exams, subexams=subexams) return render_template('user/edit-mock-test.html', mock_test=mock_test, exams=exams, subexams=subexams) # Delete Mock Test (AJAX) @user_bp.route('/user/delete-mock-test/', methods=['POST']) def delete_mock_test(mock_test_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 mt = MockTest.objects(id=mock_test_id).first() if not mt: return jsonify(success=False, message="Mock test not found"), 404 try: mt.delete() return jsonify(success=True, message="Mock test deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 # View Add Mock Tests Questions Function start @user_bp.route('/user/add-question', methods=['GET', 'POST']) def add_question(): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} form_data = {} mock_tests=[] # fetch mock tests for select box try: mock_tests = MockTest.objects.order_by('-created_date') except Exception: mock_tests = [] if request.method == 'POST': # collect values form_data = request.form.to_dict(flat=True) mock_test_id = request.form.get('mock_test') or '' question_type = request.form.get('question_type') or '' topic = (request.form.get('topic') or '').strip() question_text = (request.form.get('question') or '').strip() marks_raw = request.form.get('marks') or '' answer = (request.form.get('answer') or '').strip() # options — gather as a list of non-empty strings options = [] # if options sent as option[] inputs raw_options = request.form.getlist('options[]') if raw_options: for o in raw_options: if o and o.strip(): options.append(o.strip()) # server-side validation if not mock_test_id: errors['mock_test'] = 'Please select a mock test.' else: try: mock_test_obj = MockTest.objects.get(id=mock_test_id) except DoesNotExist: errors['mock_test'] = 'Selected mock test not found.' mock_test_obj = None except Exception: current_app.logger.exception("Error fetching mock test") errors['mock_test'] = 'Unable to find mock test.' mock_test_obj = None if not question_type: errors['question_type'] = 'Please select a question type.' if not question_text: errors['question'] = 'Question text is required.' # marks validation (optional; default 1) try: marks = int(marks_raw) if marks_raw else 1 if marks <= 0: errors['marks'] = 'Marks must be a positive integer.' except ValueError: errors['marks'] = 'Marks must be an integer.' marks = 1 # type-specific checks # type-specific checks (MCQ and Match The Following behave the same) if question_type in ('MCQ', 'MatchTheFollowing'): # need at least two options and an answer that matches one option if len(options) < 2: errors['options'] = f'{question_type} must have at least 2 options.' if not answer: errors['answer'] = 'Answer is required for ' + ('MCQ' if question_type == 'MCQ' else 'Match The Following') + '.' else: # allow answer to be option text or an index (1-based) answer_text = None if answer.isdigit(): idx = int(answer) - 1 if 0 <= idx < len(options): answer_text = options[idx] else: errors['answer'] = 'Answer option index is out of range.' else: # match by exact option string (then case-insensitive) if answer in options: answer_text = answer else: matched = next((o for o in options if o.strip().lower() == answer.strip().lower()), None) if matched: answer_text = matched else: errors['answer'] = 'Answer must match one of the options.' if not errors.get('answer'): answer = answer_text else: if not answer: errors['answer'] = 'Answer is required.' # if no validation errors, save the question if not errors: try: q = MockTestQuestion( mock_test = mock_test_obj, question_type = question_type, topic = topic or None, question = question_text, options = options if options else [], answer = answer, marks = marks ) q.save() flash('Question added successfully.', 'success') return redirect(url_for('user.view_questions')) except ValidationError as e: current_app.logger.exception("Validation error saving question") errors['general'] = str(e) except Exception: current_app.logger.exception("Failed to save question") errors['general'] = 'Failed to save question. See logs.' # GET or POST with errors -> render the form return render_template('user/add-question.html', mock_tests=mock_tests, errors=errors, form_data=form_data) # View Mock Tests Questions List Function start @user_bp.route('/user/view-questions', methods=['GET']) def view_questions(): if 'user' not in session: return redirect(url_for('user.user_login')) try: # select_related to dereference mock_test ReferenceField for template access questions = MockTestQuestion.objects.order_by('-created_date').select_related() current_app.logger.debug("view_questions: loaded %d questions", len(questions)) except Exception: current_app.logger.exception("Failed to load questions") questions = [] return render_template('user/view-questions.html', questions=questions) # View Edit Mock Tests Questions Function start @user_bp.route('/user/edit-question/', methods=['GET', 'POST']) def edit_question(question_id): if 'user' not in session: return redirect(url_for('user.user_login')) try: q = MockTestQuestion.objects.get(id=question_id) except DoesNotExist: flash("Question not found.", "danger") return redirect(url_for('user.view_questions')) except Exception: current_app.logger.exception("Error fetching question") flash("Unable to load question.", "danger") return redirect(url_for('user.view_questions')) errors = {} form_data = {} # prefill form_data for GET if request.method == 'GET': form_data = { 'mock_test': str(q.mock_test.id) if q.mock_test else '', 'question_type': q.question_type or '', 'topic': q.topic or '', 'question': q.question or '', 'options': q.options or [], 'answer': q.answer or '', 'marks': str(q.marks or 1), 'status': str(q.status) if getattr(q, 'status', None) is not None else '1' } if request.method == 'POST': # gather inputs form_data = request.form.to_dict(flat=True) mock_test_id = request.form.get('mock_test') or '' question_type = (request.form.get('question_type') or q.question_type or '').strip() topic = (request.form.get('topic') or '').strip() question_text = (request.form.get('question') or '').strip() marks_raw = request.form.get('marks') or '' answer_raw = (request.form.get('answer') or '').strip() # gather options[] inputs (may be empty) options = [o.strip() for o in request.form.getlist('options[]') if o and o.strip()] # validate mock_test link mock_test_obj = None if mock_test_id: try: mock_test_obj = MockTest.objects.get(id=mock_test_id) except DoesNotExist: errors['mock_test'] = 'Selected mock test not found.' except Exception: current_app.logger.exception("Error fetching mock test") errors['mock_test'] = 'Unable to find mock test.' if not question_text: errors['question'] = 'Question text is required.' # marks try: marks = int(marks_raw) if marks_raw else 1 if marks <= 0: errors['marks'] = 'Marks must be a positive integer.' except ValueError: errors['marks'] = 'Marks must be an integer.' marks = 1 # --- TYPE-SPECIFIC VALIDATION --- answer = None if question_type in ('MCQ', 'MatchTheFollowing'): # require at least 2 options if len(options) < 2: errors['options'] = 'Please provide at least 2 options for ' + ( 'MCQ' if question_type == 'MCQ' else 'Match The Following') + '.' if not answer_raw: errors['answer'] = 'Answer is required for ' + ( 'MCQ' if question_type == 'MCQ' else 'Match The Following') + '.' else: # Accept option text or 1-based index if answer_raw.isdigit(): idx = int(answer_raw) - 1 if 0 <= idx < len(options): answer = options[idx] else: errors['answer'] = 'Answer index is out of range.' else: # exact match then case-insensitive matched = next((o for o in options if o.strip() == answer_raw.strip()), None) if not matched: matched = next((o for o in options if o.strip().lower() == answer_raw.strip().lower()), None) if matched: answer = matched else: errors['answer'] = 'Answer must match one of the options.' else: # Non-MCQ types: accept free-text answer (still required) if not answer_raw: errors['answer'] = 'Answer is required.' else: answer = answer_raw # when switching away from MCQ-like, we'll clear options on save # status handling status_val = request.form.get('status') try: status_int = int(status_val) if status_val is not None else getattr(q, 'status', 1) except Exception: status_int = getattr(q, 'status', 1) # save if no errors if not errors: try: q.mock_test = mock_test_obj if mock_test_obj else q.mock_test q.question_type = question_type or q.question_type q.topic = topic or None q.question = question_text # Save options only for MCQ-like; otherwise clear them if question_type in ('MCQ', 'MatchTheFollowing'): q.options = options else: q.options = [] q.answer = answer q.marks = marks q.status = status_int q.save() flash('Question updated successfully.', 'success') return redirect(url_for('user.view_questions')) except ValidationError as e: current_app.logger.exception("Validation error updating question") errors['general'] = str(e) except Exception: current_app.logger.exception("Failed to update question") errors['general'] = 'Failed to update question.' # fetch mock_tests for dropdown in edit form try: mock_tests = MockTest.objects.order_by('-created_date') except Exception: mock_tests = [] return render_template('user/edit-question.html', question=q, mock_tests=mock_tests, errors=errors, form_data=form_data) # Delete question (AJAX POST) @user_bp.route('/user/delete-question/', methods=['POST']) def delete_question(question_id): if 'user' not in session: return jsonify({'success': False, 'message': 'Not authenticated.'}), 401 try: q = MockTestQuestion.objects.get(id=question_id) q.delete() return jsonify({'success': True, 'message': 'Question deleted.'}) except DoesNotExist: return jsonify({'success': False, 'message': 'Question not found.'}), 404 except Exception: current_app.logger.exception("Failed to delete question %s", question_id) return jsonify({'success': False, 'message': 'Server error.'}), 500 # ==================== AUTHENTICATION ROUTES ==================== # Forgot Password Function start @user_bp.route('/user/forgot-password', methods=['GET', 'POST']) def forgot_password(): if 'user' in session: return redirect(url_for('user.user_dashboard')) return render_template('user/forgot-password.html') # Forgot Password Function end # Reset Password Function start @user_bp.route('/user/reset-password', methods=['GET', 'POST']) def reset_password(): if 'user' in session: return redirect(url_for('user.user_dashboard')) return render_template('user/reset-password.html') # Reset Password Function end # ==================== OTHER ROUTES ==================== # MCQ Extractor Function start # @user_bp.route('/user/mcq-extractor', methods=['GET', 'POST']) # def mcq_extractor(): # if 'user' not in session: # return redirect(url_for('user.user_login')) # return render_template('user/mcq-extractor.html') # MCQ Extractor Function end # View Transactions Function start @user_bp.route('/user/view-transactions', methods=['GET']) def view_transactions(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/view-transactions.html') # View Transactions Function end # Avatar and Video Content Upload page @user_bp.route('/user/content-upload', methods=['GET']) def content_upload(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/content-upload.html') # View Change Password Function start @user_bp.route('/user/change-password', methods=['GET']) def change_password(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/change-password.html') # View Change Password Function start # User Logout @user_bp.route('/user/logout') def user_logout(): session.clear() # Clears everything including permanent session return redirect(url_for('user.user_login')) # User Logout def _parse_day_range(key, custom_start=None, custom_end=None): if not key: return (None, None) today = date.today() try: if key == 'today': start = datetime.combine(today, time.min); end = datetime.combine(today, time.max) elif key == 'yesterday': d = today - timedelta(days=1); start = datetime.combine(d, time.min); end = datetime.combine(d, time.max) elif key == '7': start = datetime.combine(today - timedelta(days=6), time.min); end = datetime.combine(today, time.max) elif key == '30': start = datetime.combine(today - timedelta(days=29), time.min); end = datetime.combine(today, time.max) elif key == 'this_month': start = datetime.combine(today.replace(day=1), time.min) last_day = calendar.monthrange(today.year, today.month)[1] end = datetime.combine(today.replace(day=last_day), time.max) elif key == 'last_month': year = today.year; month = today.month - 1 if month == 0: month = 12; year -= 1 start = datetime.combine(date(year, month, 1), time.min) last_day = calendar.monthrange(year, month)[1] end = datetime.combine(date(year, month, last_day), time.max) elif key == 'custom': if not custom_start or not custom_end: return (None, None) s = datetime.strptime(custom_start, "%Y-%m-%d") e = datetime.strptime(custom_end, "%Y-%m-%d") start = datetime.combine(s.date(), time.min); end = datetime.combine(e.date(), time.max) else: return (None, None) except Exception: return (None, None) return (start, end) @user_bp.route("/user/purchase-history", methods=['GET','POST']) def purchase_history(): if "user" not in session: return redirect(url_for("user.user_login")) # prefer GET (your JS builds a GET URL), but support POST too params = request.args if request.method == 'GET' else request.form # read filter params trans_key = params.get('transaction_date', '') t_custom_start = params.get('t_custom_start', '') t_custom_end = params.get('t_custom_end', '') reg_key = params.get('register_date', '') r_custom_start = params.get('r_custom_start', '') r_custom_end = params.get('r_custom_end', '') status = params.get('status', '') # '', '1', '0', '-1' # compute date ranges trans_start, trans_end = _parse_day_range(trans_key, t_custom_start, t_custom_end) reg_start, reg_end = _parse_day_range(reg_key, r_custom_start, r_custom_end) debug_info = { 'received': {'transaction_date': trans_key, 't_custom_start': t_custom_start, 't_custom_end': t_custom_end, 'register_date': reg_key, 'r_custom_start': r_custom_start, 'r_custom_end': r_custom_end, 'status': status}, 'applied': {} } try: base_q = Q(entry_type='credit') # apply transaction_status if present (convert to int when possible) if status != '': try: st = int(status) base_q = base_q & Q(transaction_status=st) debug_info['applied']['status'] = st except Exception: base_q = base_q & Q(transaction_status=status) debug_info['applied']['status'] = status # transaction date filter (TransactionHistory.created_date) if trans_start and trans_end: base_q = base_q & Q(created_date__gte=trans_start) & Q(created_date__lte=trans_end) debug_info['applied']['transaction_date'] = (trans_start.isoformat(), trans_end.isoformat()) # register date filter -> find user ids and filter transactions by them if reg_start and reg_end: debug_info['applied']['register_date'] = (reg_start.isoformat(), reg_end.isoformat()) # try multiple common user-created field names user_ids = [] try: # check 'created_at' user_qs = User.objects(created_at__gte=reg_start, created_at__lte=reg_end).only('id') user_ids = [u.id for u in user_qs] except Exception: user_ids = [] if not user_ids: try: user_qs = User.objects(created_date__gte=reg_start, created_date__lte=reg_end).only('id') user_ids = [u.id for u in user_qs] except Exception: user_ids = [] if not user_ids: try: user_qs = User.objects(created_on__gte=reg_start, created_on__lte=reg_end).only('id') user_ids = [u.id for u in user_qs] except Exception: user_ids = [] debug_info['applied']['register_user_count'] = len(user_ids) debug_info['applied']['register_user_sample'] = [str(u) for u in (user_ids[:5] or [])] if not user_ids: # no users matched -> no transactions q = [] else: base_q = base_q & Q(user_id__in=user_ids) # execute query (unless short-circuited above) if 'q' not in locals(): q = TransactionHistory.objects(base_q).order_by('-created_date').no_dereference() except Exception as e: current_app.logger.exception("Failed to fetch purchase transactions: %s", e) q = [] rows = [] for t in q: try: # get raw user reference without forcing dereference raw = None if hasattr(t, '_data') and 'user_id' in t._data: raw = t._data.get('user_id') if raw is None: raw = getattr(t, 'user_id', None) user_ref_id = getattr(raw, 'id', None) or getattr(raw, 'pk', None) or raw if not user_ref_id: continue user_obj = User.objects(id=user_ref_id).first() if not user_obj: continue # transaction created_date -> string (match template field name) tx_created = getattr(t, 'created_date', None) if tx_created and hasattr(tx_created, 'strftime'): tx_created_str = tx_created.strftime('%d-%m-%Y %H:%M:%S') else: tx_created_str = str(tx_created) if tx_created is not None else '' # user's created (try common names) user_created = getattr(user_obj, 'created_at', None) if user_created is None: user_created = getattr(user_obj, 'created_date', None) if user_created is None: user_created = getattr(user_obj, 'created_on', None) if user_created and hasattr(user_created, 'strftime'): user_created_str = user_created.strftime('%d-%m-%Y %H:%M:%S') else: user_created_str = str(user_created) if user_created is not None else '' rows.append({ 'id': str(getattr(t, 'id', '') or ''), 'user_name': getattr(user_obj, 'name', '—') or '—', 'user_email': getattr(user_obj, 'email', '—') or '—', 'user_phone': getattr(user_obj, 'phone', '—') or '—', 'amount': getattr(t, 'amount', 0) or 0, 'payment_status': getattr(t, 'transaction_status', None), 'created_date': tx_created_str, 'user_created_at': user_created_str, 'description': getattr(t, 'description', '') or '', }) except Exception: current_app.logger.exception("Skipping transaction due to error for id=%r", getattr(t, 'id', None)) continue current_app.logger.debug("purchase_history debug: %s", debug_info) return render_template('user/purchase-history.html', transactions=rows, view_type='purchase', page_title='Purchase History') @user_bp.route("/user/spent-history", methods=['GET','POST']) def spent_history(): if "user" not in session: return redirect(url_for("user.user_login")) params = request.args if request.method == 'GET' else request.form trans_key = params.get('transaction_date', '') t_custom_start = params.get('t_custom_start', '') t_custom_end = params.get('t_custom_end', '') reg_key = params.get('register_date', '') r_custom_start = params.get('r_custom_start', '') r_custom_end = params.get('r_custom_end', '') status = params.get('status', '') trans_start, trans_end = _parse_day_range(trans_key, t_custom_start, t_custom_end) reg_start, reg_end = _parse_day_range(reg_key, r_custom_start, r_custom_end) try: base_q = Q(entry_type='debit') if status != '': try: st = int(status) base_q = base_q & Q(transaction_status=st) except Exception: base_q = base_q & Q(transaction_status=status) if trans_start and trans_end: base_q = base_q & Q(created_date__gte=trans_start) & Q(created_date__lte=trans_end) # register-date filter: find users and restrict if reg_start and reg_end: user_ids = [] try: user_qs = User.objects(created_at__gte=reg_start, created_at__lte=reg_end).only('id') user_ids = [u.id for u in user_qs] except Exception: user_ids = [] if not user_ids: try: user_qs = User.objects(created_date__gte=reg_start, created_date__lte=reg_end).only('id') user_ids = [u.id for u in user_qs] except Exception: user_ids = [] if not user_ids: try: user_qs = User.objects(created_on__gte=reg_start, created_on__lte=reg_end).only('id') user_ids = [u.id for u in user_qs] except Exception: user_ids = [] if not user_ids: q = [] else: base_q = base_q & Q(user_id__in=user_ids) if 'q' not in locals(): q = TransactionHistory.objects(base_q).order_by('-created_date').no_dereference() except Exception: current_app.logger.exception("Failed to fetch spent transactions") q = [] rows = [] for t in q: try: raw = None if hasattr(t, '_data') and 'user_id' in t._data: raw = t._data.get('user_id') if raw is None: raw = getattr(t, 'user_id', None) user_ref_id = getattr(raw, 'id', None) or getattr(raw, 'pk', None) or raw if not user_ref_id: continue user_obj = User.objects(id=user_ref_id).first() if not user_obj: continue tx_created = getattr(t, 'created_date', None) if tx_created and hasattr(tx_created, 'strftime'): tx_created_str = tx_created.strftime('%d-%m-%Y %H:%M:%S') else: tx_created_str = str(tx_created) if tx_created is not None else '' user_created = getattr(user_obj, 'created_at', None) if user_created is None: user_created = getattr(user_obj, 'created_date', None) if user_created is None: user_created = getattr(user_obj, 'created_on', None) if user_created and hasattr(user_created, 'strftime'): user_created_str = user_created.strftime('%d-%m-%Y %H:%M:%S') else: user_created_str = str(user_created) if user_created is not None else '' rows.append({ 'id': str(getattr(t, 'id', '') or ''), 'user_name': getattr(user_obj, 'name', '—') or '—', 'user_email': getattr(user_obj, 'email', '—') or '—', 'user_phone': getattr(user_obj, 'phone', '—') or '—', 'amount': getattr(t, 'amount', 0) or 0, 'payment_status': getattr(t, 'transaction_status', None), 'created_date': tx_created_str, 'user_created_at': user_created_str, 'description': getattr(t, 'description', '') or '', }) except Exception: current_app.logger.exception("Skipping transaction due to error for id=%r", getattr(t, 'id', None)) continue return render_template('user/spent-history.html', transactions=rows, view_type='spent', page_title='Spent History') @user_bp.route('/subexams/') def get_subexams(exam_id): try: subs = SubExam.objects.filter(exam_id=exam_id, status=1).order_by('name') data = [{'id': str(s.id), 'name': (s.name if hasattr(s, 'name') else getattr(s, 'title', str(s.id)))} for s in subs] return jsonify(success=True, data=data) except Exception as e: return jsonify(success=False, message=str(e)), 500 @user_bp.route("/user/test-history", methods=['GET','POST']) def test_history(): if "user" not in session: return redirect(url_for("user.user_login")) return render_template('user/view-transactions.html')