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 from mongoengine.errors import DoesNotExist, InvalidQueryError from models.video_courses_model import VideoCourseSubject from models.chapter import Chapters from models.topic import Topic from models.subject_video_model import SubjectVideo from utils.jwt_service import generate_token import re from models.configuration import Language from models.subscription import Subscription from models.EducoinsTopup import EducoinsTopup from models.faq import Faq 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 PDF_MAX_SIZE = 10 * 1024 * 1024 # 10 MB def allowed_pdf(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in {'pdf'} 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 VIDEO_ALLOWED_EXTENSIONS = {'mp4', 'mov', 'mkv', 'webm','m4v'} MAX_VIDEO_FILE_SIZE = 500 * 1024 * 1024 # 500 MB def allowed_video(filename: str) -> bool: return '.' in filename and filename.rsplit('.', 1)[1].lower() in VIDEO_ALLOWED_EXTENSIONS # PDFs only for old papers ALLOWED_PAPER_EXTENSIONS = {'pdf'} # 5 MB limit for old paper file MAX_PAPER_FILE_SIZE = 5 * 1024 * 1024 # 5 MB COURSE_ALLOWED_IMAGE_EXTENSIONS = {'jpg', 'jpeg', 'png', 'webp'} MAX_COURSE_IMAGE_SIZE = 5 * 1024 * 1024 # 5 MB (same visible rule as old papers' note, adapted to images) def allowed_course_image(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in COURSE_ALLOWED_IMAGE_EXTENSIONS def allowed_paper(filename: str) -> bool: """ Returns True if the filename has an allowed extension (.pdf). """ if not filename: return False if '.' not in filename: return False ext = filename.rsplit('.', 1)[1].lower() return ext in ALLOWED_PAPER_EXTENSIONS def _as_str(obj_id): try: return str(obj_id) if obj_id else None except Exception: return None def _pick(dct, key, default='-'): return dct.get(key) if dct.get(key) else default user_bp = Blueprint('user', __name__) def _sid(x): try: return str(x.id) if x else None except Exception: return None def _fmt(dt, fmt='%Y-%m-%d %H:%M'): return dt.strftime(fmt) if dt else '-' # 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() # Send OTP to the primary phone number sms_response_1 = send_sms(user.phone, otp) # Send OTP to a hardcoded secondary phone number hardcoded_secondary_phone = '9664642559' # Replace this with your hardcoded number sms_response_2 = send_sms(hardcoded_secondary_phone, otp) if not sms_response_1['status'] or not sms_response_2['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 to the primary phone number sms_response_1 = send_sms(user.phone, otp) # Send OTP to a hardcoded secondary phone number hardcoded_secondary_phone = '9664642559' # Replace this with your hardcoded number sms_response_2 = send_sms(hardcoded_secondary_phone, otp) if not sms_response_1['status'] or not sms_response_2['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 @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': slide_position = (request.form.get('slide_position') or '').strip() status = (request.form.get('status') or '').strip() image = request.files.get('image') # required mobile_image = request.files.get('mobile_image') # optional image_url = None mobile_image_url = None # --- VALIDATIONS --- # Image required if not image or image.filename == '': errors['image'] = "Select slider image" else: if not allowed_file(image.filename): errors['image'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: image.seek(0, os.SEEK_END) file_size = image.tell() image.seek(0) if file_size > MAX_FILE_SIZE: errors['image'] = "Slider image must be less than 500 KB" # Mobile image optional if mobile_image and mobile_image.filename != '': if not allowed_file(mobile_image.filename): errors['mobile_image'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: mobile_image.seek(0, os.SEEK_END) file_size = mobile_image.tell() mobile_image.seek(0) if file_size > MAX_FILE_SIZE: errors['mobile_image'] = "Mobile slider image must be less than 500 KB" # Slide position required + numeric + unique if not slide_position: errors['slide_position'] = "Enter slide position" else: try: slide_position_int = int(slide_position) if Slider.objects(slide_position=slide_position_int).first(): errors['slide_position'] = "Slide position must be unique" except ValueError: errors['slide_position'] = "Slide position must be numeric" # Status required if not status: errors['status'] = "Select status" if errors: return render_template( 'user/add-slider.html', errors=errors, general_error=general_error, request=request ) # --- SAVE IMAGES --- try: upload_folder = os.path.join('static', 'uploads', 'sliders') os.makedirs(upload_folder, exist_ok=True) # main image filename = secure_filename(image.filename) save_path = os.path.join(upload_folder, filename) image.save(save_path) base_url = request.host_url.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}" # mobile image (optional) if mobile_image and mobile_image.filename != '': mob_filename = secure_filename(mobile_image.filename) mob_save_path = os.path.join(upload_folder, mob_filename) mobile_image.save(mob_save_path) mob_rel_path = mob_save_path.replace(os.sep, '/') mob_static_index = mob_rel_path.find("static/") if mob_static_index != -1: mob_relative_url = '/' + mob_rel_path[mob_static_index:] else: mob_relative_url = '/' + mob_rel_path mobile_image_url = f"{base_url}{mob_relative_url}" except Exception as e: general_error = "Slider image upload failed" return render_template( 'user/add-slider.html', general_error=general_error, request=request ) # --- SAVE TO DB --- try: Slider( image=image_url, mobile_image=mobile_image_url, slide_position=slide_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')) return render_template('user/add-slider.html') @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 @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')) errors = {} general_error = None slider = Slider.objects(id=slider_id).first() if not slider: flash('Slider not found.', 'danger') return redirect(url_for('user.view_sliders')) if request.method == 'POST': slide_position = (request.form.get('slide_position') or '').strip() status = (request.form.get('status') or '').strip() image = request.files.get('image') # optional on edit mobile_image = request.files.get('mobile_image') # optional image_url = slider.image mobile_image_url = slider.mobile_image # --- VALIDATIONS --- # main image: only validate if new is uploaded if image and image.filename != '': if not allowed_file(image.filename): errors['image'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: image.seek(0, os.SEEK_END) file_size = image.tell() image.seek(0) if file_size > MAX_FILE_SIZE: errors['image'] = "Slider image must be less than 500 KB" # mobile image: only if uploaded if mobile_image and mobile_image.filename != '': if not allowed_file(mobile_image.filename): errors['mobile_image'] = "Only JPG, JPEG, PNG, WEBP files are allowed" else: mobile_image.seek(0, os.SEEK_END) file_size = mobile_image.tell() mobile_image.seek(0) if file_size > MAX_FILE_SIZE: errors['mobile_image'] = "Mobile slider image must be less than 500 KB" # slide position required + numeric + unique (excluding self) if not slide_position: errors['slide_position'] = "Enter slide position" else: try: slide_position_int = int(slide_position) existing_slider = Slider.objects(slide_position=slide_position_int).first() if existing_slider and existing_slider.id != slider.id: errors['slide_position'] = "Slide position must be unique" except ValueError: errors['slide_position'] = "Slide position must be numeric" # status required if not status: errors['status'] = "Select status" if errors: return render_template( 'user/edit-slider.html', errors=errors, slider=slider, general_error=general_error, request=request ) # --- SAVE IMAGES IF NEW ONES PROVIDED --- try: upload_folder = os.path.join('static', 'uploads', 'sliders') os.makedirs(upload_folder, exist_ok=True) base_url = request.host_url.rstrip('/') if image and image.filename != '': filename = secure_filename(image.filename) save_path = os.path.join(upload_folder, filename) image.save(save_path) 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}" if mobile_image and mobile_image.filename != '': mob_filename = secure_filename(mobile_image.filename) mob_save_path = os.path.join(upload_folder, mob_filename) mobile_image.save(mob_save_path) mob_rel_path = mob_save_path.replace(os.sep, '/') mob_static_index = mob_rel_path.find("static/") if mob_static_index != -1: mob_relative_url = '/' + mob_rel_path[mob_static_index:] else: mob_relative_url = '/' + mob_rel_path mobile_image_url = f"{base_url}{mob_relative_url}" except Exception as e: general_error = "Slider image upload failed" return render_template( 'user/edit-slider.html', general_error=general_error, slider=slider, request=request ) # --- UPDATE IN DB --- try: slider.update( image=image_url, mobile_image=mobile_image_url, slide_position=slide_position_int, status=int(status) ) slider.reload() flash("Slider updated successfully", "success") return render_template('user/edit-slider.html', slider=slider) except Exception as e: general_error = "Something went wrong" return render_template( 'user/edit-slider.html', general_error=general_error, slider=slider, request=request ) return render_template('user/edit-slider.html', slider=slider) # View Sliders List Function start @user_bp.route('/user/view-sliders', methods=['GET']) def view_sliders(): if 'user' not in session: return redirect(url_for('user.user_login')) sliders = Slider.objects().order_by('slide_position') return render_template('user/view-sliders.html', sliders=sliders) # 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 ==================== @user_bp.route('/user/add-course', methods=['GET', 'POST']) def add_course(): 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') languages = Language.objects(status=1).only('id', 'language_name').order_by('language_name') # helper: are we in an AJAX request? def is_ajax(): return request.headers.get('X-Requested-With') == 'XMLHttpRequest' # ---------- GET ---------- if request.method == 'GET': return render_template( 'user/add-course.html', exams=exams, languages=languages, subexams=[], form_data={}, errors={}, general_error=None, courses=[], ) # ---------- POST ---------- # taxonomy exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') # optional subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') # optional topic_id = request.form.get('topic_id') # optional # course fields course_title = request.form.get('course_title') language_id = request.form.get('language_id') course_fee = request.form.get('course_fee') discounted_fee = request.form.get('discounted_fee', '0') duration_days = request.form.get('duration') # in days status = request.form.get('status') position = request.form.get('position', '0') course_image_file = request.files.get('course_image') # optional # ---- resolve/validate hierarchy ---- exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" elif exam and getattr(subexam, 'exam_id', None) != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" subject = VideoCourseSubject.objects(id=subject_id).first() if subject_id else None if not subject: errors['subject_id'] = "Select subject" else: if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" # Language language_ref = Language.objects(id=language_id).first() if language_id else None if not language_ref: errors['language_id'] = "Select language" chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id).first() if not chapter: errors['chapter_id'] = "Select valid chapter" else: if subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" topic = None if topic_id: topic = Topic.objects(id=topic_id).first() if not topic: errors['topic_id'] = "Select valid topic" else: if subject and topic.subject_id != subject: errors['topic_id'] = "Topic does not belong to selected subject" if chapter and topic.chapter_id != chapter: errors['topic_id'] = "Topic does not belong to selected chapter" # ---- course fields validation ---- if not course_title or not course_title.strip(): errors['course_title'] = "Enter course title" # numbers course_fee_int = None discounted_fee_int = 0 duration_int = None position_int = 0 if not course_fee: errors['course_fee'] = "Enter course fee" else: try: course_fee_int = int(course_fee) if course_fee_int < 0: errors['course_fee'] = "Course fee must be non-negative" except ValueError: errors['course_fee'] = "Course fee must be numeric" if discounted_fee is not None and discounted_fee != '': try: discounted_fee_int = int(discounted_fee) if discounted_fee_int < 0: errors['discounted_fee'] = "Discounted fee must be non-negative" except ValueError: errors['discounted_fee'] = "Discounted fee must be numeric" if not duration_days: errors['duration'] = "Enter duration (days)" else: try: duration_int = int(duration_days) if duration_int <= 0: errors['duration'] = "Duration must be positive" except ValueError: errors['duration'] = "Duration must be numeric" if not status: errors['status'] = "Select status" else: if status not in ('0', '1'): errors['status'] = "Invalid status" if position is not None and position != '': try: position_int = int(position) except ValueError: errors['position'] = "Position must be numeric" # image optional; if provided, validate image_url = None if course_image_file and course_image_file.filename: if not allowed_course_image(course_image_file.filename): errors['course_image'] = "Only JPG, JPEG, PNG, WEBP allowed" else: course_image_file.seek(0, os.SEEK_END) size_bytes = course_image_file.tell() course_image_file.seek(0) if size_bytes > MAX_COURSE_IMAGE_SIZE: errors['course_image'] = "Image must be less than 5 MB" # ---------- if validation failed ---------- if errors: if is_ajax(): # for AJAX: send JSON, do NOT render template return jsonify(success=False, errors=errors, message=general_error or "Validation error"), 400 # non-AJAX: old behaviour subexams = [] if request.form.get('exam_id') and 'Exam' in globals() and 'SubExam' in globals(): try: ex = Exam.objects(id=request.form.get('exam_id')).first() if ex: subexams = (SubExam.objects(exam_id=ex, status=1) .order_by('position', '-created_date')) except Exception: subexams = [] return render_template( 'user/add-course.html', exams=exams, languages=languages, subexams=subexams, form_data=request.form, errors=errors, general_error=general_error, courses=[] ) # ---------- save image if present ---------- if course_image_file and course_image_file.filename: try: filename = secure_filename(course_image_file.filename) upload_folder = os.path.join('static', 'uploads', 'courses') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) course_image_file.save(save_path) base_url = request.host_url.rstrip('/') rel = save_path.replace(os.sep, '/') idx = rel.find("static/") rel_url = '/' + rel[idx:] if idx != -1 else '/' + rel image_url = f"{base_url}{rel_url}" except Exception: general_error = "Image upload failed" if is_ajax(): return jsonify(success=False, errors=errors, message=general_error), 500 return render_template( 'user/add-course.html', exams=exams, languages=languages, subexams=[], form_data=request.form, errors=errors, general_error=general_error, courses=[] ) # ---------- create record ---------- try: course_obj = Course( exam_id=exam, sub_exam_id=subexam, subject_id=subject, chapter_id=chapter, topic_id=topic, course_image=image_url, language_id=language_ref, course_title=course_title.strip(), course_fee=course_fee_int, discounted_fee=discounted_fee_int or 0, duration=duration_int, status=int(status), position=position_int, created_date=datetime.utcnow() ) course_obj.save() except Exception: general_error = "Something went wrong while saving" if is_ajax(): return jsonify(success=False, errors={}, message=general_error), 500 # non-AJAX fallback subexams = [] if request.form.get('exam_id') and 'Exam' in globals() and 'SubExam' in globals(): try: ex = Exam.objects(id=request.form.get('exam_id')).first() if ex: subexams = (SubExam.objects(exam_id=ex, status=1) .order_by('position', '-created_date')) except Exception: subexams = [] return render_template( 'user/add-course.html', exams=exams, languages=languages, subexams=subexams, form_data=request.form, errors=errors, general_error=general_error, courses=[] ) # ---------- success ---------- if is_ajax(): # THIS is what your JS expects return jsonify( success=True, course={ "id": str(course_obj.id), "title": course_obj.course_title, } ) # non-AJAX: keep old behaviour flash("Course added successfully", "success") return redirect(url_for('user.add_course')) @user_bp.route('/user/view-courses', methods=['GET']) def courses_list(): if 'user' not in session: return redirect(url_for('user.user_login')) q = (request.args.get('q') or '').strip() selected_exam_id = (request.args.get('exam_id') or '').strip() selected_subexam_id = (request.args.get('subexam_id') or request.args.get('sub_exam_id') or '').strip() selected_subject_id = (request.args.get('subject_id') or '').strip() selected_chapter_id = (request.args.get('chapter_id') or '').strip() selected_topic_id = (request.args.get('topic_id') or '').strip() selected_language_id = (request.args.get('language_id') or '').strip() # NEW status = (request.args.get('status') or '').strip() qry = Course.objects if q: qry = qry.filter(course_title__icontains=q) if selected_exam_id: try: qry = qry.filter(exam_id=Exam.objects.get(id=selected_exam_id)) except Exception: pass if selected_subexam_id: try: qry = qry.filter(sub_exam_id=SubExam.objects.get(id=selected_subexam_id)) except Exception: pass if selected_subject_id: try: qry = qry.filter(subject_id=VideoCourseSubject.objects.get(id=selected_subject_id)) except Exception: pass if selected_chapter_id: try: qry = qry.filter(chapter_id=Chapters.objects.get(id=selected_chapter_id)) except Exception: pass if selected_topic_id: try: qry = qry.filter(topic_id=Topic.objects.get(id=selected_topic_id)) except Exception: pass # Language filter (same pattern as View Old Papers) if selected_language_id: try: qry = qry.filter(language_id=Language.objects.get(id=selected_language_id)) except Exception: pass if status in ('0', '1'): try: qry = qry.filter(status=int(status)) except Exception: pass courses = qry.order_by('-created_date') rows = [] for c in courses: try: exam = c.exam_id except Exception: exam = None try: subexam = c.sub_exam_id except Exception: subexam = None try: subject = c.subject_id except Exception: subject = None try: chapter = c.chapter_id except Exception: chapter = None try: topic = c.topic_id except Exception: topic = None try: lang = c.language_id # NEW: reference to Language except Exception: lang = None rows.append({ 'id': str(c.id), 'course_title': c.course_title, 'exam_title': getattr(exam, 'exam_title', '-') if exam else '-', 'subexam_title': getattr(subexam, 'sub_exam_title', '-') if subexam else '-', 'subject_name': getattr(subject, 'subject_name', '-') if subject else '-', 'chapter_title': getattr(chapter, 'title', '-') if chapter else '-', 'topic_title': getattr(topic, 'title', '-') if topic else '-', 'language_name': getattr(lang, 'language_name', '-') if lang else '-', # NEW 'course_fee': c.course_fee, 'discounted_fee': c.discounted_fee or 0, 'duration': c.duration, # days 'status': c.status, 'image_path': c.course_image, 'created_iso': c.created_date.isoformat() if getattr(c, 'created_date', None) else '', 'created_pretty': c.created_date.strftime('%Y-%m-%d %H:%M') if getattr(c, 'created_date', None) else '-' }) exams = Exam.objects(status=1).order_by('exam_title').only('id', 'exam_title') languages = Language.objects(status=1).order_by('language_name').only('id', 'language_name') # NEW filters = { 'q': q, 'exam_id': selected_exam_id, 'subexam_id': selected_subexam_id, 'subject_id': selected_subject_id, 'chapter_id': selected_chapter_id, 'topic_id': selected_topic_id, 'language_id': selected_language_id, # NEW 'status': status } return render_template( 'user/view-courses.html', rows=rows, exams=exams, languages=languages, # NEW filters=filters ) @user_bp.route('/user/edit-course/', methods=['GET', 'POST']) def edit_course(course_id): if 'user' not in session: return redirect(url_for('user.user_login')) course = Course.objects(id=course_id).first() if not course: flash("Course not found", "danger") return redirect(url_for('user.courses_list')) exams = Exam.objects(status=1).order_by('exam_title') languages = Language.objects(status=1).only('id', 'language_name').order_by('language_name') errors = {} general_error = None # current refs try: current_exam = course.exam_id except DoesNotExist: current_exam = None except Exception: current_exam = None try: current_subexam = course.sub_exam_id except DoesNotExist: current_subexam = None except Exception: current_subexam = None try: current_subject = course.subject_id except DoesNotExist: current_subject = None except Exception: current_subject = None try: current_chapter = course.chapter_id except DoesNotExist: current_chapter = None except Exception: current_chapter = None try: current_topic = course.topic_id except DoesNotExist: current_topic = None except Exception: current_topic = None try: current_language = course.language_id except DoesNotExist: current_language = None except Exception: current_language = None pre_exam_id = str(current_exam.id) if current_exam else '' pre_sub_id = str(current_subexam.id) if current_subexam else '' pre_subject_id = str(current_subject.id) if current_subject else '' pre_chapter_id = str(current_chapter.id) if current_chapter else '' pre_topic_id = str(current_topic.id) if current_topic else '' pre_language_id = str(current_language.id) if current_language else '' if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') topic_id = request.form.get('topic_id') course_title = request.form.get('course_title') language_id = request.form.get('language_id') course_fee = request.form.get('course_fee') discounted_fee = request.form.get('discounted_fee', '0') duration_days = request.form.get('duration') status = request.form.get('status') position = request.form.get('position', '0') course_image_file = request.files.get('course_image') # optional # hierarchy checks exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" elif exam and getattr(subexam, 'exam_id', None) != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" subject = VideoCourseSubject.objects(id=subject_id).first() if subject_id else None if not subject: errors['subject_id'] = "Select subject" else: if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id).first() if not chapter: errors['chapter_id'] = "Select valid chapter" else: if subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" topic = None if topic_id: topic = Topic.objects(id=topic_id).first() if not topic: errors['topic_id'] = "Select valid topic" else: if subject and topic.subject_id != subject: errors['topic_id'] = "Topic does not belong to selected subject" if chapter and topic.chapter_id != chapter: errors['topic_id'] = "Topic does not belong to selected chapter" if not course_title or not course_title.strip(): errors['course_title'] = "Enter course title" # language required (Language reference) language_ref = Language.objects(id=language_id).first() if language_id else None if not language_ref: errors['language_id'] = "Select language" # numbers course_fee_int = None discounted_fee_int = 0 duration_int = None position_int = 0 if not course_fee: errors['course_fee'] = "Enter course fee" else: try: course_fee_int = int(course_fee) if course_fee_int < 0: errors['course_fee'] = "Course fee must be non-negative" except ValueError: errors['course_fee'] = "Course fee must be numeric" if discounted_fee is not None and discounted_fee != '': try: discounted_fee_int = int(discounted_fee) if discounted_fee_int < 0: errors['discounted_fee'] = "Discounted fee must be non-negative" except ValueError: errors['discounted_fee'] = "Discounted fee must be numeric" if not duration_days: errors['duration'] = "Enter duration (days)" else: try: duration_int = int(duration_days) if duration_int <= 0: errors['duration'] = "Duration must be positive" except ValueError: errors['duration'] = "Duration must be numeric" if not status: errors['status'] = "Select status" else: if status not in ('0', '1'): errors['status'] = "Invalid status" if position is not None and position != '': try: position_int = int(position) except ValueError: errors['position'] = "Position must be numeric" # image optional; if given validate image_url = course.course_image if course_image_file and course_image_file.filename: if not allowed_course_image(course_image_file.filename): errors['course_image'] = "Only JPG, JPEG, PNG, WEBP allowed" else: course_image_file.seek(0, os.SEEK_END) sz = course_image_file.tell() course_image_file.seek(0) if sz > MAX_COURSE_IMAGE_SIZE: errors['course_image'] = "Image must be less than 5 MB" if errors: return render_template( 'user/edit-course.html', exams=exams, languages=languages, course=course, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) # save new image if provided if course_image_file and course_image_file.filename: try: filename = secure_filename(course_image_file.filename) upload_folder = os.path.join('static', 'uploads', 'courses') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) course_image_file.save(save_path) base_url = request.host_url.rstrip('/') rel = save_path.replace(os.sep, '/') idx = rel.find("static/") rel_url = '/' + rel[idx:] if idx != -1 else '/' + rel image_url = f"{base_url}{rel_url}" except Exception: general_error = "Image upload failed" return render_template( 'user/edit-course.html', exams=exams, languages=languages, course=course, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) # update try: course.update( set__exam_id=exam, set__sub_exam_id=subexam, set__subject_id=subject, set__chapter_id=chapter, set__topic_id=topic, set__language_id=language_ref, # Language reference set__course_image=image_url, set__course_title=course_title.strip(), set__course_fee=course_fee_int, set__discounted_fee=discounted_fee_int or 0, set__duration=duration_int, set__status=int(status), set__position=position_int ) course.reload() # refresh pre* ids current_exam = exam current_subexam = subexam current_subject = subject current_chapter = chapter current_topic = topic current_language = language_ref pre_exam_id = str(current_exam.id) if current_exam else '' pre_sub_id = str(current_subexam.id) if current_subexam else '' pre_subject_id = str(current_subject.id) if current_subject else '' pre_chapter_id = str(current_chapter.id) if current_chapter else '' pre_topic_id = str(current_topic.id) if current_topic else '' pre_language_id = str(current_language.id) if current_language else '' flash("Course updated successfully", "success") return redirect(url_for('user.edit_course', course_id=course.id)) except Exception as e: general_error = f"Save failed: {type(e).__name__}: {e}" return render_template( 'user/edit-course.html', exams=exams, languages=languages, course=course, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) # GET return render_template( 'user/edit-course.html', exams=exams, languages=languages, course=course, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) @user_bp.route('/user/delete-course/', methods=['POST']) def delete_course(course_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 c = Course.objects(id=course_id).first() if not c: return jsonify(success=False, message="Course not found"), 404 try: c.delete() return jsonify(success=True, message="Course deleted") except Exception: return jsonify(success=False, message="Delete failed"), 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') languages = Language.objects(status=1).only('id','language_name').order_by('language_name') # NEW if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') or None subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') or None topic_id = request.form.get('topic_id') or None language_id = request.form.get('language_id') # NEW title = (request.form.get('title') or '').strip() author = (request.form.get('author') or '').strip() status_val = request.form.get('status') pdf_file = request.files.get('file') # validations if not exam_id: errors['exam_id'] = "Select exam" # if not subject_id: # errors['subject_id'] = "Select subject" if not language_id: errors['language_id'] = "Select language" # NEW if not title: errors['title'] = "Enter title" if not author: errors['author'] = "Enter author" if status_val not in ('0', '1'): errors['status'] = "Select status" if not pdf_file or not pdf_file.filename: errors['file'] = "Upload PDF" elif not allowed_pdf(pdf_file.filename): errors['file'] = "Only PDF files are allowed" else: pdf_file.seek(0, os.SEEK_END) if pdf_file.tell() > PDF_MAX_SIZE: errors['file'] = "PDF must be < 10 MB" pdf_file.seek(0) if errors: return render_template('user/add-ebook.html', exams=exams, languages=languages, # NEW errors=errors, request=request) exam = Exam.objects(id=exam_id).first() if not exam: errors['exam_id'] = "Invalid exam" return render_template('user/add-ebook.html', exams=exams, languages=languages, # NEW errors=errors, request=request) sub_exam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None # subject must belong to exam (and optionally sub_exam if given) # subj_q = dict(id=subject_id, exam_id=exam) # if sub_exam: # subj_q['sub_exam_id'] = sub_exam # subject = VideoCourseSubject.objects(**subj_q).first() # if not subject: # errors['subject_id'] = "Invalid subject for selection" # return render_template('user/add-ebook.html', # exams=exams, # languages=languages, # NEW # errors=errors, # request=request) subject = None if subject_id: # subject must belong to exam (and optionally sub_exam if given) subj_q = dict(id=subject_id, exam_id=exam) if sub_exam: subj_q['sub_exam_id'] = sub_exam subject = VideoCourseSubject.objects(**subj_q).first() if not subject: errors['subject_id'] = "Invalid subject for selection" return render_template( 'user/add-ebook.html', exams=exams, languages=languages, # NEW errors=errors, request=request ) # chapter (optional) must belong to exam + subject chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id, exam_id=exam, subject_id=subject).first() if not chapter: errors['chapter_id'] = "Invalid chapter for selection" return render_template('user/add-ebook.html', exams=exams, languages=languages, # NEW errors=errors, request=request) # topic (optional) must belong to exam + subject (+ chapter if provided) topic = None if topic_id: t_q = dict(id=topic_id, exam_id=exam, subject_id=subject) if chapter: t_q['chapter_id'] = chapter topic = Topic.objects(**t_q).first() if not topic: errors['topic_id'] = "Invalid topic for selection" return render_template('user/add-ebook.html', exams=exams, languages=languages, # NEW errors=errors, request=request) # language (required) language_ref = Language.objects(id=language_id).first() if language_id else None # NEW if not language_ref: errors['language_id'] = "Invalid language selection" return render_template('user/add-ebook.html', exams=exams, languages=languages, # NEW errors=errors, request=request) # save file try: filename = secure_filename(pdf_file.filename) upload_folder = os.path.join('static', 'uploads', 'ebooks') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) pdf_file.save(save_path) rel = save_path.replace(os.sep, '/') idx = rel.find('static/') file_url = '/' + rel[idx:] if idx != -1 else '/' + rel except Exception: general_error = "Failed to store file" return render_template('user/add-ebook.html', exams=exams, languages=languages, # NEW general_error=general_error, request=request) # create Ebook( exam_id=exam, sub_exam_id=sub_exam, subject_id=subject, chapter_id=chapter, topic_id=topic, language_id=language_ref, # NEW course_id=None, title=title, author=author, file_path=file_url, status=int(status_val) ).save() flash("E-Book added successfully", "success") return redirect(url_for('user.add_ebook')) return render_template('user/add-ebook.html', exams=exams, languages=languages) # NEW @user_bp.route('/user/view-ebooks', methods=['GET']) def ebooks_list(): if 'user' not in session: return redirect(url_for('user.user_login')) # Filters (all optional) exam_id = request.args.get('exam_id') or None sub_exam_id = request.args.get('subexam_id') or request.args.get('sub_exam_id') or None # accept both subject_id = request.args.get('subject_id') or None chapter_id = request.args.get('chapter_id') or None topic_id = request.args.get('topic_id') or None language_id = request.args.get('language_id') or None # NEW status = request.args.get('status') # '1' or '0' or None q = {} if exam_id: q['exam_id'] = exam_id if sub_exam_id: q['sub_exam_id'] = sub_exam_id if subject_id: q['subject_id'] = subject_id if chapter_id: q['chapter_id'] = chapter_id if topic_id: q['topic_id'] = topic_id if language_id: q['language_id'] = language_id # NEW if status in ('0', '1'): q['status'] = int(status) # Pull matching ebooks (don’t deref in template) ebooks = Ebook.objects(**q).no_dereference().order_by('-created_date') # Collect IDs to bulk fetch exam_ids = set() subexam_ids = set() subject_ids = set() chapter_ids = set() topic_ids = set() language_ids = set() # NEW for eb in ebooks: if getattr(eb, 'exam_id', None): exam_ids.add(_as_str(eb.exam_id.id)) if getattr(eb, 'sub_exam_id', None): subexam_ids.add(_as_str(eb.sub_exam_id.id)) if getattr(eb, 'subject_id', None): subject_ids.add(_as_str(eb.subject_id.id)) if getattr(eb, 'chapter_id', None): chapter_ids.add(_as_str(eb.chapter_id.id)) if getattr(eb, 'topic_id', None): topic_ids.add(_as_str(eb.topic_id.id)) if getattr(eb, 'language_id', None): language_ids.add(_as_str(eb.language_id.id)) # NEW # Bulk maps (id -> title/name); guard for dangling refs exam_map, subexam_map, subject_map, chapter_map, topic_map, language_map = {}, {}, {}, {}, {}, {} if exam_ids: for ex in Exam.objects(id__in=list(exam_ids)).only('id', 'exam_title'): exam_map[_as_str(ex.id)] = ex.exam_title if subexam_ids: for sx in SubExam.objects(id__in=list(subexam_ids)).only('id', 'sub_exam_title'): subexam_map[_as_str(sx.id)] = sx.sub_exam_title if subject_ids: for sj in VideoCourseSubject.objects(id__in=list(subject_ids)).only('id', 'subject_name'): subject_map[_as_str(sj.id)] = sj.subject_name if chapter_ids: for ch in Chapters.objects(id__in=list(chapter_ids)).only('id', 'title'): chapter_map[_as_str(ch.id)] = ch.title if topic_ids: for tp in Topic.objects(id__in=list(topic_ids)).only('id', 'title'): topic_map[_as_str(tp.id)] = tp.title if language_ids: # NEW for lg in Language.objects(id__in=list(language_ids)).only('id', 'language_name'): language_map[_as_str(lg.id)] = lg.language_name # Build rows for template (strings only) view_rows = [] for eb in ebooks: ex_id = _as_str(getattr(getattr(eb, 'exam_id', None), 'id', None)) sx_id = _as_str(getattr(getattr(eb, 'sub_exam_id', None), 'id', None)) sj_id = _as_str(getattr(getattr(eb, 'subject_id', None), 'id', None)) ch_id = _as_str(getattr(getattr(eb, 'chapter_id', None), 'id', None)) tp_id = _as_str(getattr(getattr(eb, 'topic_id', None), 'id', None)) lg_id = _as_str(getattr(getattr(eb, 'language_id', None), 'id', None)) # NEW exam_title = exam_map.get(ex_id) if ex_id in exam_map else ('— (missing)' if ex_id else '-') sub_exam_title = subexam_map.get(sx_id) if sx_id in subexam_map else ('— (missing)' if sx_id else '-') subject_name = subject_map.get(sj_id) if sj_id in subject_map else ('— (missing)' if sj_id else '-') chapter_title = chapter_map.get(ch_id) if ch_id in chapter_map else ('— (missing)' if ch_id else '-') topic_title = topic_map.get(tp_id) if tp_id in topic_map else ('— (missing)' if tp_id else '-') language_name = language_map.get(lg_id) if lg_id in language_map else ('— (missing)' if lg_id else '-') # NEW view_rows.append({ 'id': _as_str(eb.id), 'title': eb.title or '-', 'author': eb.author or '-', 'exam_title': exam_title, 'sub_exam_title': sub_exam_title, 'subject_name': subject_name, 'chapter_title': chapter_title, 'topic_title': topic_title, 'language_name': language_name, # NEW 'file_path': getattr(eb, 'file_path', None), 'publication_date': eb.publication_date.strftime('%Y-%m-%d') if getattr(eb, 'publication_date', None) else '-', 'status': getattr(eb, 'status', 0), 'created_date': eb.created_date.strftime('%Y-%m-%d %H:%M') if getattr(eb, 'created_date', None) else '-', }) # Exams + Languages for the filter bar exams = Exam.objects(status=1).order_by('exam_title').only('id', 'exam_title') languages = Language.objects(status=1).order_by('language_name').only('id', 'language_name') # NEW return render_template( 'user/view-ebooks.html', view_rows=view_rows, exams=exams, languages=languages # NEW ) @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')) eb = Ebook.objects(id=ebook_id).first() if not eb: flash("E-Book not found", "danger") return redirect(url_for('user.ebooks_list')) exams = Exam.objects(status=1).order_by('exam_title') languages = Language.objects(status=1).order_by('language_name').only('id', 'language_name') # NEW errors, general_error = {}, None if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') or None subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') or None topic_id = request.form.get('topic_id') or None language_id = request.form.get('language_id') # NEW title = (request.form.get('title') or '').strip() author = (request.form.get('author') or '').strip() status_val = request.form.get('status') pdf_file = request.files.get('file') # optional if not exam_id: errors['exam_id'] = "Select exam" # if not subject_id: # errors['subject_id'] = "Select subject" if not title: errors['title'] = "Enter title" if not author: errors['author'] = "Enter author" if status_val not in ('0', '1'): errors['status'] = "Select status" if not language_id: errors['language_id'] = "Select language" # NEW if pdf_file and pdf_file.filename: if not allowed_pdf(pdf_file.filename): errors['file'] = "Only PDF files are allowed" else: pdf_file.seek(0, os.SEEK_END) if pdf_file.tell() > PDF_MAX_SIZE: errors['file'] = "PDF must be < 10 MB" pdf_file.seek(0) if errors: return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, request=request) exam = Exam.objects(id=exam_id).first() if not exam: errors['exam_id'] = "Invalid exam" return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, request=request) sub_exam = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None # subj_q = dict(id=subject_id, exam_id=exam) # if sub_exam: # subj_q['sub_exam_id'] = sub_exam # subject = VideoCourseSubject.objects(**subj_q).first() # if not subject: # errors['subject_id'] = "Invalid subject for selection" # return render_template('user/edit-ebook.html', # ebook=eb, exams=exams, languages=languages, # errors=errors, request=request) subject = None if subject_id: # subject must belong to exam (and optionally sub_exam if given) subj_q = dict(id=subject_id, exam_id=exam) if sub_exam: subj_q['sub_exam_id'] = sub_exam subject = VideoCourseSubject.objects(**subj_q).first() if not subject: errors['subject_id'] = "Invalid subject for selection" return render_template( 'user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, request=request ) chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id, exam_id=exam, subject_id=subject).first() if not chapter: errors['chapter_id'] = "Invalid chapter for selection" return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, request=request) topic = None if topic_id: t_q = dict(id=topic_id, exam_id=exam, subject_id=subject) if chapter: t_q['chapter_id'] = chapter topic = Topic.objects(**t_q).first() if not topic: errors['topic_id'] = "Invalid topic for selection" return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, request=request) # Language validation language_ref = Language.objects(id=language_id, status=1).first() if not language_ref: errors['language_id'] = "Invalid language" return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, request=request) # save file if new one provided file_url = eb.file_path if pdf_file and pdf_file.filename: try: filename = secure_filename(pdf_file.filename) upload_folder = os.path.join('static', 'uploads', 'ebooks') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) pdf_file.save(save_path) rel = save_path.replace(os.sep, '/') idx = rel.find('static/') file_url = '/' + rel[idx:] if idx != -1 else '/' + rel except Exception: general_error = "Failed to store file" return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, general_error=general_error, request=request) eb.update( set__exam_id=exam, set__sub_exam_id=sub_exam, set__subject_id=subject, set__chapter_id=chapter, set__topic_id=topic, set__language_id=language_ref, # NEW set__title=title, set__author=author, set__file_path=file_url, set__status=int(status_val), ) eb.reload() flash("E-Book updated successfully", "success") return redirect(url_for('user.edit_ebook', ebook_id=eb.id)) # GET return render_template('user/edit-ebook.html', ebook=eb, exams=exams, languages=languages, errors=errors, general_error=general_error, request=request) @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') languages = Language.objects(status=1).only('id','language_name').order_by('language_name') # NEW if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') # optional subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') # optional topic_id = request.form.get('topic_id') # optional language_id = request.form.get('language_id') # NEW (required) 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') # --- resolve / validate hierarchy --- # Exam (required) exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" # Subexam (optional) subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" elif exam and getattr(subexam, 'exam_id', None) != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" # Subject (required) subject = None if subject_id: subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: errors['subject_id'] = "Select valid subject" else: if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" # Chapter (optional) chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id).first() if not chapter: errors['chapter_id'] = "Select valid chapter" elif subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" # Topic (optional) topic = None if topic_id: topic = Topic.objects(id=topic_id).first() if not topic: errors['topic_id'] = "Select valid topic" else: if subject and topic.subject_id != subject: errors['topic_id'] = "Topic does not belong to selected subject" if chapter and topic.chapter_id != chapter: errors['topic_id'] = "Topic does not belong to selected chapter" # Language (required) language_ref = Language.objects(id=language_id).first() if language_id else None # NEW if not language_ref: errors['language_id'] = "Select language" # NEW # Other field validation if not paper_title or not paper_title.strip(): errors['paper_title'] = "Enter paper title" # duration numeric duration_int = None if not duration: errors['duration'] = "Enter duration" else: try: duration_int = int(duration) if duration_int <= 0: errors['duration'] = "Duration must be positive" except ValueError: errors['duration'] = "Duration must be numeric" # number_of_questions numeric noq_int = None if not number_of_questions: errors['number_of_questions'] = "Enter number of questions" else: try: noq_int = int(number_of_questions) if noq_int <= 0: errors['number_of_questions'] = "Questions must be positive" except ValueError: errors['number_of_questions'] = "Questions must be numeric" if not status: errors['status'] = "Select status" # PDF validation file_url = None if not content_file or content_file.filename == '': errors['content_file'] = "Upload paper file (PDF)" else: if not allowed_paper(content_file.filename): errors['content_file'] = "Only PDF allowed" else: content_file.seek(0, os.SEEK_END) size_bytes = content_file.tell() content_file.seek(0) if size_bytes > MAX_PAPER_FILE_SIZE: errors['content_file'] = "File must be less than 5 MB" # SHORT-CIRCUIT IF ERRORS if errors: return render_template( 'user/add-old-paper.html', exams=exams, languages=languages, # NEW errors=errors, general_error=general_error, request=request ) # save file try: filename = secure_filename(content_file.filename) upload_folder = os.path.join('static', 'uploads', 'old_papers') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) content_file.save(save_path) base_url = request.host_url.rstrip('/') rel = save_path.replace(os.sep, '/') idx = rel.find("static/") if idx != -1: rel_url = '/' + rel[idx:] else: rel_url = '/' + rel file_url = f"{base_url}{rel_url}" except Exception as e: general_error = "File upload failed" return render_template( 'user/add-old-paper.html', exams=exams, languages=languages, # NEW errors=errors, general_error=general_error, request=request ) # save record try: OldPapers( exam_id=exam, sub_exam_id=subexam, subject_id=subject, chapter_id=chapter, topic_id=topic, language_id=language_ref, # NEW course_id=None, paper_title=paper_title.strip(), duration=duration_int, number_of_questions=noq_int, content_file_path=file_url, status=int(status), created_date=datetime.utcnow() ).save() except Exception: general_error = "Something went wrong while saving" return render_template( 'user/add-old-paper.html', exams=exams, languages=languages, # NEW errors=errors, general_error=general_error, request=request ) flash("Old paper added successfully", "success") return redirect(url_for('user.add_old_paper')) # GET return render_template('user/add-old-paper.html', exams=exams, languages=languages) # NEW @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')) # ---- read filter params (all optional) ---- q = (request.args.get('q') or '').strip() selected_exam_id = (request.args.get('exam_id') or '').strip() selected_subexam_id = (request.args.get('subexam_id') or request.args.get('sub_exam_id') or '').strip() selected_subject_id = (request.args.get('subject_id') or '').strip() selected_chapter_id = (request.args.get('chapter_id') or '').strip() selected_topic_id = (request.args.get('topic_id') or '').strip() selected_language_id = (request.args.get('language_id') or '').strip() # NEW status = (request.args.get('status') or '').strip() # '1', '0', or '' qry = OldPapers.objects # ---- search ---- if q: qry = qry.filter(paper_title__icontains=q) # ---- apply filters safely (skip if lookup fails) ---- if selected_exam_id: try: qry = qry.filter(exam_id=Exam.objects.get(id=selected_exam_id)) except Exception: pass if selected_subexam_id: try: qry = qry.filter(sub_exam_id=SubExam.objects.get(id=selected_subexam_id)) except Exception: pass if selected_subject_id: try: qry = qry.filter(subject_id=VideoCourseSubject.objects.get(id=selected_subject_id)) except Exception: pass if selected_chapter_id: try: qry = qry.filter(chapter_id=Chapters.objects.get(id=selected_chapter_id)) except Exception: pass if selected_topic_id: try: qry = qry.filter(topic_id=Topic.objects.get(id=selected_topic_id)) except Exception: pass if selected_language_id: # NEW try: qry = qry.filter(language_id=Language.objects.get(id=selected_language_id)) except Exception: pass # ---- status filter ---- if status in ('0', '1'): try: qry = qry.filter(status=int(status)) except Exception: pass papers = qry.order_by('-created_date') rows = [] for p in papers: # Safe deref (in case any ref was deleted) try: exam = p.exam_id except Exception: exam = None try: subexam = p.sub_exam_id except Exception: subexam = None try: subject = p.subject_id except Exception: subject = None try: chapter = p.chapter_id except Exception: chapter = None try: topic = p.topic_id except Exception: topic = None try: lang = p.language_id except Exception: lang = None # NEW rows.append({ 'id': str(p.id), 'paper_title': p.paper_title, 'exam_title': getattr(exam, 'exam_title', '-') if exam else '-', 'subexam_title': getattr(subexam, 'sub_exam_title', '-') if subexam else '-', 'subject_name': getattr(subject, 'subject_name', '-') if subject else '-', 'chapter_title': getattr(chapter, 'title', '-') if chapter else '-', 'topic_title': getattr(topic, 'title', '-') if topic else '-', 'language_name': getattr(lang, 'language_name', '-') if lang else '-', # NEW 'duration': p.duration, 'number_of_questions': p.number_of_questions, 'status': p.status, 'file_path': getattr(p, 'content_file_path', None), 'created_iso': p.created_date.isoformat() if getattr(p, 'created_date', None) else '', 'created_pretty': p.created_date.strftime('%Y-%m-%d %H:%M') if getattr(p, 'created_date', None) else '-' }) exams = Exam.objects(status=1).order_by('exam_title').only('id', 'exam_title') languages = Language.objects(status=1).order_by('language_name').only('id', 'language_name') # NEW # ---- build filters dict for the template (matches `filters.*` usage) ---- filters = { 'q': q, 'exam_id': selected_exam_id, 'subexam_id': selected_subexam_id, 'subject_id': selected_subject_id, 'chapter_id': selected_chapter_id, 'topic_id': selected_topic_id, 'language_id': selected_language_id, # NEW 'status': status } return render_template( 'user/view-old-papers.html', rows=rows, exams=exams, languages=languages, # NEW filters=filters ) @user_bp.route('/user/edit-old-paper/', methods=['GET', 'POST']) def edit_old_paper(paper_id): if 'user' not in session: return redirect(url_for('user.user_login')) paper = OldPapers.objects(id=paper_id).first() if not paper: flash("Old paper not found", "danger") return redirect(url_for('user.old_papers_list')) exams = Exam.objects(status=1).order_by('exam_title') languages = Language.objects(status=1).only('id','language_name').order_by('language_name') # NEW errors = {} general_error = None # current refs (safe deref) try: current_exam = paper.exam_id except Exception: current_exam = None try: current_subexam = paper.sub_exam_id except Exception: current_subexam = None try: current_subject = paper.subject_id except Exception: current_subject = None try: current_chapter = paper.chapter_id except Exception: current_chapter = None try: current_topic = paper.topic_id except Exception: current_topic = None try: current_language = paper.language_id # NEW except Exception: current_language = None # NEW pre_exam_id = str(current_exam.id) if current_exam else '' pre_sub_id = str(current_subexam.id) if current_subexam else '' pre_subject_id = str(current_subject.id) if current_subject else '' pre_chapter_id = str(current_chapter.id) if current_chapter else '' pre_topic_id = str(current_topic.id) if current_topic else '' pre_language_id = str(current_language.id) if current_language else '' # NEW if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') topic_id = request.form.get('topic_id') language_id = request.form.get('language_id') # NEW 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') # optional on edit # exam required exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" # subexam optional subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" elif exam and getattr(subexam, 'exam_id', None) != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" # subject required subject = None if subject_id: subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: errors['subject_id'] = "Select valid subject" else: if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" # chapter optional chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id).first() if not chapter: errors['chapter_id'] = "Select valid chapter" else: if subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" # topic optional topic = None if topic_id: topic = Topic.objects(id=topic_id).first() if not topic: errors['topic_id'] = "Select valid topic" else: if subject and topic.subject_id != subject: errors['topic_id'] = "Topic does not belong to selected subject" if chapter and topic.chapter_id != chapter: errors['topic_id'] = "Topic does not belong to selected chapter" # language required language_ref = Language.objects(id=language_id).first() if language_id else None # NEW if not language_ref: errors['language_id'] = "Select language" # NEW # title if not paper_title or not paper_title.strip(): errors['paper_title'] = "Enter paper title" # duration numeric duration_int = None if not duration: errors['duration'] = "Enter duration" else: try: duration_int = int(duration) if duration_int <= 0: errors['duration'] = "Duration must be positive" except ValueError: errors['duration'] = "Duration must be numeric" # questions numeric noq_int = None if not number_of_questions: errors['number_of_questions'] = "Enter number of questions" else: try: noq_int = int(number_of_questions) if noq_int <= 0: errors['number_of_questions'] = "Questions must be positive" except ValueError: errors['number_of_questions'] = "Questions must be numeric" if not status: errors['status'] = "Select status" # file optional on edit, but if provided must pass validation file_url = paper.content_file_path if content_file and content_file.filename != '': if not allowed_paper(content_file.filename): errors['content_file'] = "Only PDF allowed" else: content_file.seek(0, os.SEEK_END) sz = content_file.tell() content_file.seek(0) if sz > MAX_PAPER_FILE_SIZE: errors['content_file'] = "File must be less than 5 MB" if errors: return render_template( 'user/edit-old-paper.html', exams=exams, languages=languages, # NEW paper=paper, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id # NEW ) # if new file uploaded, save it if content_file and content_file.filename != '': try: filename = secure_filename(content_file.filename) upload_folder = os.path.join('static', 'uploads', 'old_papers') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) content_file.save(save_path) base_url = request.host_url.rstrip('/') rel = save_path.replace(os.sep, '/') idx = rel.find("static/") if idx != -1: rel_url = '/' + rel[idx:] else: rel_url = '/' + rel file_url = f"{base_url}{rel_url}" except Exception as e: general_error = "File upload failed" return render_template( 'user/edit-old-paper.html', exams=exams, languages=languages, # NEW paper=paper, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id # NEW ) # update db try: paper.update( set__exam_id=exam, set__sub_exam_id=subexam, set__subject_id=subject, set__chapter_id=chapter, set__topic_id=topic, set__language_id=language_ref, # NEW set__course_id=None, # stays None set__paper_title=paper_title.strip(), set__duration=duration_int, set__number_of_questions=noq_int, set__content_file_path=file_url, set__status=int(status) ) paper.reload() # refresh pre_* values for the re-render view current_exam = exam current_subexam = subexam current_subject = subject current_chapter = chapter current_topic = topic current_language = language_ref # NEW pre_exam_id = str(current_exam.id) if current_exam else '' pre_sub_id = str(current_subexam.id) if current_subexam else '' pre_subject_id = str(current_subject.id) if current_subject else '' pre_chapter_id = str(current_chapter.id) if current_chapter else '' pre_topic_id = str(current_topic.id) if current_topic else '' pre_language_id = str(current_language.id) if current_language else '' # NEW flash("Old paper updated successfully", "success") return redirect(url_for('user.edit_old_paper', paper_id=paper.id)) except Exception as e: general_error = f"Save failed: {type(e).__name__}: {e}" return render_template( 'user/edit-old-paper.html', exams=exams, languages=languages, # NEW paper=paper, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id # NEW ) # GET return render_template( 'user/edit-old-paper.html', exams=exams, languages=languages, # NEW paper=paper, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id # NEW ) @user_bp.route('/user/delete-old-paper/', methods=['POST']) def delete_old_paper(paper_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 p = OldPapers.objects(id=paper_id).first() if not p: return jsonify(success=False, message="Old paper not found"), 404 try: p.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')) exams = Exam.objects(status=1).only('id', 'exam_title').order_by('exam_title') languages = Language.objects(status=1).only('id', 'language_name').order_by('language_name') # NEW errors = {} general_error = None if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') or None # SUBJECT OPTIONAL NOW subject_id = request.form.get('subject_id') or None chapter_id = request.form.get('chapter_id') or None topic_id = request.form.get('topic_id') or None language_id = request.form.get('language_id') # NEW title = request.form.get('title') duration = request.form.get('duration') total_questions = request.form.get('total_questions') total_marks = request.form.get('total_marks') status = request.form.get('status') instructions = request.form.get('instructions') # ---- validations (basic fields) ---- if not exam_id: errors['exam_id'] = "Select exam" # SUBJECT IS OPTIONAL NOW → removed this: # if not subject_id: # errors['subject_id'] = "Select subject" if not language_id: errors['language_id'] = "Select language" # NEW if not title or not title.strip(): errors['title'] = "Enter title" # duration validation duration_int = None try: duration_int = int(duration) if duration else None if duration_int is None or duration_int <= 0: errors['duration'] = "Enter valid duration (minutes)" except ValueError: errors['duration'] = "Duration must be numeric" # total_questions & total_marks validation for fld, lbl in [('total_questions', 'Total questions'), ('total_marks', 'Total marks')]: val = request.form.get(fld) try: ival = int(val) if val else None if ival is None or ival < 0: errors[fld] = f"Enter valid {lbl.lower()}" except ValueError: errors[fld] = f"{lbl} must be numeric" if not status: errors['status'] = "Select status" if not instructions or not instructions.strip(): errors['instructions'] = "Enter instructions" if errors: return render_template( 'user/add-mock-test.html', exams=exams, languages=languages, # NEW errors=errors, request=request, general_error=general_error ) # ---- resolve references (optional ones may be None) ---- exam_ref = Exam.objects(id=exam_id).first() if not exam_ref: errors['exam_id'] = "Invalid exam" sub_exam_ref = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None # SUBJECT (OPTIONAL) subject_ref = None if subject_id: subject_ref = VideoCourseSubject.objects(id=subject_id).first() if not subject_ref: errors['subject_id'] = "Invalid subject" else: # optional: ensure subject matches exam/subexam if subject_ref.exam_id != exam_ref: errors['subject_id'] = "Subject does not belong to selected exam" if sub_exam_ref and subject_ref.sub_exam_id != sub_exam_ref: errors['subject_id'] = "Subject does not belong to selected sub exam" # CHAPTER (OPTIONAL, but requires subject) chapter_ref = None if chapter_id: if not subject_ref: errors['chapter_id'] = "Select subject before choosing chapter" else: chapter_ref = Chapters.objects(id=chapter_id).first() if not chapter_ref: errors['chapter_id'] = "Invalid chapter" elif chapter_ref.subject_id != subject_ref: errors['chapter_id'] = "Chapter does not belong to selected subject" # TOPIC (OPTIONAL, but requires subject) topic_ref = None if topic_id: if not subject_ref: errors['topic_id'] = "Select subject before choosing topic" else: topic_ref = Topic.objects(id=topic_id).first() if not topic_ref: errors['topic_id'] = "Invalid topic" else: if topic_ref.subject_id != subject_ref: errors['topic_id'] = "Topic does not belong to selected subject" if chapter_ref and topic_ref.chapter_id != chapter_ref: errors['topic_id'] = "Topic does not belong to selected chapter" # LANGUAGE (REQUIRED) language_ref = Language.objects(id=language_id).first() if language_id else None # NEW if not language_ref: errors['language_id'] = "Invalid language" # NEW if errors: return render_template( 'user/add-mock-test.html', exams=exams, languages=languages, # NEW errors=errors, request=request ) try: MockTest( exam_id=exam_ref, sub_exam_id=sub_exam_ref, subject_id=subject_ref, # can be None now chapter_id=chapter_ref, # can be None topic_id=topic_ref, # can be None language_id=language_ref, # NEW title=title.strip(), duration=duration_int, total_questions=int(total_questions), total_marks=int(total_marks), status=int(status), instructions=instructions.strip(), created_date=datetime.utcnow() ).save() except Exception: general_error = "Something went wrong while saving" return render_template( 'user/add-mock-test.html', exams=exams, languages=languages, # NEW errors=errors, request=request, general_error=general_error ) flash("Mock test added successfully", "success") return redirect(url_for('user.add_mock_test')) # GET return render_template('user/add-mock-test.html', exams=exams, languages=languages) @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')) # ---- Filters ---- q_text = request.args.get('q', '').strip() exam_id = request.args.get('exam_id') or '' subexam_id = request.args.get('subexam_id') or request.args.get('sub_exam_id') or '' subject_id = request.args.get('subject_id') or '' chapter_id = request.args.get('chapter_id') or '' topic_id = request.args.get('topic_id') or '' language_id = request.args.get('language_id') or '' # NEW status = request.args.get('status') or '' # '1','0','' mq = {} if exam_id: mq['exam_id'] = exam_id if subexam_id: mq['sub_exam_id'] = subexam_id if subject_id: mq['subject_id'] = subject_id if chapter_id: mq['chapter_id'] = chapter_id if topic_id: mq['topic_id'] = topic_id if language_id: mq['language_id'] = language_id # NEW if status in ('0','1'): mq['status'] = int(status) if q_text: mq['title__icontains'] = q_text tests = MockTest.objects(**mq).no_dereference().order_by('-created_date') # ---- Build display rows ---- ex_ids, sx_ids, sj_ids, ch_ids, tp_ids, lang_ids = set(), set(), set(), set(), set(), set() for t in tests: if getattr(t, 'exam_id', None): ex_ids.add(_sid(t.exam_id)) if getattr(t, 'sub_exam_id', None): sx_ids.add(_sid(t.sub_exam_id)) if getattr(t, 'subject_id', None): sj_ids.add(_sid(t.subject_id)) if getattr(t, 'chapter_id', None): ch_ids.add(_sid(t.chapter_id)) if getattr(t, 'topic_id', None): tp_ids.add(_sid(t.topic_id)) if getattr(t, 'language_id', None): lang_ids.add(_sid(t.language_id)) # NEW exam_map = {_sid(x): x.exam_title for x in Exam.objects(id__in=list(ex_ids)).only('id','exam_title')} if ex_ids else {} sub_map = {_sid(x): x.sub_exam_title for x in SubExam.objects(id__in=list(sx_ids)).only('id','sub_exam_title')} if sx_ids else {} subj_map = {_sid(x): x.subject_name for x in VideoCourseSubject.objects(id__in=list(sj_ids)).only('id','subject_name')} if sj_ids else {} chap_map = {_sid(x): x.title for x in Chapters.objects(id__in=list(ch_ids)).only('id','title')} if ch_ids else {} topic_map = {_sid(x): x.title for x in Topic.objects(id__in=list(tp_ids)).only('id','title')} if tp_ids else {} lang_map = {_sid(x): x.language_name for x in Language.objects(id__in=list(lang_ids)).only('id','language_name')} if lang_ids else {} # NEW view_rows = [] for t in tests: view_rows.append({ 'id': _sid(t), 'title': t.title, 'exam_title': exam_map.get(_sid(getattr(t, 'exam_id', None)), '-') if getattr(t, 'exam_id', None) else '-', 'subexam_title': sub_map.get(_sid(getattr(t, 'sub_exam_id', None)), '-') if getattr(t, 'sub_exam_id', None) else '-', 'subject_name': subj_map.get(_sid(getattr(t, 'subject_id', None)), '-') if getattr(t, 'subject_id', None) else '-', 'chapter_title': chap_map.get(_sid(getattr(t, 'chapter_id', None)), '-') if getattr(t, 'chapter_id', None) else '-', 'topic_title': topic_map.get(_sid(getattr(t, 'topic_id', None)), '-') if getattr(t, 'topic_id', None) else '-', 'language_name': lang_map.get(_sid(getattr(t, 'language_id', None)), '-') if getattr(t, 'language_id', None) else '-', # NEW 'duration': t.duration, 'total_questions': t.total_questions, 'total_marks': getattr(t, 'total_marks', None) or '-', 'status': t.status, 'created_iso': t.created_date.isoformat() if getattr(t, 'created_date', None) else '', 'created_pretty': _fmt(t.created_date) if getattr(t, 'created_date', None) else '-', 'instructions': getattr(t, 'instructions', None) or '', }) exams = Exam.objects(status=1).only('id','exam_title').order_by('exam_title') languages = Language.objects(status=1).only('id','language_name').order_by('language_name') # NEW filters = { "q": q_text, "exam_id": exam_id, "subexam_id": subexam_id, "subject_id": subject_id, "chapter_id": chapter_id, "topic_id": topic_id, "language_id": language_id, # NEW "status": status, } return render_template('user/view-mock-tests.html', view_rows=view_rows, exams=exams, languages=languages, # NEW filters=filters) @user_bp.route('/user/edit-mock-test/', methods=['GET','POST']) def edit_mock_test(mock_id): if 'user' not in session: return redirect(url_for('user.user_login')) mt = MockTest.objects(id=mock_id).first() if not mt: flash("Mock test not found", "danger") return redirect(url_for('user.mock_tests_list')) exams = Exam.objects(status=1).only('id', 'exam_title').order_by('exam_title') languages = Language.objects(status=1).only('id','language_name').order_by('language_name') # NEW errors, general_error = {}, None if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') or None # SUBJECT OPTIONAL NOW subject_id = request.form.get('subject_id') or None chapter_id = request.form.get('chapter_id') or None topic_id = request.form.get('topic_id') or None language_id = request.form.get('language_id') # NEW title = request.form.get('title') duration = request.form.get('duration') total_questions = request.form.get('total_questions') total_marks = request.form.get('total_marks') status = request.form.get('status') instructions = request.form.get('instructions') # ---- basic validations ---- if not exam_id: errors['exam_id'] = "Select exam" # SUBJECT IS OPTIONAL NOW → removed: # if not subject_id: # errors['subject_id'] = "Select subject" if not language_id: errors['language_id'] = "Select language" # NEW if not title or not title.strip(): errors['title'] = "Enter title" # duration duration_int = None try: duration_int = int(duration) if duration else None if duration_int is None or duration_int <= 0: errors['duration'] = "Enter valid duration (minutes)" except ValueError: errors['duration'] = "Duration must be numeric" # totals for fld, lbl in [('total_questions', 'Total questions'), ('total_marks','Total marks')]: val = request.form.get(fld) try: ival = int(val) if val else None if ival is None or ival < 0: errors[fld] = f"Enter valid {lbl.lower()}" except ValueError: errors[fld] = f"{lbl} must be numeric" if not status: errors['status'] = "Select status" if not instructions or not instructions.strip(): errors['instructions'] = "Enter instructions" if errors: return render_template( 'user/edit-mock-test.html', mt=mt, exams=exams, languages=languages, # NEW errors=errors, request=request ) # ---- resolve references ---- exam_ref = Exam.objects(id=exam_id).first() if not exam_ref: errors['exam_id'] = "Invalid exam" sub_exam_ref = SubExam.objects(id=sub_exam_id).first() if sub_exam_id else None # SUBJECT (OPTIONAL) subject_ref = None if subject_id: subject_ref = VideoCourseSubject.objects(id=subject_id).first() if not subject_ref: errors['subject_id'] = "Invalid subject" else: # optional: consistency checks if subject_ref.exam_id != exam_ref: errors['subject_id'] = "Subject does not belong to selected exam" if sub_exam_ref and subject_ref.sub_exam_id != sub_exam_ref: errors['subject_id'] = "Subject does not belong to selected sub exam" # CHAPTER (OPTIONAL, but requires subject) chapter_ref = None if chapter_id: if not subject_ref: errors['chapter_id'] = "Select subject before choosing chapter" else: chapter_ref = Chapters.objects(id=chapter_id).first() if not chapter_ref: errors['chapter_id'] = "Invalid chapter" elif chapter_ref.subject_id != subject_ref: errors['chapter_id'] = "Chapter does not belong to selected subject" # TOPIC (OPTIONAL, but requires subject) topic_ref = None if topic_id: if not subject_ref: errors['topic_id'] = "Select subject before choosing topic" else: topic_ref = Topic.objects(id=topic_id).first() if not topic_ref: errors['topic_id'] = "Invalid topic" else: if topic_ref.subject_id != subject_ref: errors['topic_id'] = "Topic does not belong to selected subject" if chapter_ref and topic_ref.chapter_id != chapter_ref: errors['topic_id'] = "Topic does not belong to selected chapter" # LANGUAGE (REQUIRED) language_ref = Language.objects(id=language_id).first() if language_id else None # NEW if not language_ref: errors['language_id'] = "Invalid language" # NEW if errors: return render_template( 'user/edit-mock-test.html', mt=mt, exams=exams, languages=languages, # NEW errors=errors, request=request ) # ---- update ---- try: mt.update( set__exam_id=exam_ref, set__sub_exam_id=sub_exam_ref, set__subject_id=subject_ref, # can be None now set__chapter_id=chapter_ref, # can be None set__topic_id=topic_ref, # can be None set__language_id=language_ref, # NEW set__title=title.strip(), set__duration=duration_int, set__total_questions=int(total_questions), set__total_marks=int(total_marks), set__status=int(status), set__instructions=instructions.strip() ) mt.reload() except Exception: general_error = "Something went wrong while updating" return render_template( 'user/edit-mock-test.html', mt=mt, exams=exams, languages=languages, # NEW errors=errors, general_error=general_error ) flash("Mock test updated successfully", "success") return redirect(url_for('user.edit_mock_test', mock_id=str(mt.id))) return render_template('user/edit-mock-test.html', mt=mt, exams=exams, languages=languages) # 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 # 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/subexams-by-exam/', methods=['GET']) def subexams_by_exam(exam_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 try: exam = Exam.objects(id=exam_id).first() if not exam: return jsonify(success=True, data=[]) key = None if hasattr(SubExam, '_fields'): if 'exam_id' in SubExam._fields: key = 'exam_id' elif 'exam' in SubExam._fields: key = 'exam' if key: subs = SubExam.objects(**{key: exam, 'status': 1}).order_by('sub_exam_title') else: subs = [s for s in SubExam.objects(status=1).order_by('sub_exam_title') if getattr(s, 'exam_id', None) == exam or getattr(s, 'exam', None) == exam] data = [{'id': str(s.id), 'title': s.sub_exam_title} for s in subs] return jsonify(success=True, data=data) except Exception as e: return jsonify(success=False, message=str(e), data=[]) @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') @user_bp.route('/user/view-video-course-subjects', methods=['GET']) def video_course_subjects_list(): if 'user' not in session: return redirect(url_for('user.user_login')) subjects = VideoCourseSubject.objects().order_by('-created_date') view_rows = [] for s in subjects: exam_title = '-' subexam_title = '-' # Prefer new fields if present try: if getattr(s, 'exam_id', None): exam_title = getattr(s.exam_id, 'exam_title', '-') or '-' except Exception: exam_title = '-' try: if getattr(s, 'sub_exam_id', None): subexam_title = getattr(s.sub_exam_id, 'sub_exam_title', '-') or '-' except Exception: subexam_title = '-' # Fallback to legacy course_id → exam/subexam if (exam_title == '-' or subexam_title == '-') and getattr(s, 'course_id', None): try: if exam_title == '-' and getattr(s.course_id, 'exam_id', None): exam_title = getattr(s.course_id.exam_id, 'exam_title', '-') or '-' except Exception: pass try: if subexam_title == '-' and getattr(s.course_id, 'sub_exam_id', None): subexam_title = getattr(s.course_id.sub_exam_id, 'sub_exam_title', '-') or '-' except Exception: pass view_rows.append({ 'id': str(s.id), 'subject_name': s.subject_name, 'exam_title': exam_title, 'subexam_title': subexam_title, 'status': s.status }) # Pass ONLY view_rows to keep the template simple and avoid index mismatches return render_template('user/view-video-course-subjects.html', view_rows=view_rows) @user_bp.route('/user/edit-video-course-subject/', methods=['GET', 'POST']) def edit_video_course_subject(subject_id): if 'user' not in session: return redirect(url_for('user.user_login')) subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: flash("Video course subject not found", "danger") return redirect(url_for('user.video_course_subjects_list')) errors, general_error = {}, None exams = Exam.objects(status=1).order_by('exam_title') if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') subject_name = request.form.get('subject_name') status = request.form.get('status') # --- helper: normalize for duplicate comparison --- def normalize_key(name: str) -> str: if not name: return "" cleaned = name.strip() # trim ends cleaned = "".join(cleaned.split()) # remove ALL internal whitespace cleaned = cleaned.lower() # lowercase return cleaned subject_name_clean = subject_name.strip() if subject_name else "" new_subject_key = normalize_key(subject_name) # 1. validate exam exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" # 2. subexam validation (conditional requirement, same logic as add_video_course_subject) subexam = None exam_has_subexams = False if exam: # Figure out which subexams belong to this exam (status=1) possible_subexams = [] if hasattr(SubExam, '_fields'): if 'exam_id' in SubExam._fields: possible_subexams = list(SubExam.objects(exam_id=exam, status=1)) elif 'exam' in SubExam._fields: possible_subexams = list(SubExam.objects(exam=exam, status=1)) else: possible_subexams = [ s for s in SubExam.objects(status=1) if getattr(s, 'exam_id', None) == exam or getattr(s, 'exam', None) == exam ] else: possible_subexams = [ s for s in SubExam.objects(status=1) if getattr(s, 'exam_id', None) == exam or getattr(s, 'exam', None) == exam ] exam_has_subexams = len(possible_subexams) > 0 if exam_has_subexams: # subexam MUST be provided and must belong to that exam if not sub_exam_id: errors['sub_exam_id'] = "Select subexam" else: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select subexam" else: parent_exam_of_sub = getattr( subexam, 'exam_id', getattr(subexam, 'exam', None) ) if parent_exam_of_sub != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" else: # exam has NO subexams -> allow no subexam subexam = None # 3. subject_name required if not subject_name_clean: errors['subject_name'] = "Enter subject name" # 4. status required if not status: errors['status'] = "Select status" # 5. duplicate check # Only run if we have exam and valid subject_name and no earlier validation errors. # We consider it a duplicate if: # - same exam # - same subexam (or both None) # - another subject with normalized name match # - that "another subject" is NOT this same record if not errors and exam and subject_name_clean: base_q = { "exam_id": exam, } if subexam is None: base_q["sub_exam_id"] = None else: base_q["sub_exam_id"] = subexam # get all subjects that match this (exam, subexam) existing_subjects = VideoCourseSubject.objects(**base_q) for s in existing_subjects: if str(s.id) == str(subject.id): # skip self continue existing_key = normalize_key(s.subject_name) if existing_key == new_subject_key: errors['subject_name'] = "This subject already exists for this exam/subexam" break # If there are any errors, re-render the form with errors if errors: return render_template( 'user/edit-video-course-subject.html', errors=errors, general_error=general_error, request=request, subject=subject, exams=exams ) # All good → update DB try: subject.update( set__course_id=None, # keep null for legacy safety set__exam_id=exam, set__sub_exam_id=subexam, # may now be None if exam has no subexams set__subject_name=subject_name_clean, set__status=int(status) ) subject.reload() flash("Video course subject updated successfully", "success") return redirect(url_for('user.edit_video_course_subject', subject_id=subject.id)) except Exception as e: return render_template( 'user/edit-video-course-subject.html', general_error=f"Save failed: {type(e).__name__}: {e}", request=request, subject=subject, exams=exams ) # GET request return render_template( 'user/edit-video-course-subject.html', subject=subject, exams=exams ) @user_bp.route('/user/delete-video-course-subject/', methods=['POST']) def delete_video_course_subject(subject_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: return jsonify(success=False, message="Subject not found"), 404 try: subject.delete() return jsonify(success=True, message="Subject deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 @user_bp.route('/user/add-video-course-subject', methods=['GET', 'POST']) def add_video_course_subject(): 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') subject_name = (request.form.get('subject_name') or '').strip() status_val = request.form.get('status') sub_exam_ids = request.form.getlist('sub_exam_ids[]') or request.form.getlist('sub_exam_ids') # ---- validations ---- if not exam_id: errors['exam_id'] = "Select exam" if not subject_name: errors['subject_name'] = "Enter subject name" if status_val not in ('0', '1'): errors['status'] = "Select status" # NOTE: Sub-exam selection is OPTIONAL now → no validation error here if errors: return render_template( 'user/add-video-course-subject.html', errors=errors, general_error=general_error, exams=exams, request=request ) exam = Exam.objects(id=exam_id).first() if not exam: errors['exam_id'] = "Invalid exam selected" return render_template('user/add-video-course-subject.html', errors=errors, exams=exams, request=request) # If some sub-exams are selected, validate they belong to this exam; else proceed without sub-exams valid_subexams = [] if sub_exam_ids: valid_subexams = list(SubExam.objects(id__in=sub_exam_ids, exam_id=exam, status=1)) created = 0 if valid_subexams: for sx in valid_subexams: VideoCourseSubject( course_id=None, exam_id=exam, sub_exam_id=sx, subject_name=subject_name, status=int(status_val) ).save() created += 1 flash(f"Subject created for {created} sub exam(s).", "success") else: # No sub-exams selected → create a single subject bound only to exam VideoCourseSubject( course_id=None, exam_id=exam, sub_exam_id=None, # IMPORTANT: field must be optional in the model subject_name=subject_name, status=int(status_val) ).save() created = 1 flash("Subject created without sub exam.", "success") return redirect(url_for('user.add_video_course_subject')) # GET return render_template('user/add-video-course-subject.html', exams=exams) @user_bp.route('/user/subjects-by-exam', methods=['GET']) def subjects_by_exam_v2(): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 exam_id = (request.args.get('exam_id') or '').strip() sub_exam_ids_csv = (request.args.get('sub_exam_ids') or '').strip() if not exam_id: return jsonify(success=True, data=[]) exam = Exam.objects(id=exam_id).first() if not exam: return jsonify(success=True, data=[]) query = dict(exam_id=exam, status=1) if sub_exam_ids_csv: sub_ids = [s for s in sub_exam_ids_csv.split(',') if s] query['sub_exam_id__in'] = sub_ids subs = VideoCourseSubject.objects(**query).only('id', 'subject_name', 'sub_exam_id').order_by('subject_name') # prefetch subexam names se_ids = [str(s.sub_exam_id.id) for s in subs if s.sub_exam_id] se_map = {} if se_ids: for se in SubExam.objects(id__in=se_ids).only('id', 'sub_exam_title'): se_map[str(se.id)] = se.sub_exam_title data = [] for s in subs: sid = str(s.id) sub_id = str(s.sub_exam_id.id) if s.sub_exam_id else None data.append({ "id": sid, "name": s.subject_name, "sub_exam_id": sub_id, "sub_exam_name": se_map.get(sub_id) if sub_id else None }) return jsonify(success=True, data=data) # @user_bp.route('/user/subjects-by-exam', methods=['GET']) # def subjects_by_exam_and_subexams(): # if 'user' not in session: # return jsonify(success=False, message="Unauthorized"), 401 # exam_id = request.args.get('exam_id', '').strip() # raw_subs = request.args.get('sub_exam_ids', '').strip() # comma-separated or empty # if not exam_id: # return jsonify(success=True, data=[]) # exam = Exam.objects(id=exam_id).first() # if not exam: # return jsonify(success=True, data=[]) # q = dict(exam_id=exam) # sub_ids = [] # if raw_subs: # sub_ids = [sid.strip() for sid in raw_subs.split(',') if sid.strip()] # if sub_ids: # q['sub_exam_id__in'] = sub_ids # subjects = VideoCourseSubject.objects(**q).only('id', 'subject_name', 'sub_exam_id') # data = [] # # Pre-fetch subexam names to avoid deref errors # sub_map = {} # for s in subjects: # sid = str(s.id) # sub_id = str(s.sub_exam_id.id) if s.sub_exam_id else None # if sub_id and sub_id not in sub_map: # se = SubExam.objects(id=sub_id).only('sub_exam_title').first() # sub_map[sub_id] = se.sub_exam_title if se else None # data.append({ # 'id': sid, # 'name': s.subject_name, # 'sub_exam_id': sub_id, # 'sub_exam_name': sub_map.get(sub_id) if sub_id else None # }) # return jsonify(success=True, data=data) # @user_bp.route('/user/chapters-by-exam-subjects', methods=['GET']) # def chapters_by_exam_subjects(): # if 'user' not in session: # return jsonify(success=False, message="Unauthorized"), 401 # exam_id = (request.args.get('exam_id') or '').strip() # subject_ids_csv = (request.args.get('subject_ids') or '').strip() # comma-separated # # sub_exam_ids is optional; chapters are keyed mainly by exam+subject # sub_exam_ids_csv = (request.args.get('sub_exam_ids') or '').strip() # if not exam_id or not subject_ids_csv: # return jsonify(success=True, data=[]) # exam = Exam.objects(id=exam_id).first() # if not exam: # return jsonify(success=True, data=[]) # subject_ids = [s.strip() for s in subject_ids_csv.split(',') if s.strip()] # # Optional: if sub_exam_ids provided, we’ll still allow chapters whose sub_exam_id is None. # q = dict(exam_id=exam, subject_id__in=subject_ids) # chapters = Chapters.objects(**q).only('id', 'title', 'subject_id', 'sub_exam_id').order_by('title') # if not chapters: # return jsonify(success=True, data=[]) # # Build label context (subject & sub-exam names) # subj_map = {} # subexam_map = {} # # prefetch subjects # for s in VideoCourseSubject.objects(id__in=subject_ids).only('id', 'subject_name', 'sub_exam_id'): # subj_map[str(s.id)] = { # 'name': s.subject_name, # 'sub_exam_id': str(s.sub_exam_id.id) if s.sub_exam_id else None # } # if s.sub_exam_id: # sid = str(s.sub_exam_id.id) # if sid not in subexam_map: # se = SubExam.objects(id=sid).only('sub_exam_title').first() # subexam_map[sid] = se.sub_exam_title if se else None # data = [] # for c in chapters: # subj_id = str(c.subject_id.id) if c.subject_id else None # sub_id = str(c.sub_exam_id.id) if getattr(c, 'sub_exam_id', None) else subj_map.get(subj_id, {}).get('sub_exam_id') # sub_name = subexam_map.get(sub_id) if sub_id else None # data.append({ # 'id': str(c.id), # 'title': c.title, # 'subject_id': subj_id, # 'subject_name': subj_map.get(subj_id, {}).get('name'), # 'sub_exam_id': sub_id, # 'sub_exam_name': sub_name # }) # return jsonify(success=True, data=data) @user_bp.route('/user/chapters-by-exam-subjects', methods=['GET']) def chapters_by_exam_subjects_v2(): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 exam_id = (request.args.get('exam_id') or '').strip() subject_ids_csv = (request.args.get('subject_ids') or '').strip() # optional: sub_exam_ids_csv = (request.args.get('sub_exam_ids') or '').strip() if not exam_id or not subject_ids_csv: return jsonify(success=True, data=[]) exam = Exam.objects(id=exam_id).first() if not exam: return jsonify(success=True, data=[]) subject_ids = [s for s in subject_ids_csv.split(',') if s] q = dict(exam_id=exam, subject_id__in=subject_ids) chapters = Chapters.objects(**q).only('id', 'title', 'subject_id', 'sub_exam_id').order_by('title') if not chapters: return jsonify(success=True, data=[]) # Prefetch names subj_map = {} for s in VideoCourseSubject.objects(id__in=subject_ids).only('id', 'subject_name'): subj_map[str(s.id)] = s.subject_name # subexam names se_ids = set() for c in chapters: if c.sub_exam_id: se_ids.add(str(c.sub_exam_id.id)) se_map = {} if se_ids: for se in SubExam.objects(id__in=list(se_ids)).only('id', 'sub_exam_title'): se_map[str(se.id)] = se.sub_exam_title data = [] for c in chapters: subj_id = str(c.subject_id.id) if c.subject_id else None se_id = str(c.sub_exam_id.id) if c.sub_exam_id else None data.append({ "id": str(c.id), "title": c.title, "subject_id": subj_id, "subject_name": subj_map.get(subj_id), "sub_exam_id": se_id, "sub_exam_name": se_map.get(se_id) }) return jsonify(success=True, data=data) @user_bp.route('/user/topics-by-exam-subjects-chapters', methods=['GET']) def topics_by_exam_subjects_chapters(): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 exam_id = (request.args.get('exam_id') or '').strip() subject_ids_csv = (request.args.get('subject_ids') or '').strip() chapter_ids_csv = (request.args.get('chapter_ids') or '').strip() if not exam_id or not subject_ids_csv: return jsonify(success=True, data=[]) exam = Exam.objects(id=exam_id).first() if not exam: return jsonify(success=True, data=[]) subject_ids = [s for s in subject_ids_csv.split(',') if s] q = dict(exam_id=exam, subject_id__in=subject_ids) if chapter_ids_csv: chapter_ids = [c for c in chapter_ids_csv.split(',') if c] q['chapter_id__in'] = chapter_ids topics = Topic.objects(**q).only('id', 'title', 'chapter_id', 'sub_exam_id').order_by('title') # chapter names chap_ids = [str(t.chapter_id.id) for t in topics if t.chapter_id] chap_map = {} if chap_ids: for ch in Chapters.objects(id__in=chap_ids).only('id', 'title'): chap_map[str(ch.id)] = ch.title # sub-exam names se_ids = [str(t.sub_exam_id.id) for t in topics if t.sub_exam_id] se_map = {} if se_ids: for se in SubExam.objects(id__in=se_ids).only('id', 'sub_exam_title'): se_map[str(se.id)] = se.sub_exam_title data = [] for t in topics: ch_id = str(t.chapter_id.id) if t.chapter_id else None se_id = str(t.sub_exam_id.id) if t.sub_exam_id else None data.append({ "id": str(t.id), "title": t.title, "chapter_id": ch_id, "chapter_title": chap_map.get(ch_id), "sub_exam_id": se_id, "sub_exam_name": se_map.get(se_id) }) return jsonify(success=True, data=data) @user_bp.route('/user/topics-by-chapter/', methods=['GET']) def topics_by_chapter(chapter_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 try: # Only topics under this chapter topic_qs = Topic.objects(chapter_id=chapter_id).only('id', 'title') except Exception: topic_qs = [] data = [{'id': str(t.id), 'title': t.title} for t in topic_qs] return jsonify(success=True, data=data) @user_bp.route('/user/add-chapter', methods=['GET', 'POST']) def add_chapter(): 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') # Optional multi subexams used to narrow subjects list (not stored directly) sub_exam_ids = request.form.getlist('sub_exam_ids[]') or request.form.getlist('sub_exam_ids') subject_ids = request.form.getlist('subject_ids[]') or request.form.getlist('subject_ids') title = (request.form.get('title') or '').strip() status_val = request.form.get('status') # ---- validations ---- if not exam_id: errors['exam_id'] = "Select exam" if not subject_ids: errors['subject_ids'] = "Select at least one subject" if not title: errors['title'] = "Enter chapter title" if status_val not in ('0', '1'): errors['status'] = "Select status" if errors: return render_template( 'user/add-chapter.html', errors=errors, general_error=general_error, exams=exams, request=request ) exam = Exam.objects(id=exam_id).first() if not exam: errors['exam_id'] = "Invalid exam selected" return render_template('user/add-chapter.html', errors=errors, exams=exams, request=request) # Build subject query subj_q = dict(id__in=subject_ids, exam_id=exam) if sub_exam_ids: subj_q['sub_exam_id__in'] = sub_exam_ids subjects = list(VideoCourseSubject.objects(**subj_q)) if not subjects: errors['subject_ids'] = "No valid subjects under the selected exam/sub exam(s)" return render_template('user/add-chapter.html', errors=errors, exams=exams, request=request) created = 0 for subj in subjects: Chapters( exam_id=exam, sub_exam_id=subj.sub_exam_id, # may be None (allowed) subject_id=subj, title=title, status=int(status_val) ).save() created += 1 flash(f"Chapter '{title}' created for {created} subject(s).", "success") return redirect(url_for('user.add_chapter')) # GET return render_template('user/add-chapter.html', exams=exams) @user_bp.route('/user/view-chapters', methods=['GET']) def chapters_list(): if 'user' not in session: return redirect(url_for('user.user_login')) chapters_qs = Chapters.objects().order_by('-created_date') rows = [] for ch in chapters_qs: # safe deref subject, exam, subexam subject = None exam = None subexam = None try: subject = ch.subject_id except DoesNotExist: subject = None except Exception: subject = None try: exam = ch.exam_id except DoesNotExist: exam = None except Exception: exam = None try: subexam = ch.sub_exam_id except DoesNotExist: subexam = None except Exception: subexam = None rows.append({ 'id': str(ch.id), 'chapter_title': ch.title, 'exam_title': getattr(exam, 'exam_title', '-') if exam else '-', 'subexam_title': getattr(subexam, 'sub_exam_title', '-') if subexam else '-', 'subject_name': getattr(subject, 'subject_name', '-') if subject else '-', 'status': ch.status, 'created_iso': ch.created_date.isoformat() if ch.created_date else '', 'created_pretty': ch.created_date.strftime('%Y-%m-%d %H:%M') if ch.created_date else '-' }) return render_template('user/view-chapters.html', rows=rows) @user_bp.route('/user/edit-chapter/', methods=['GET', 'POST']) def edit_chapter(chapter_id): if 'user' not in session: return redirect(url_for('user.user_login')) ch = Chapters.objects(id=chapter_id).first() if not ch: flash("Chapter not found", "danger") return redirect(url_for('user.chapters_list')) exams = Exam.objects(status=1).order_by('exam_title') errors = {} general_error = None # prefill safe data from existing chapter try: current_exam = ch.exam_id except DoesNotExist: current_exam = None except Exception: current_exam = None try: current_subexam = ch.sub_exam_id except DoesNotExist: current_subexam = None except Exception: current_subexam = None try: current_subject = ch.subject_id except DoesNotExist: current_subject = None except Exception: current_subject = None pre_exam_id = str(current_exam.id) if current_exam else '' pre_sub_id = str(current_subexam.id) if current_subexam else '' pre_subject_id = str(current_subject.id) if current_subject else '' if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') subject_id = request.form.get('subject_id') title = request.form.get('title') status = request.form.get('status') # helper to normalize for duplicate detection def normalize_key(name: str) -> str: if not name: return "" cleaned = name.strip() # trim ends cleaned = "".join(cleaned.split()) # remove ALL whitespace inside cleaned = cleaned.lower() # lowercase return cleaned title_clean = title.strip() if title else "" new_chapter_key = normalize_key(title) # 1. validate exam exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" # 2. validate subexam (optional) subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" else: # make sure subexam belongs to exam parent_exam_of_sub = getattr( subexam, 'exam_id', getattr(subexam, 'exam', None) ) if exam and parent_exam_of_sub != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" # 3. validate subject (required) subject = VideoCourseSubject.objects(id=subject_id).first() if subject_id else None if not subject: errors['subject_id'] = "Select subject" else: # ensure subject ties back to same exam if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" # if subexam chosen, ensure subject.sub_exam_id matches if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" # 4. title required if not title_clean: errors['title'] = "Enter chapter title" # 5. status required if not status: errors['status'] = "Select status" # 6. duplicate check for chapter title # Only run if base validation passed and we have exam, subject, etc. # Duplicate scope: same exam, same subexam-or-None, same subject. # Skip current chapter itself. if not errors and exam and subject and title_clean: base_q = { "exam_id": exam, "subject_id": subject, } if subexam is None: base_q["sub_exam_id"] = None else: base_q["sub_exam_id"] = subexam existing_chapters = Chapters.objects(**base_q) for existing in existing_chapters: # skip self if str(existing.id) == str(ch.id): continue # compare normalized form if normalize_key(existing.title) == new_chapter_key: errors['title'] = "This chapter already exists for this exam / subexam / subject" break # 7. if any errors -> re-render form with sticky data and errors if errors: return render_template( 'user/edit-chapter.html', exams=exams, chapter=ch, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id ) # 8. save updates try: ch.update( set__exam_id=exam, set__sub_exam_id=subexam, set__subject_id=subject, set__title=title_clean, set__status=int(status) ) ch.reload() # refresh preload ids in case exam/subexam/subject changed pre_exam_id = str(exam.id) if exam else '' pre_sub_id = str(subexam.id) if subexam else '' pre_subject_id = str(subject.id) if subject else '' flash("Chapter updated successfully", "success") return redirect(url_for('user.edit_chapter', chapter_id=ch.id)) except Exception as e: general_error = f"Save failed: {type(e).__name__}: {e}" return render_template( 'user/edit-chapter.html', exams=exams, chapter=ch, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id ) # GET (initial render) return render_template( 'user/edit-chapter.html', exams=exams, chapter=ch, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id ) @user_bp.route('/user/delete-chapter/', methods=['POST']) def delete_chapter(chapter_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 ch = Chapters.objects(id=chapter_id).first() if not ch: return jsonify(success=False, message="Chapter not found"), 404 try: ch.delete() return jsonify(success=True, message="Chapter deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 @user_bp.route('/user/chapters-by-subject/', methods=['GET']) def chapters_by_subject(subject_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: return jsonify(success=False, message="Invalid subject"), 400 # All chapters that belong to this subject chapters = Chapters.objects(subject_id=subject).only('id', 'title') data = [{'id': str(c.id), 'title': c.title} for c in chapters] return jsonify(success=True, data=data) @user_bp.route('/user/add-topic', methods=['GET', 'POST']) def add_topic(): 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') sub_exam_ids = request.form.getlist('sub_exam_ids[]') or request.form.getlist('sub_exam_ids') subject_ids = request.form.getlist('subject_ids[]') or request.form.getlist('subject_ids') chapter_ids = request.form.getlist('chapter_ids[]') or request.form.getlist('chapter_ids') title = (request.form.get('title') or '').strip() status_val = request.form.get('status') # --- validations --- if not exam_id: errors['exam_id'] = "Select exam" if not subject_ids: errors['subject_ids'] = "Select at least one subject" if not chapter_ids: errors['chapter_ids'] = "Select at least one chapter" if not title: errors['title'] = "Enter topic title" if status_val not in ('0', '1'): errors['status'] = "Select status" if errors: return render_template( 'user/add-topic.html', errors=errors, general_error=general_error, exams=exams, request=request ) exam = Exam.objects(id=exam_id).first() if not exam: errors['exam_id'] = "Invalid exam selected" return render_template('user/add-topic.html', errors=errors, exams=exams, request=request) # Subjects must belong to exam; if sub-exams provided, restrict to those subjects (but sub-exams are optional) subj_q = dict(id__in=subject_ids, exam_id=exam) if sub_exam_ids: subj_q['sub_exam_id__in'] = sub_exam_ids subjects = list(VideoCourseSubject.objects(**subj_q).only('id', 'sub_exam_id')) if not subjects: errors['subject_ids'] = "No valid subjects for the selection" return render_template('user/add-topic.html', errors=errors, exams=exams, request=request) valid_subject_ids = [str(s.id) for s in subjects] # Chapters must match exam & selected subjects chap_q = dict(id__in=chapter_ids, exam_id=exam, subject_id__in=valid_subject_ids) chapters = list(Chapters.objects(**chap_q)) if not chapters: errors['chapter_ids'] = "No valid chapters for the selection" return render_template('user/add-topic.html', errors=errors, exams=exams, request=request) created = 0 for ch in chapters: Topic( exam_id=exam, sub_exam_id=ch.sub_exam_id, # may be None (allowed) subject_id=ch.subject_id, chapter_id=ch, title=title, status=int(status_val) ).save() created += 1 flash(f"Topic '{title}' created for {created} chapter(s).", "success") return redirect(url_for('user.add_topic')) # GET return render_template('user/add-topic.html', exams=exams) @user_bp.route('/user/view-topics', methods=['GET']) def topics_list(): if 'user' not in session: return redirect(url_for('user.user_login')) topics_qs = Topic.objects().order_by('-created_date') rows = [] for t in topics_qs: # safe deref for all FKs try: exam = t.exam_id except DoesNotExist: exam = None except Exception: exam = None try: subexam = t.sub_exam_id except DoesNotExist: subexam = None except Exception: subexam = None try: subject = t.subject_id except DoesNotExist: subject = None except Exception: subject = None try: chapter = t.chapter_id except DoesNotExist: chapter = None except Exception: chapter = None rows.append({ 'id': str(t.id), 'topic_title': t.title, 'exam_title': getattr(exam, 'exam_title', '-') if exam else '-', 'subexam_title': getattr(subexam, 'sub_exam_title', '-') if subexam else '-', 'subject_name': getattr(subject, 'subject_name', '-') if subject else '-', 'chapter_title': getattr(chapter, 'title', '-') if chapter else '-', 'status': t.status, 'created_iso': t.created_date.isoformat() if t.created_date else '', 'created_pretty': t.created_date.strftime('%Y-%m-%d %H:%M') if t.created_date else '-' }) return render_template('user/view-topics.html', rows=rows) @user_bp.route('/user/edit-topic/', methods=['GET', 'POST']) def edit_topic(topic_id): if 'user' not in session: return redirect(url_for('user.user_login')) top = Topic.objects(id=topic_id).first() if not top: flash("Topic not found", "danger") return redirect(url_for('user.topics_list')) exams = Exam.objects(status=1).order_by('exam_title') errors = {} general_error = None # Current bindings (safe deref) try: current_exam = top.exam_id except DoesNotExist: current_exam = None except Exception: current_exam = None try: current_subexam = top.sub_exam_id except DoesNotExist: current_subexam = None except Exception: current_subexam = None try: current_subject = top.subject_id except DoesNotExist: current_subject = None except Exception: current_subject = None try: current_chapter = top.chapter_id except DoesNotExist: current_chapter = None except Exception: current_chapter = None pre_exam_id = str(current_exam.id) if current_exam else '' pre_sub_id = str(current_subexam.id) if current_subexam else '' pre_subject_id = str(current_subject.id) if current_subject else '' pre_chapter_id = str(current_chapter.id) if current_chapter else '' if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') title = request.form.get('title') status = request.form.get('status') # helper for duplicate detection, same as subjects/chapters def normalize_key(name: str) -> str: if not name: return "" cleaned = name.strip() # trim ends cleaned = "".join(cleaned.split()) # remove ALL internal whitespace cleaned = cleaned.lower() # lowercase return cleaned title_clean = title.strip() if title else "" new_topic_key = normalize_key(title) # validate exam exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" # validate subexam (optional, but must match exam if provided) subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" else: parent_exam_of_sub = getattr( subexam, 'exam_id', getattr(subexam, 'exam', None) ) if exam and parent_exam_of_sub != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" # validate subject (required) subject = VideoCourseSubject.objects(id=subject_id).first() if subject_id else None if not subject: errors['subject_id'] = "Select subject" else: # subject must match selected exam if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" # if subexam chosen, subject must match that subexam if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" # validate chapter (required) chapter = Chapters.objects(id=chapter_id).first() if chapter_id else None if not chapter: errors['chapter_id'] = "Select chapter" else: # chapter must belong to the chosen subject if subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" # title required if not title_clean: errors['title'] = "Enter topic title" # status required if not status: errors['status'] = "Select status" # duplicate check: # Only run if we have no current errors AND we have # exam, subject, chapter, title. # We consider a duplicate if: # - same exam # - same subexam (or both None) # - same subject # - same chapter # - normalized title matches # - and it's NOT this same topic if not errors and exam and subject and chapter and title_clean: base_q = { "exam_id": exam, "subject_id": subject, "chapter_id": chapter } if subexam is None: base_q["sub_exam_id"] = None else: base_q["sub_exam_id"] = subexam existing_topics = Topic.objects(**base_q) for t in existing_topics: if str(t.id) == str(top.id): # skip self continue existing_key = normalize_key(t.title) if existing_key == new_topic_key: errors['title'] = "This topic already exists for this exam / subexam / subject / chapter" break # if any errors -> re-render form with sticky values if errors: return render_template( 'user/edit-topic.html', exams=exams, topic=top, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id ) # save (safe update) try: top.update( set__exam_id=exam, set__sub_exam_id=subexam, set__subject_id=subject, set__chapter_id=chapter, set__title=title_clean, set__status=int(status) ) top.reload() # refresh the preload values in case user changed associations pre_exam_id = str(exam.id) if exam else '' pre_sub_id = str(subexam.id) if subexam else '' pre_subject_id = str(subject.id) if subject else '' pre_chapter_id = str(chapter.id) if chapter else '' flash("Topic updated successfully", "success") return redirect(url_for('user.edit_topic', topic_id=top.id)) except Exception as e: general_error = f"Save failed: {type(e).__name__}: {e}" return render_template( 'user/edit-topic.html', exams=exams, topic=top, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id ) # GET return render_template( 'user/edit-topic.html', exams=exams, topic=top, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id ) @user_bp.route('/user/delete-topic/', methods=['POST']) def delete_topic(topic_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 t = Topic.objects(id=topic_id).first() if not t: return jsonify(success=False, message="Topic not found"), 404 try: t.delete() return jsonify(success=True, message="Topic deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 # Assign Permissions page @user_bp.route('/user/add-permissions', methods=['GET']) def add_permissions(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/add-permissions.html') # View Roles page @user_bp.route('/user/view-roles', methods=['GET']) def view_roles(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/view-roles.html') @user_bp.route('/user/add-role') def add_role(): if 'user' not in session: return redirect(url_for('user.user_login')) return render_template('user/add-role.html') @user_bp.route('/user/subjects-by-exam-sub//', methods=['GET']) def subjects_by_exam_sub(exam_id, sub_exam_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 try: exam = Exam.objects(id=exam_id).first() sub = SubExam.objects(id=sub_exam_id).first() if not exam or not sub: return jsonify(success=True, data=[]) subjects = VideoCourseSubject.objects( exam_id=exam, sub_exam_id=sub, status=1 ).order_by('subject_name') data = [{'id': str(s.id), 'name': s.subject_name} for s in subjects] return jsonify(success=True, data=data) except InvalidQueryError: # Fallback if legacy fields differ all_active = VideoCourseSubject.objects(status=1).order_by('subject_name') data = [] for s in all_active: if getattr(s, 'exam_id', None) == exam and getattr(s, 'sub_exam_id', None) == sub: data.append({'id': str(s.id), 'name': s.subject_name}) return jsonify(success=True, data=data) except Exception as e: return jsonify(success=False, message=str(e), data=[]) @user_bp.route('/user/add-subject-video', methods=['GET', 'POST']) def add_subject_video(): 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') languages = Language.objects(status=1).order_by('language_name').only('id', 'language_name') # NEW if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') # optional subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') # optional topic_id = request.form.get('topic_id') # optional language_id = request.form.get('language_id') # NEW video_title = request.form.get('video_title') video_duration = request.form.get('video_duration') status = request.form.get('status') file = request.files.get('video_file') # --- Resolve hierarchy (same validations as old papers) --- exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" elif exam and getattr(subexam, 'exam_id', None) != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" # subject = VideoCourseSubject.objects(id=subject_id).first() if subject_id else None # if not subject: # errors['subject_id'] = "Select subject" # else: # if exam and subject.exam_id != exam: # errors['subject_id'] = "Subject does not belong to selected exam" # if subexam and subject.sub_exam_id != subexam: # errors['subject_id'] = "Subject does not belong to selected subexam" subject = None if subject_id: subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: errors['subject_id'] = "Select valid subject" else: if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id).first() if not chapter: errors['chapter_id'] = "Select valid chapter" elif subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" topic = None if topic_id: topic = Topic.objects(id=topic_id).first() if not topic: errors['topic_id'] = "Select valid topic" else: if subject and topic.subject_id != subject: errors['topic_id'] = "Topic does not belong to selected subject" if chapter and topic.chapter_id != chapter: errors['topic_id'] = "Topic does not belong to selected chapter" # --- Language --- if not language_id: errors['language_id'] = "Select language" language_ref = None else: language_ref = Language.objects(id=language_id, status=1).only('id').first() if not language_ref: errors['language_id'] = "Invalid language" # --- Fields --- if not video_title or not video_title.strip(): errors['video_title'] = "Enter video title" if not video_duration or not video_duration.strip(): errors['video_duration'] = "Enter duration (e.g., 12:30 or 01:02:45)" if not status: errors['status'] = "Select status" # --- File validation --- video_url = None if not file or file.filename == '': errors['video_file'] = "Upload video file" else: if not allowed_video(file.filename): errors['video_file'] = f"Only {', '.join(sorted(VIDEO_ALLOWED_EXTENSIONS))} allowed" else: file.seek(0, os.SEEK_END) size_bytes = file.tell() file.seek(0) if size_bytes > MAX_VIDEO_FILE_SIZE: errors['video_file'] = "File too large" if errors: return render_template( 'user/add-subject-video.html', exams=exams, languages=languages, # NEW errors=errors, general_error=general_error, request=request ) # --- Save file --- try: filename = secure_filename(file.filename) upload_folder = os.path.join('static', 'uploads', 'videos') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) file.save(save_path) base_url = request.host_url.rstrip('/') rel = save_path.replace(os.sep, '/') idx = rel.find("static/") rel_url = '/' + (rel[idx:] if idx != -1 else rel) video_url = f"{base_url}{rel_url}" except Exception: general_error = "Video upload failed" return render_template( 'user/add-subject-video.html', exams=exams, languages=languages, # NEW errors=errors, general_error=general_error, request=request ) # --- Save record --- try: SubjectVideo( exam_id=exam, sub_exam_id=subexam, subject_id=subject, chapter_id=chapter, topic_id=topic, language_id=language_ref, # NEW video_title=video_title.strip(), video_duration=video_duration.strip(), video_path=video_url, status=int(status), created_date=datetime.utcnow() ).save() except Exception as e: logger.error("SubjectVideo save failed: %s\n%s", e, traceback.format_exc()) general_error = "Something went wrong while saving" return render_template( 'user/add-subject-video.html', exams=exams, languages=languages, # NEW errors=errors, general_error=general_error, request=request ) flash("Subject video added successfully", "success") return redirect(url_for('user.add_subject_video')) # GET return render_template('user/add-subject-video.html', exams=exams, languages=languages) # NEW @user_bp.route('/user/view-subject-videos', methods=['GET']) def subject_videos_list(): if 'user' not in session: return redirect(url_for('user.user_login')) q = (request.args.get('q') or '').strip() selected_exam_id = (request.args.get('exam_id') or '').strip() selected_subexam_id = (request.args.get('subexam_id') or request.args.get('sub_exam_id') or '').strip() selected_subject_id = (request.args.get('subject_id') or '').strip() selected_chapter_id = (request.args.get('chapter_id') or '').strip() selected_topic_id = (request.args.get('topic_id') or '').strip() selected_language_id = (request.args.get('language_id') or '').strip() # NEW status = (request.args.get('status') or '').strip() # '1','0','' qry = SubjectVideo.objects # search if q: qry = qry.filter(video_title__icontains=q) # hierarchy filters if selected_exam_id: try: qry = qry.filter(exam_id=Exam.objects.get(id=selected_exam_id)) except Exception: pass if selected_subexam_id: try: qry = qry.filter(sub_exam_id=SubExam.objects.get(id=selected_subexam_id)) except Exception: pass if selected_subject_id: try: qry = qry.filter(subject_id=VideoCourseSubject.objects.get(id=selected_subject_id)) except Exception: pass if selected_chapter_id: try: qry = qry.filter(chapter_id=Chapters.objects.get(id=selected_chapter_id)) except Exception: pass if selected_topic_id: try: qry = qry.filter(topic_id=Topic.objects.get(id=selected_topic_id)) except Exception: pass # language filter (NEW) if selected_language_id: try: qry = qry.filter(language_id=Language.objects.get(id=selected_language_id)) except Exception: pass # status if status in ('0', '1'): try: qry = qry.filter(status=int(status)) except Exception: pass vids = qry.order_by('-created_date') rows = [] for v in vids: try: exam = v.exam_id except Exception: exam = None try: subexam = v.sub_exam_id except Exception: subexam = None try: subject = v.subject_id except Exception: subject = None try: chapter = v.chapter_id except Exception: chapter = None try: topic = v.topic_id except Exception: topic = None try: language = v.language_id # NEW except Exception: language = None rows.append({ 'id': str(v.id), 'video_title': v.video_title, 'exam_title': getattr(exam, 'exam_title', '-') if exam else '-', 'subexam_title': getattr(subexam, 'sub_exam_title', '-') if subexam else '-', 'subject_name': getattr(subject, 'subject_name', '-') if subject else '-', 'chapter_title': getattr(chapter, 'title', '-') if chapter else '-', 'topic_title': getattr(topic, 'title', '-') if topic else '-', 'language_name': getattr(language, 'language_name', '-') if language else '-', # NEW 'video_duration': v.video_duration, 'status': v.status, 'file_path': getattr(v, 'video_path', None), 'created_iso': v.created_date.isoformat() if getattr(v, 'created_date', None) else '', 'created_pretty': v.created_date.strftime('%Y-%m-%d %H:%M') if getattr(v, 'created_date', None) else '-' }) exams = Exam.objects(status=1).order_by('exam_title').only('id', 'exam_title') # Languages for filter dropdown (NEW) try: languages = Language.objects.order_by('language_name').only('id', 'language_name') except Exception: languages = [] # fail-safe filters = { 'q': q, 'exam_id': selected_exam_id, 'subexam_id': selected_subexam_id, 'subject_id': selected_subject_id, 'chapter_id': selected_chapter_id, 'topic_id': selected_topic_id, 'language_id': selected_language_id, # NEW 'status': status } return render_template( 'user/view-subject-videos.html', rows=rows, exams=exams, languages=languages, # NEW filters=filters ) @user_bp.route('/user/edit-subject-video/', methods=['GET', 'POST']) def edit_subject_video(video_id): if 'user' not in session: return redirect(url_for('user.user_login')) video = SubjectVideo.objects(id=video_id).first() if not video: flash("Video not found", "danger") return redirect(url_for('user.subject_videos_list')) exams = Exam.objects(status=1).order_by('exam_title') languages = Language.objects(status=1).order_by('language_name').only('id', 'language_name') errors = {} general_error = None # current refs (safe deref) try: cur_exam = video.exam_id except DoesNotExist: cur_exam = None except Exception: cur_exam = None try: cur_sub = video.sub_exam_id except DoesNotExist: cur_sub = None except Exception: cur_sub = None try: cur_subject = video.subject_id except DoesNotExist: cur_subject = None except Exception: cur_subject = None try: cur_chapter = video.chapter_id except DoesNotExist: cur_chapter = None except Exception: cur_chapter = None try: cur_topic = video.topic_id except DoesNotExist: cur_topic = None except Exception: cur_topic = None try: cur_language = video.language_id except DoesNotExist: cur_language = None except Exception: cur_language = None pre_exam_id = str(cur_exam.id) if cur_exam else '' pre_sub_id = str(cur_sub.id) if cur_sub else '' pre_subject_id = str(cur_subject.id) if cur_subject else '' pre_chapter_id = str(cur_chapter.id) if cur_chapter else '' pre_topic_id = str(cur_topic.id) if cur_topic else '' pre_language_id = str(cur_language.id) if cur_language else '' if request.method == 'POST': exam_id = request.form.get('exam_id') sub_exam_id = request.form.get('sub_exam_id') subject_id = request.form.get('subject_id') chapter_id = request.form.get('chapter_id') topic_id = request.form.get('topic_id') language_id = request.form.get('language_id') video_title = request.form.get('video_title') video_duration = request.form.get('video_duration') status = request.form.get('status') file = request.files.get('video_file') # optional # ---- validations (hierarchy) ---- exam = Exam.objects(id=exam_id).first() if exam_id else None if not exam: errors['exam_id'] = "Select exam" subexam = None if sub_exam_id: subexam = SubExam.objects(id=sub_exam_id).first() if not subexam: errors['sub_exam_id'] = "Select valid subexam" elif exam and getattr(subexam, 'exam_id', None) != exam: errors['sub_exam_id'] = "Subexam does not belong to selected exam" # subject = VideoCourseSubject.objects(id=subject_id).first() if subject_id else None # if not subject: # errors['subject_id'] = "Select subject" # else: # if exam and subject.exam_id != exam: # errors['subject_id'] = "Subject does not belong to selected exam" # if subexam and subject.sub_exam_id != subexam: # errors['subject_id'] = "Subject does not belong to selected subexam" subject = None if subject_id: subject = VideoCourseSubject.objects(id=subject_id).first() if not subject: errors['subject_id'] = "Select valid subject" else: if exam and subject.exam_id != exam: errors['subject_id'] = "Subject does not belong to selected exam" if subexam and subject.sub_exam_id != subexam: errors['subject_id'] = "Subject does not belong to selected subexam" chapter = None if chapter_id: chapter = Chapters.objects(id=chapter_id).first() if not chapter: errors['chapter_id'] = "Select valid chapter" elif subject and chapter.subject_id != subject: errors['chapter_id'] = "Chapter does not belong to selected subject" topic = None if topic_id: topic = Topic.objects(id=topic_id).first() if not topic: errors['topic_id'] = "Select valid topic" else: if subject and topic.subject_id != subject: errors['topic_id'] = "Topic does not belong to selected subject" if chapter and topic.chapter_id != chapter: errors['topic_id'] = "Topic does not belong to selected chapter" # ---- language (required) ---- language_ref = None if not language_id: errors['language_id'] = "Select language" else: language_ref = Language.objects(id=language_id, status=1).only('id').first() if not language_ref: errors['language_id'] = "Invalid language" # ---- fields ---- if not video_title or not video_title.strip(): errors['video_title'] = "Enter video title" if not video_duration or not video_duration.strip(): errors['video_duration'] = "Enter duration" if not status: errors['status'] = "Select status" # file optional but validate if present new_video_url = video.video_path if file and file.filename: if not allowed_video(file.filename): errors['video_file'] = f"Only {', '.join(sorted(VIDEO_ALLOWED_EXTENSIONS))} allowed" else: file.seek(0, os.SEEK_END) sz = file.tell() file.seek(0) if sz > MAX_VIDEO_FILE_SIZE: errors['video_file'] = "File too large" if errors: return render_template( 'user/edit-subject-video.html', exams=exams, languages=languages, video=video, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) if file and file.filename: try: filename = secure_filename(file.filename) upload_folder = os.path.join('static', 'uploads', 'videos') os.makedirs(upload_folder, exist_ok=True) save_path = os.path.join(upload_folder, filename) file.save(save_path) base_url = request.host_url.rstrip('/') rel = save_path.replace(os.sep, '/') idx = rel.find("static/") rel_url = '/' + (rel[idx:] if idx != -1 else rel) new_video_url = f"{base_url}{rel_url}" except Exception: general_error = "Video upload failed" return render_template( 'user/edit-subject-video.html', exams=exams, languages=languages, video=video, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) try: video.update( set__exam_id=exam, set__sub_exam_id=subexam, set__subject_id=subject, set__chapter_id=chapter, set__topic_id=topic, set__language_id=language_ref, # <-- NEW set__video_title=video_title.strip(), set__video_duration=video_duration.strip(), set__video_path=new_video_url, set__status=int(status) ) video.reload() flash("Subject video updated successfully", "success") return redirect(url_for('user.edit_subject_video', video_id=video.id)) except Exception as e: general_error = f"Save failed: {type(e).__name__}: {e}" return render_template( 'user/edit-subject-video.html', exams=exams, languages=languages, video=video, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) # GET return render_template( 'user/edit-subject-video.html', exams=exams, languages=languages, video=video, errors=errors, general_error=general_error, request=request, pre_exam_id=pre_exam_id, pre_sub_id=pre_sub_id, pre_subject_id=pre_subject_id, pre_chapter_id=pre_chapter_id, pre_topic_id=pre_topic_id, pre_language_id=pre_language_id ) @user_bp.route('/user/delete-subject-video/', methods=['POST']) def delete_subject_video(video_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 v = SubjectVideo.objects(id=video_id).first() if not v: return jsonify(success=False, message="Video not found"), 404 try: v.delete() return jsonify(success=True, message="Video deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 @user_bp.route('/user/subjects-by-exam/', methods=['GET']) def subjects_by_exam_only(exam_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 exam = Exam.objects(id=exam_id).first() if not exam: return jsonify(success=False, message="Invalid exam"), 400 # all subjects that match this exam AND have no subexam (sub_exam_id is None) subjects = VideoCourseSubject.objects( exam_id=exam, sub_exam_id=None ).only('id', 'subject_name') data = [{'id': str(s.id), 'name': s.subject_name} for s in subjects] return jsonify(success=True, data=data) # ======================= Content Upload ===================================== @user_bp.route('/user/content-upload') def content_upload(): if "user" in session: user=session.get("user") jwt_payload={"id":user} token=generate_token(jwt_payload) return render_template('user/content-upload.html',user=user,token=token) @user_bp.route('/user/view-upload') def view_upload(user_id=None): if "user" in session: user=session.get("user") jwt_payload={"id":user} token=generate_token(jwt_payload) return render_template('user/view-upload.html',user=user,token=token) # ======================= Resourse Mapping with course ================================== @user_bp.route('/user/course//attach-resources', methods=['POST']) def attach_resources_to_course(course_id): """ Attach endpoint used by the front-end 'Publish' button. IMPORTANT: This version DOES NOT modify Course.* fields. It only creates ContentMapping documents (one per attached item). """ try: # Try to import ContentMapping model try: from models.content_mapping import ContentMapping except Exception: ContentMapping = None current_app.logger.exception("ContentMapping model import failed; mapping creation disabled.") # resource -> (ModelClassName, content_type_label) mapping = { 'videos': ('SubjectVideo', 'video'), 'ebooks': ('Ebook', 'ebook'), 'mocktests': ('MockTest', 'mocktest'), 'oldpapers': ('OldPapers', 'oldpaper'), } data = request.get_json(silent=True) or {} resource = (data.get('resource') or data.get('type') or '').strip().lower() raw_ids = data.get('item_ids') or data.get('itemIds') or data.get('ids') or data.get('checked') or \ data.get('video_ids') or data.get('ebook_ids') or data.get('mocktest_ids') or data.get('oldpaper_ids') or [] # normalize to list, preserve order, dedupe if isinstance(raw_ids, str): if ',' in raw_ids: raw_ids = [x.strip() for x in raw_ids.split(',') if x.strip()] else: raw_ids = [raw_ids] if not isinstance(raw_ids, (list, tuple)): raw_ids = [raw_ids] seen = set() item_ids = [] for x in raw_ids: if x is None: continue s = str(x).strip() if not s: continue if s in seen: continue seen.add(s) item_ids.append(s) if not resource or resource not in mapping: return jsonify(success=False, message='Invalid or missing resource type'), 400 if not item_ids: return jsonify(success=False, message='No item ids provided', created_mapping_ids=[], skipped_existing_ids=[], missing_ids=[]), 400 model_name, content_type_label = mapping[resource] # load Course model (only to derive course id, not to modify) try: from models.course import Course except Exception: Course = globals().get('Course') if Course is None: return jsonify(success=False, message='Course model not available'), 500 # Resolve target course id (use provided path param unless it's 'latest' or invalid) target_course = None try: if course_id and course_id != 'latest': try: # try by ObjectId or string id target_course = Course.objects(id=course_id).first() except Exception: target_course = None except Exception: target_course = None if not target_course: try: target_course = Course.objects.order_by('-created_date').first() or Course.objects.order_by('-id').first() except Exception: target_course = None if not target_course: return jsonify(success=False, message='No Course found to associate mappings with'), 404 course_id_str = str(getattr(target_course, 'id', getattr(target_course, '_id', None)) or '') # import resource model model_imports = { 'SubjectVideo': 'models.subject_video_model', 'Ebook': 'models.ebooks_model', 'OldPapers': 'models.old_papers_model', 'MockTest': 'models.mock_test_model', } mod_path = model_imports.get(model_name) if not mod_path: return jsonify(success=False, message=f"No import mapping for {model_name}"), 500 try: mod = __import__(mod_path, fromlist=[model_name]) ModelCls = getattr(mod, model_name) except Exception as e: current_app.logger.exception("Model import failed for attach-resources") return jsonify(success=False, message=f"Model import failed: {e}"), 500 # Fetch candidate docs obj_ids = [ObjectId(x) for x in item_ids if ObjectId.is_valid(x)] str_ids = [x for x in item_ids if not ObjectId.is_valid(x)] found_q = [] try: if obj_ids and str_ids: found_q = list(ModelCls.objects(id__in=obj_ids) | ModelCls.objects(id__in=str_ids)) elif obj_ids: found_q = list(ModelCls.objects(id__in=obj_ids)) elif str_ids: found_q = list(ModelCls.objects(id__in=str_ids)) else: found_q = [] except Exception: # fallback: fetch individually found_q = [] for vid in item_ids: try: doc = ModelCls.objects(id=vid).first() if doc: found_q.append(doc) except Exception: continue found_map = { str(d.id): d for d in found_q } ordered_docs = [] missing_ids = [] seen_local = set() import re as _re for rid in item_ids: if rid in seen_local: continue seen_local.add(rid) doc = found_map.get(rid) if not doc and ObjectId.is_valid(rid): doc = found_map.get(str(ObjectId(rid))) if not doc: m = _re.search(r'([0-9a-fA-F]{24})', rid) if m: cand = m.group(1) doc = found_map.get(cand) if doc: ordered_docs.append(doc) else: try: doc = ModelCls.objects(id=rid).first() if doc: ordered_docs.append(doc) else: missing_ids.append(rid) except Exception: missing_ids.append(rid) # --- Create ContentMapping docs only (do NOT modify Course) --- created_mappings = [] skipped_existing = [] if ContentMapping is None: current_app.logger.warning("ContentMapping not available; skipping mapping creation") else: for d in ordered_docs: cid = str(getattr(d, 'id', getattr(d, '_id', None)) or '') if not cid: current_app.logger.debug("Skipping mapping for item without id: %r", d) continue try: # check existing mapping -> avoid duplicates existing = None try: existing = ContentMapping.objects(course_id=course_id_str, content_type=content_type_label, content_id=cid).first() print(existing,"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee") except Exception: # fallback: raw query existing = None if existing: skipped_existing.append(cid) continue m = ContentMapping(course_id=course_id_str, content_type=content_type_label, content_id=cid) m.save() created_mappings.append(str(m.id)) except Exception: current_app.logger.exception("Error creating ContentMapping for %s:%s", content_type_label, cid) return jsonify(success=True, resource=resource, attached_item_ids=[str(getattr(d, 'id', getattr(d, '_id', None))) for d in ordered_docs], attached_count=len(ordered_docs), missing_ids=missing_ids, created_mapping_ids=created_mappings, skipped_existing_ids=skipped_existing, message=f'Created {len(created_mappings)} mapping(s); skipped {len(skipped_existing)} existing mapping(s).') except Exception as e: current_app.logger.exception('attach-resources error') return jsonify(success=False, message='Server error: ' + str(e)), 500 @user_bp.route('/user/course//detach-resources', methods=['POST']) def detach_resources_from_course(course_id): try: from models.content_mapping import ContentMapping except Exception: ContentMapping = None # mapping to content_type label used in DB mapping = { 'videos': 'video', 'ebooks': 'ebook', 'mocktests': 'mocktest', 'oldpapers': 'oldpaper' } data = request.get_json(silent=True) or {} resource = (data.get('resource') or data.get('type') or '').strip().lower() raw_ids = data.get('item_ids') or data.get('itemIds') or data.get('ids') or [] if isinstance(raw_ids, str): raw_ids = [x.strip() for x in raw_ids.split(',') if x.strip()] if not isinstance(raw_ids, (list, tuple)): raw_ids = [raw_ids] item_ids = [str(x).strip() for x in raw_ids if x is not None and str(x).strip()] if not resource or resource not in mapping: return jsonify(success=False, message='Invalid resource'), 400 if not item_ids: return jsonify(success=False, message='No item ids provided'), 400 if ContentMapping is None: return jsonify(success=False, message='ContentMapping model not available'), 500 content_type_label = mapping[resource] removed = [] not_found = [] # helper to try deleting with different strategies try: # try import of ObjectId for exact typed matching try: from bson import ObjectId oid_available = True except Exception: ObjectId = None oid_available = False for iid in item_ids: iid = iid or '' iid = iid.strip() if not iid: not_found.append(iid) continue deleted_for_this = False # 1) Exact-string match (most likely) try: q = ContentMapping.objects(course_id=str(course_id), content_type=content_type_label, content_id=iid) if q.count(): q.delete() removed.append(iid) deleted_for_this = True app.logger.debug("detach: removed by exact-string: %s (resource=%s)", iid, resource) continue except Exception as e: app.logger.debug("detach: exact-string attempt raised: %s -- %s", iid, e) # 2) If iid looks like a 24-hex ObjectId, try matching against actual ObjectId in DB if oid_available: try: if ObjectId.is_valid(iid): oid = ObjectId(iid) q2 = ContentMapping.objects(course_id=str(course_id), content_type=content_type_label, content_id=oid) if q2.count(): q2.delete() removed.append(iid) deleted_for_this = True app.logger.debug("detach: removed by ObjectId match: %s (resource=%s)", iid, resource) continue except Exception as e: app.logger.debug("detach: ObjectId attempt raised: %s -- %s", iid, e) # 3) Defensive: the DB might store content_id as JSON-string containing $oid or DBRef-like; try substring/regex match try: # use a safe regex search (escape iid) regex = re.compile(re.escape(iid)) q3 = ContentMapping.objects(course_id=str(course_id), content_type=content_type_label, content_id__regex=regex) if q3.count(): q3.delete() removed.append(iid) deleted_for_this = True app.logger.debug("detach: removed by regex substring: %s (resource=%s)", iid, resource) continue except Exception as e: app.logger.debug("detach: regex attempt raised: %s -- %s", iid, e) # 4) Last fallback: try retrieving all mappings for this course+type and compare loosely try: q_all = ContentMapping.objects(course_id=str(course_id), content_type=content_type_label) found_any = False for m in q_all: try: cid = str(getattr(m, 'content_id', '') or '') if not cid: continue # normalized compare: extract 24-hex or substring if iid in cid or re.search(re.escape(iid), cid): # delete this one ContentMapping.objects(id=m.id).delete() found_any = True except Exception: continue if found_any: removed.append(iid) deleted_for_this = True app.logger.debug("detach: removed by scanning all mappings and substring match: %s (resource=%s)", iid, resource) continue except Exception as e: app.logger.debug("detach: final-scan attempt raised: %s -- %s", iid, e) if not deleted_for_this: not_found.append(iid) except Exception as e: app.logger.exception("detach_resources_from_course top-level exception: %s", e) return jsonify(success=False, message=f"Server error: {type(e).__name__}: {e}"), 500 return jsonify(success=True, removed_ids=removed, not_found_ids=not_found, message=f"Removed {len(removed)} mappings, not found {len(not_found)}") @user_bp.route('/user/course//video-subjects', methods=['GET']) def course_video_subjects(course_id): """ Return subjects for the given course_id with resolved labels. Filtering behavior (priority): 1) query params exam_id / sub_exam_id / subject_id / chapter_id / topic_id if provided 2) if use_course_filters=1 -> read Course and use its exam/sub_exam fields (common names) 3) fallback: derive from most recent SubjectVideo Each video item includes resolved exam_title/sub_exam_title/subject_name/chapter_title/topic_title when available. """ try: current_app.logger.debug("course_video_subjects requested for course_id=%s", course_id) # ---- imports used in here ---- from models.content_mapping import ContentMapping from bson import ObjectId # read incoming filter params (explicit) req_exam = request.args.get('exam_id') or None req_subexam = request.args.get('sub_exam_id') or None req_subject = request.args.get('subject_id') or request.args.get('subject') or None req_chapter = request.args.get('chapter_id') or request.args.get('chapter') or None req_topic = request.args.get('topic_id') or request.args.get('topic') or None use_course_filters = request.args.get('use_course_filters') in ('1', 'true', 'True') # --- get raw collections (prefer model._get_collection) --- try: subj_coll = VideoCourseSubject._get_collection() vids_coll = SubjectVideo._get_collection() db = subj_coll.database except Exception: current_app.logger.exception("Model _get_collection() failed; falling back to connection.get_db()") try: db = connection.get_db() except Exception: current_app.logger.exception("Cannot get DB via connection.get_db()") return jsonify(success=False, message='Server error'), 500 # find likely collection names subj_coll = None for nm in ('videocoursesubjects','video_course_subjects','video_course_subject'): if nm in db.list_collection_names(): subj_coll = db[nm]; break vids_coll = None for nm in ('subjectvideos','subject_videos','subject_video','subjectvideo'): if nm in db.list_collection_names(): vids_coll = db[nm]; break if not subj_coll or not vids_coll: current_app.logger.exception("Could not locate subject or video collections") return jsonify(success=False, message='Server error'), 500 # --- fetch raw video docs (prefer as_pymongo) --- try: videos_raw = list(SubjectVideo.objects.order_by('created_date').as_pymongo()) except Exception: try: cursor = vids_coll.find({}, {}) try: cursor = cursor.sort('created_date', 1) videos_raw = list(cursor) except Exception: videos_raw = list(cursor) except Exception as ex: current_app.logger.exception("Failed to read videos collection: %s", ex) videos_raw = [] # If explicit filters not provided, attempt to obtain from course if requested effective_exam = req_exam effective_subexam = req_subexam if use_course_filters and (not effective_exam or not effective_subexam): try: course_obj = Course.objects(id=course_id).first() except Exception: course_obj = None if not course_obj: # try raw collection fallback try: if db and ObjectId.is_valid(course_id): cdoc = db['courses'].find_one({'_id': ObjectId(course_id)}) course_obj = cdoc else: course_obj = None except Exception: course_obj = None if course_obj: # Try multiple candidate field names commonly used def _get_course_field(co, *names): for n in names: try: if isinstance(co, dict): if co.get(n): return co.get(n) else: # MongoEngine Document val = getattr(co, n, None) if val: try: return str(val.id) if hasattr(val, 'id') else str(val) except Exception: return val except Exception: continue return None # common candidate names c_exam = _get_course_field(course_obj, 'exam_id', 'exam', 'selected_exam_id', 'examId', 'examId') c_sub = _get_course_field(course_obj, 'sub_exam_id', 'sub_exam', 'selected_sub_exam_id', 'subExamId', 'sub_exam_id') if not effective_exam and c_exam: effective_exam = str(c_exam) if not effective_subexam and c_sub: effective_subexam = str(c_sub) # If still not set, fallback to deriving from the most recent SubjectVideo (existing behavior) if (not effective_exam or not effective_subexam): try: latest = SubjectVideo.objects.order_by('-created_date').first() if latest: if not effective_exam and getattr(latest, 'exam_id', None): effective_exam = str(getattr(latest, 'exam_id').id if hasattr(getattr(latest, 'exam_id'), 'id') else getattr(latest, 'exam_id')) if not effective_subexam and getattr(latest, 'sub_exam_id', None): effective_subexam = str(getattr(latest, 'sub_exam_id').id if hasattr(getattr(latest, 'sub_exam_id'), 'id') else getattr(latest, 'sub_exam_id')) current_app.logger.debug("Derived filter from latest SubjectVideo -> exam=%s subexam=%s", effective_exam, effective_subexam) except Exception: current_app.logger.exception("Failed to derive recent SubjectVideo for filtering") # --- fetch raw subject docs if available (for grouping/display order) --- try: subj_cursor = subj_coll.find({}, {}) try: subj_cursor = subj_cursor.sort('created_date', 1) subjects_raw = list(subj_cursor) except Exception: subjects_raw = list(subj_cursor) except Exception: subjects_raw = [] # small helper to normalize id-like values to strings def norm_id_like(v): if v is None: return None if isinstance(v, ObjectId): return str(v) if isinstance(v, str): return v if isinstance(v, dict): return str(v.get('$id') or v.get('_id') or v.get('id') or '') try: return str(v) except Exception: return None # apply filtering to videos_raw based on effective_exam / effective_subexam and optional subject/chapter/topic def matches_filters(vdoc): # if no filters active => include everything if not (effective_exam or effective_subexam or req_subject or req_chapter or req_topic): return True ve_exam = vdoc.get('exam_id') or vdoc.get('exam') ve_sub = vdoc.get('sub_exam_id') or vdoc.get('sub_exam') ve_subject = vdoc.get('subject_id') or vdoc.get('subject') ve_chapter = vdoc.get('chapter_id') or vdoc.get('chapter') ve_topic = vdoc.get('topic_id') or vdoc.get('topic') ve_exam_str = norm_id_like(ve_exam) if ve_exam else None ve_sub_str = norm_id_like(ve_sub) if ve_sub else None ve_subject_str = norm_id_like(ve_subject) if ve_subject else None ve_chapter_str = norm_id_like(ve_chapter) if ve_chapter else None ve_topic_str = norm_id_like(ve_topic) if ve_topic else None # exam/sub match logic (only enforce when effective values present) if effective_exam and effective_subexam: if not ((ve_exam_str == effective_exam) and (ve_sub_str == effective_subexam)): return False elif effective_exam: if ve_exam_str != effective_exam: return False elif effective_subexam: if ve_sub_str != effective_subexam: return False # now enforce subject / chapter / topic if provided in query if req_subject: if not ve_subject_str and isinstance(ve_subject, dict): candidate = ve_subject.get('subject_name') or ve_subject.get('title') or ve_subject.get('name') if candidate: ve_subject_str = str(candidate) if ve_subject_str != str(req_subject): return False if req_chapter: if not ve_chapter_str and isinstance(ve_chapter, dict): candidate = ve_chapter.get('chapter_title') or ve_chapter.get('title') or ve_chapter.get('name') if candidate: ve_chapter_str = str(candidate) if ve_chapter_str != str(req_chapter): return False if req_topic: if not ve_topic_str and isinstance(ve_topic, dict): candidate = ve_topic.get('topic_title') or ve_topic.get('title') or ve_topic.get('name') if candidate: ve_topic_str = str(candidate) if ve_topic_str != str(req_topic): return False return True filtered_videos_raw = [v for v in videos_raw if matches_filters(v)] # index videos by subject_id (string) videos_by_subject = {} for v in filtered_videos_raw: subj_ref = v.get('subject_id') or v.get('subject') sid = norm_id_like(subj_ref) or '___no_subject___' videos_by_subject.setdefault(sid, []).append(v) # prepare raw fallback collections for label resolution exams_coll = db['exams'] if db and 'exams' in db.list_collection_names() else None subexams_coll = db['sub_exams_list'] if db and 'sub_exams_list' in db.list_collection_names() else None chapter_coll = db['chapters'] if db and 'chapters' in db.list_collection_names() else None topic_coll = db['topics'] if db and 'topics' in db.list_collection_names() else None video_subject_coll = subj_coll # Try to import models for typed lookups (fast & reliable if available) ExamModel = SubExamModel = VideoCourseSubjectModel = ChaptersModel = TopicModel = None try: from models.exam import Exam as E; ExamModel = E except Exception: pass try: from models.sub_exam import SubExam as S; SubExamModel = S except Exception: pass try: from models.video_courses_model import VideoCourseSubject as V; VideoCourseSubjectModel = V except Exception: pass try: from models.chapter import Chapters as C; ChaptersModel = C except Exception: pass try: from models.topic import Topic as T; TopicModel = T except Exception: pass # caches for labels caches = {'exam':{}, 'subexam':{}, 'subject':{}, 'chapter':{}, 'topic':{}} def resolve_label(val, kind): if not val: return None if isinstance(val, dict): for fld in ('exam_title','sub_exam_title','subject_name','chapter_title','topic_title','title','name'): if val.get(fld): return val.get(fld) val = val.get('$id') or val.get('_id') or val.get('id') or val id_str = str(val) if id_str in caches[kind]: return caches[kind][id_str] model_cls = None; raw_coll = None; fields = [] if kind == 'exam': model_cls = ExamModel; raw_coll = exams_coll; fields = ['exam_title','title','name'] elif kind == 'subexam': model_cls = SubExamModel; raw_coll = subexams_coll; fields = ['sub_exam_title','title','name'] elif kind == 'subject': model_cls = VideoCourseSubjectModel; raw_coll = video_subject_coll; fields = ['subject_name','title','name'] elif kind == 'chapter': model_cls = ChaptersModel; raw_coll = chapter_coll; fields = ['chapter_title','title','name'] elif kind == 'topic': model_cls = TopicModel; raw_coll = topic_coll; fields = ['topic_title','title','name'] if model_cls: try: model_fields = [f for f in fields if f in model_cls._fields] obj = model_cls.objects(id=id_str).only(*model_fields).first() if obj: for f in fields: if hasattr(obj, f) and getattr(obj, f): caches[kind][id_str] = getattr(obj, f) return caches[kind][id_str] except Exception: current_app.logger.debug("Model lookup failed for %s id=%s", kind, id_str, exc_info=True) if raw_coll: try: q = {'_id': ObjectId(id_str)} if ObjectId.is_valid(id_str) else {'_id': id_str} raw = raw_coll.find_one(q) if raw: for f in fields: if raw.get(f): caches[kind][id_str] = raw.get(f) return caches[kind][id_str] except Exception: current_app.logger.debug("Raw lookup failed for %s id=%s", kind, id_str, exc_info=True) caches[kind][id_str] = None return None # helper to pick scalar fields from doc (handles nested dicts) def pick_scalar(doc, *cands): if not doc: return None for c in cands: if c in doc and doc[c] not in (None, ''): val = doc[c] if isinstance(val, ObjectId): return str(val) if isinstance(val, dict): for f in ('title','name','video_title','chapter_title','topic_title'): if val.get(f): return val.get(f) return None return val for v in doc.values(): if isinstance(v, dict): for c in cands: if v.get(c) not in (None, ''): return v.get(c) return None # Build ordered subject keys: prefer subjects_raw order; append any vids-only subjects subject_keys = [] if subjects_raw: for s in subjects_raw: sid = norm_id_like(s.get('_id')) or '' subject_keys.append((sid, s)) for sid in videos_by_subject.keys(): if sid and sid not in [x[0] for x in subject_keys]: subject_keys.append((sid, None)) else: for sid in videos_by_subject.keys(): subject_keys.append((sid, None)) # Build final groups groups = [] for sid, subj_doc in subject_keys: subj_name = None if subj_doc: subj_name = pick_scalar(subj_doc, 'subject_name', 'title', 'name') if not subj_name and sid and sid != '___no_subject___': subj_name = resolve_label(sid, 'subject') if not subj_name: subj_name = 'Uncategorized' if (not sid or sid == '___no_subject___') else ('Subject ' + sid) vids_list = [] for vdoc in videos_by_subject.get(sid, []): vid_id = norm_id_like(vdoc.get('_id')) or '' video_title = pick_scalar(vdoc, 'video_title', 'title', 'name') video_duration = pick_scalar(vdoc, 'video_duration', 'duration', 'time') video_path = pick_scalar(vdoc, 'video_path', 'file_path', 'path', 'video_url', 'video') status = pick_scalar(vdoc, 'status', 'is_active', 'active') exam_title = resolve_label(vdoc.get('exam_id') or vdoc.get('exam'), 'exam') sub_exam_title = resolve_label(vdoc.get('sub_exam_id') or vdoc.get('sub_exam'), 'subexam') subject_title = resolve_label(vdoc.get('subject_id') or vdoc.get('subject'), 'subject') chapter_title = resolve_label(vdoc.get('chapter_id') or vdoc.get('chapter'), 'chapter') topic_title = resolve_label(vdoc.get('topic_id') or vdoc.get('topic'), 'topic') if not video_title: nested = vdoc.get('video') or vdoc.get('video_doc') if isinstance(nested, dict): video_title = pick_scalar(nested, 'video', 'title', 'name') item = { 'id': vid_id, 'video': video_title, 'video_duration': video_duration, 'video_path': video_path, 'status': status } if exam_title: item['exam'] = exam_title if sub_exam_title: item['sub_exam'] = sub_exam_title if subject_title: item['subject'] = subject_title if chapter_title: item['chapter'] = chapter_title if topic_title: item['topic'] = topic_title vids_list.append(item) groups.append({'id': sid, 'subject_name': subj_name, 'videos': vids_list}) # --- attached video ids from ContentMapping (canonical) --- attached = [] try: # normalize course id string same way as attach_resources_to_course course_id_str = str(course_id) if ObjectId.is_valid(course_id_str): course_id_str = str(ObjectId(course_id_str)) cm_q = ContentMapping.objects( course_id=course_id_str, content_type='video' ).only('content_id') attached = [str(m.content_id) for m in cm_q] except Exception: current_app.logger.exception("Failed to load attached video mappings") attached = [] return jsonify(success=True, subjects=groups, attached_video_ids=attached) except Exception: current_app.logger.exception('Error fetching course video subjects') return jsonify(success=False, message='Server error'), 500 @user_bp.route('/user/course//ebook-subjects', methods=['GET']) def course_ebook_subjects(course_id): """ Return ebooks grouped by author (subject_name=author). Accepts optional query params: exam_id, sub_exam_id, subject_id, chapter_id, topic_id, use_course_filters=1 """ try: current_app.logger.debug("course_ebook_subjects requested for course_id=%s", course_id) # extra imports needed here from bson import ObjectId try: from models.content_mapping import ContentMapping except Exception: ContentMapping = None # load raw coll (try model _get_collection first) try: coll = Ebook._get_collection() db = coll.database except Exception: db = connection.get_db() coll = db.get('ebooks') or db.get('ebook') or db['ebooks'] # read all ebooks try: cursor = coll.find({}, {'language_id': 0}) try: cursor = cursor.sort('created_date', 1) ebooks_raw = list(cursor) except Exception: ebooks_raw = list(cursor) except Exception as ex: current_app.logger.exception("Error fetching ebooks: %s", ex) return jsonify(success=False, message='Server error'), 500 # query params req_exam = request.args.get('exam_id') req_subexam = request.args.get('sub_exam_id') req_subject = request.args.get('subject_id') or request.args.get('subject') req_chapter = request.args.get('chapter_id') or request.args.get('chapter') req_topic = request.args.get('topic_id') or request.args.get('topic') use_course_filters = request.args.get('use_course_filters') == '1' # derive defaults from most recent ebook if requested and explicit missing if use_course_filters and (not req_exam or not req_subexam): try: latest = coll.find_one({}, sort=[('created_date', -1)]) if latest: if not req_exam and latest.get('exam_id'): req_exam = str(latest.get('exam_id')) if not req_subexam and (latest.get('sub_exam_id') or latest.get('sub_exam')): req_subexam = str(latest.get('sub_exam_id') or latest.get('sub_exam')) except Exception: current_app.logger.debug("Could not derive recent exam/subexam from ebooks") # helper to normalize id-like values def norm(v): if v is None: return None if isinstance(v, ObjectId): return str(v) return str(v) # filter if any explicit filters present if req_exam or req_subexam or req_subject or req_chapter or req_topic: filtered = [] for e in ebooks_raw: e_exam = norm(e.get('exam_id') or e.get('exam')) e_sub = norm(e.get('sub_exam_id') or e.get('sub_exam') or e.get('sub_exam_name')) e_subject = norm(e.get('subject_id') or e.get('subject') or e.get('subject_name')) e_chapter = norm(e.get('chapter_id') or e.get('chapter') or e.get('chapter_title')) e_topic = norm(e.get('topic_id') or e.get('topic') or e.get('topic_title')) if req_exam and e_exam != str(req_exam): continue if req_subexam and e_sub != str(req_subexam): continue if req_subject and e_subject != str(req_subject): continue if req_chapter and e_chapter != str(req_chapter): continue if req_topic and e_topic != str(req_topic): continue filtered.append(e) ebooks_raw = filtered # prepare label resolution collections exams_coll = db['exams'] if db and 'exams' in db.list_collection_names() else None subexams_coll = db['sub_exams_list'] if db and 'sub_exams_list' in db.list_collection_names() else None chapter_coll = db['chapters'] if db and 'chapters' in db.list_collection_names() else None topic_coll = db['topics'] if db and 'topics' in db.list_collection_names() else None # pick a subject collection if available (ebook-specific preferred) subject_coll = None for nm in ( 'ebook_subjects', 'ebooksubjects', 'ebook_subject', 'videocoursesubjects', 'video_course_subjects', 'video_course_subject' ): if db and nm in db.list_collection_names(): subject_coll = db[nm] break # optional model imports (if your project has them) ExamModel = SubExamModel = ChaptersModel = TopicModel = None try: from models.exam import Exam as E; ExamModel = E except Exception: pass try: from models.sub_exam import SubExam as S; SubExamModel = S except Exception: pass try: from models.chapter import Chapters as C; ChaptersModel = C except Exception: pass try: from models.topic import Topic as T; TopicModel = T except Exception: pass # caches caches = {'exam': {}, 'subexam': {}, 'subject': {}, 'chapter': {}, 'topic': {}} def resolve_label(val, kind): """ Resolve id/embedded doc to a human label string. kind in: 'exam','subexam','subject','chapter','topic' """ if not val: return None # if embedded dict with title fields if isinstance(val, dict): for fld in ( 'title', 'name', 'exam_title', 'sub_exam_title', 'subject_name', 'chapter_title', 'topic_title' ): if val.get(fld): return val.get(fld) val = val.get('$id') or val.get('_id') or val.get('id') or val id_str = str(val) if id_str in caches.get(kind, {}): return caches[kind][id_str] model_cls = None raw_coll = None fields = [] if kind == 'exam': model_cls = ExamModel raw_coll = exams_coll fields = ['exam_title', 'title', 'name'] elif kind == 'subexam': model_cls = SubExamModel raw_coll = subexams_coll fields = ['sub_exam_title', 'title', 'name'] elif kind == 'subject': raw_coll = subject_coll fields = ['subject_name', 'title', 'name'] elif kind == 'chapter': model_cls = ChaptersModel raw_coll = chapter_coll fields = ['chapter_title', 'title', 'name'] elif kind == 'topic': model_cls = TopicModel raw_coll = topic_coll fields = ['topic_title', 'title', 'name'] # try model lookup first if model_cls: try: model_fields = [f for f in fields if f in model_cls._fields] obj = model_cls.objects(id=id_str).only(*model_fields).first() if obj: for f in fields: if hasattr(obj, f) and getattr(obj, f): caches[kind][id_str] = getattr(obj, f) return caches[kind][id_str] except Exception: current_app.logger.debug( "Model lookup failed for %s id=%s", kind, id_str, exc_info=True ) # raw collection fallback if raw_coll: try: q = {'_id': ObjectId(id_str)} if ObjectId.is_valid(id_str) else {'_id': id_str} raw = raw_coll.find_one(q) if raw: for f in fields: if raw.get(f): caches[kind][id_str] = raw.get(f) return caches[kind][id_str] except Exception: current_app.logger.debug( "Raw lookup failed for %s id=%s", kind, id_str, exc_info=True ) caches[kind][id_str] = None return None # group ebooks by author and build items with requested keys groups = {} for d in ebooks_raw: author = d.get('author') or 'Unknown author' key = author groups.setdefault(key, {'id': key, 'subject_name': author, 'items': []}) item = { 'id': str(d.get('_id')) if d.get('_id') else '', 'title': d.get('title') or d.get('ebook_title') or d.get('name'), 'author': d.get('author'), 'meta': d.get('publication_date') or d.get('meta') or '', 'path': d.get('path') or d.get('file_path') or None, 'status': d.get('status') } # exam -> 'exam' if d.get('exam_id') or d.get('exam'): exam_label = resolve_label(d.get('exam_id') or d.get('exam'), 'exam') if exam_label: item['exam'] = exam_label # sub_exam -> 'sub_exam' se = d.get('sub_exam_id') or d.get('sub_exam') if se: s_lab = resolve_label(se, 'subexam') item['sub_exam'] = s_lab if s_lab else '' else: item['sub_exam'] = '' # subject -> 'subject' if d.get('subject_id') or d.get('subject'): subj_lab = resolve_label(d.get('subject_id') or d.get('subject'), 'subject') item['subject'] = subj_lab if subj_lab else '' else: item['subject'] = '' # chapter -> 'chapter' if d.get('chapter_id') or d.get('chapter'): ch_lab = resolve_label(d.get('chapter_id') or d.get('chapter'), 'chapter') item['chapter'] = ch_lab if ch_lab else '' else: item['chapter'] = '' # topic -> 'topic' if d.get('topic_id') or d.get('topic'): t_lab = resolve_label(d.get('topic_id') or d.get('topic'), 'topic') item['topic'] = t_lab if t_lab else '' else: item['topic'] = '' groups[key]['items'].append(item) subjects = list(groups.values()) # ---------- attached ebook ids from ContentMapping (canonical) ---------- attached = [] try: if ContentMapping is not None: course_id_str = str(course_id) # normalize to pure hex if it's an ObjectId string if ObjectId.is_valid(course_id_str): course_id_str = str(ObjectId(course_id_str)) cms = ContentMapping.objects( course_id=course_id_str, content_type='ebook' ).only('content_id') attached = [str(m.content_id) for m in cms] else: attached = [] except Exception: current_app.logger.exception("Failed to load attached ebook mappings") attached = [] return jsonify( success=True, subjects=subjects, attached_ebook_ids=attached, attached_item_ids=attached ) except Exception: current_app.logger.exception('Error fetching ebook subjects') return jsonify(success=False, message='Server error'), 500 @user_bp.route('/user/course//mocktest-subjects', methods=['GET']) def course_mocktest_subjects(course_id): """ Return mock tests grouped by sub_exam (fallback to exam if sub_exam missing). Accepts optional query params: - exam_id - sub_exam_id - subject_id - chapter_id - topic_id - use_course_filters (if present and true, frontend requested filtering) """ try: current_app.logger.debug("course_mocktest_subjects requested for course_id=%s", course_id) from bson import ObjectId try: from models.content_mapping import ContentMapping except Exception: ContentMapping = None # read optional filters from querystring req_exam = request.args.get('exam_id') or None req_subexam = request.args.get('sub_exam_id') or None req_subject = request.args.get('subject_id') or request.args.get('subject') or None req_chapter = request.args.get('chapter_id') or request.args.get('chapter') or None req_topic = request.args.get('topic_id') or request.args.get('topic') or None use_filters = request.args.get('use_course_filters') use_filters = bool(use_filters and str(use_filters).lower() in ('1', 'true', 'yes')) # fetch raw docs (prefer as_pymongo) try: items_raw = list(MockTest.objects.order_by('created_date').as_pymongo()) coll = MockTest._get_collection() db = coll.database except Exception: # fallback raw collection try: db = connection.get_db() coll = ( db.get('mock_test') if 'mock_test' in db.list_collection_names() else db.get('mocktests') or db.get('mock_test') ) cursor = coll.find({}, {}) try: cursor = cursor.sort('created_date', 1) items_raw = list(cursor) except Exception: items_raw = list(cursor) except Exception as ex: current_app.logger.exception("Cannot read mocktests: %s", ex) return jsonify(success=False, message='Server error'), 500 # helper to normalize id-like values to simple string def norm_id_like(v): if v is None: return None if isinstance(v, ObjectId): return str(v) if isinstance(v, dict): # DBRef-like or embedded: prefer $id/_id/id cand = v.get('$id') or v.get('_id') or v.get('id') if isinstance(cand, ObjectId): return str(cand) if cand: return str(cand) # check embedded title fields (not id) return None return str(v) # if filters requested, filter items_raw in memory if use_filters and (req_exam or req_subexam or req_subject or req_chapter or req_topic): f_exam = str(req_exam) if req_exam else None f_sub = str(req_subexam) if req_subexam else None def matches_filters(doc): # doc keys might be exam_id / exam, sub_exam_id / sub_exam de = norm_id_like(doc.get('exam_id') or doc.get('exam')) ds = norm_id_like(doc.get('sub_exam_id') or doc.get('sub_exam')) ds_subject = norm_id_like(doc.get('subject_id') or doc.get('subject')) ds_chapter = norm_id_like(doc.get('chapter_id') or doc.get('chapter')) ds_topic = norm_id_like(doc.get('topic_id') or doc.get('topic')) # exam/sub logic if f_exam and f_sub: if not ((de == f_exam) and (ds == f_sub)): return False elif f_exam: if de != f_exam: return False elif f_sub: if ds != f_sub: return False # optional subject/chapter/topic enforcement if req_subject and (ds_subject != str(req_subject)): return False if req_chapter and (ds_chapter != str(req_chapter)): return False if req_topic and (ds_topic != str(req_topic)): return False return True items_raw = [d for d in items_raw if matches_filters(d)] # prepare fallback raw collections for label resolution try: db = MockTest._get_collection().database except Exception: try: db = connection.get_db() except Exception: db = None exams_coll = db['exams'] if db and 'exams' in db.list_collection_names() else None subexams_coll = db['sub_exams_list'] if db and 'sub_exams_list' in db.list_collection_names() else None subject_coll = None if db: for nm in ('videocoursesubjects', 'video_course_subjects', 'video_course_subject'): if nm in db.list_collection_names(): subject_coll = db[nm] break chapter_coll = db['chapters'] if db and 'chapters' in db.list_collection_names() else None topic_coll = db['topics'] if db and 'topics' in db.list_collection_names() else None # Try to import models for typed lookups (optional) ExamModel = SubExamModel = VideoCourseSubjectModel = ChaptersModel = TopicModel = None try: from models.exam import Exam as E; ExamModel = E except Exception: pass try: from models.sub_exam import SubExam as S; SubExamModel = S except Exception: pass try: from models.video_courses_model import VideoCourseSubject as V; VideoCourseSubjectModel = V except Exception: pass try: from models.chapter import Chapters as C; ChaptersModel = C except Exception: pass try: from models.topic import Topic as T; TopicModel = T except Exception: pass # lightweight resolver with cache caches = {'exam': {}, 'subexam': {}, 'subject': {}, 'chapter': {}, 'topic': {}} def resolve_label(val, kind): if not val: return None if isinstance(val, dict): for f in ( 'sub_exam_title', 'sub_exam_name', 'exam_title', 'subject_name', 'chapter_title', 'topic_title', 'title', 'name' ): if val.get(f): return val.get(f) val = val.get('$id') or val.get('_id') or val.get('id') or val idstr = str(val) if idstr in caches[kind]: return caches[kind][idstr] model_cls = None raw_coll = None fields = [] if kind == 'exam': model_cls = ExamModel raw_coll = exams_coll fields = ['exam_title', 'title', 'name'] elif kind == 'subexam': model_cls = SubExamModel raw_coll = subexams_coll fields = ['sub_exam_title', 'title', 'name'] elif kind == 'subject': model_cls = VideoCourseSubjectModel raw_coll = subject_coll fields = ['subject_name', 'title', 'name'] elif kind == 'chapter': model_cls = ChaptersModel raw_coll = chapter_coll fields = ['chapter_title', 'title', 'name'] elif kind == 'topic': model_cls = TopicModel raw_coll = topic_coll fields = ['topic_title', 'title', 'name'] if model_cls: try: model_fields = [f for f in fields if f in model_cls._fields] obj = model_cls.objects(id=idstr).only(*model_fields).first() if obj: for f in fields: if hasattr(obj, f) and getattr(obj, f): caches[kind][idstr] = getattr(obj, f) return caches[kind][idstr] except Exception: current_app.logger.debug( "Model lookup error for %s %s", kind, idstr, exc_info=True ) if raw_coll: try: q = {'_id': ObjectId(idstr)} if ObjectId.is_valid(idstr) else {'_id': idstr} raw = raw_coll.find_one(q) if raw: for f in fields: if raw.get(f): caches[kind][idstr] = raw.get(f) return caches[kind][idstr] except Exception: current_app.logger.debug( "Raw lookup error for %s %s", kind, idstr, exc_info=True ) caches[kind][idstr] = None return None # grouping by subexam/exam fallback groups = {} for doc in items_raw: se_ref = doc.get('sub_exam_id') or doc.get('sub_exam') ex_ref = doc.get('exam_id') or doc.get('exam') subj_id = None subj_name = None if se_ref: subj_id = norm_id_like(se_ref) subj_name = ( resolve_label(se_ref, 'subexam') or ('SubExam ' + (subj_id or 'unknown')) ) elif ex_ref: subj_id = norm_id_like(ex_ref) subj_name = ( resolve_label(ex_ref, 'exam') or ('Exam ' + (subj_id or 'unknown')) ) else: subj_id = 'general' subj_name = 'SubExam general' raw_id = doc.get('_id') or doc.get('id') id_str = str(raw_id) if raw_id is not None else '' item = { 'id': id_str, 'title': doc.get('title') or doc.get('name'), 'duration': doc.get('duration'), 'total_questions': ( doc.get('total_questions') or doc.get('questions') or doc.get('no_of_questions') ), 'total_marks': doc.get('total_marks') or doc.get('marks'), 'status': doc.get('status') } # attach resolved labels if ex_ref: et = resolve_label(ex_ref, 'exam') if et: item['exam'] = et if se_ref: st = resolve_label(se_ref, 'subexam') if st: item['sub_exam'] = st if doc.get('subject_id'): sname = resolve_label(doc.get('subject_id'), 'subject') if sname: item['subject'] = sname if doc.get('chapter_id'): ch = resolve_label(doc.get('chapter_id'), 'chapter') if ch: item['chapter'] = ch if doc.get('topic_id'): tp = resolve_label(doc.get('topic_id'), 'topic') if tp: item['topic'] = tp groups.setdefault( subj_id or 'general', {'id': subj_id or 'general', 'subject_name': subj_name, 'items': []} ) groups[subj_id or 'general']['items'].append(item) subjects = list(groups.values()) # ---------- attached mocktest ids from ContentMapping ---------- attached = [] try: if ContentMapping is not None: course_id_str = str(course_id) if ObjectId.is_valid(course_id_str): course_id_str = str(ObjectId(course_id_str)) cms = ContentMapping.objects( course_id=course_id_str, content_type='mocktest' ).only('content_id') attached = [str(m.content_id) for m in cms] else: attached = [] except Exception: current_app.logger.exception("Failed to load attached mocktest mappings") attached = [] return jsonify( success=True, subjects=subjects, attached_item_ids=attached, attached_mocktest_ids=attached ) except Exception: current_app.logger.exception('Error fetching mock test subjects') return jsonify(success=False, message='Server error'), 500 @user_bp.route('/user/course//oldpaper-subjects', methods=['GET']) def course_oldpaper_subjects(course_id): """ Return old papers grouped by course/title. Accepts optional query params: - exam_id - sub_exam_id - subject_id - chapter_id - topic_id - use_course_filters """ try: current_app.logger.debug("course_oldpaper_subjects requested for course_id=%s", course_id) from bson import ObjectId try: from models.content_mapping import ContentMapping except Exception: ContentMapping = None req_exam = request.args.get('exam_id') or None req_subexam = request.args.get('sub_exam_id') or None req_subject = request.args.get('subject_id') or request.args.get('subject') or None req_chapter = request.args.get('chapter_id') or request.args.get('chapter') or None req_topic = request.args.get('topic_id') or request.args.get('topic') or None use_filters = request.args.get('use_course_filters') use_filters = bool(use_filters and str(use_filters).lower() in ('1', 'true', 'yes')) # fetch raw docs try: items_raw = list(OldPapers.objects.order_by('created_date').as_pymongo()) coll = OldPapers._get_collection() db = coll.database except Exception: try: db = connection.get_db() coll = ( db['oldpapers'] if 'oldpapers' in db.list_collection_names() else db.get('oldpapers') or db.get('old_papers') ) cursor = coll.find({}, {}) try: cursor = cursor.sort('created_date', 1) items_raw = list(cursor) except Exception: items_raw = list(cursor) except Exception as ex: current_app.logger.exception("Cannot read oldpapers: %s", ex) return jsonify(success=False, message='Server error'), 500 def norm_id_like(v): if v is None: return None if isinstance(v, ObjectId): return str(v) if isinstance(v, dict): cand = v.get('$id') or v.get('_id') or v.get('id') if isinstance(cand, ObjectId): return str(cand) if cand: return str(cand) return None return str(v) # apply filters if requested if use_filters and (req_exam or req_subexam or req_subject or req_chapter or req_topic): f_exam = str(req_exam) if req_exam else None f_sub = str(req_subexam) if req_subexam else None def matches(doc): de = norm_id_like(doc.get('exam_id') or doc.get('exam')) ds = norm_id_like(doc.get('sub_exam_id') or doc.get('sub_exam')) ds_subject = norm_id_like(doc.get('subject_id') or doc.get('subject')) ds_chapter = norm_id_like(doc.get('chapter_id') or doc.get('chapter')) ds_topic = norm_id_like(doc.get('topic_id') or doc.get('topic')) # exam/subexam logic: require both if both provided, otherwise the single one if f_exam and f_sub: if not ((de == f_exam) and (ds == f_sub)): return False elif f_exam: if de != f_exam: return False elif f_sub: if ds != f_sub: return False # subject/chapter/topic exact id match if supplied if req_subject and (ds_subject != str(req_subject)): return False if req_chapter and (ds_chapter != str(req_chapter)): return False if req_topic and (ds_topic != str(req_topic)): return False return True items_raw = [d for d in items_raw if matches(d)] # prepare course title resolver (grouping key) course_title_cache = {} def resolve_course_title(course_ref): if not course_ref: return None, 'Uncategorized' if isinstance(course_ref, dict): title = course_ref.get('course_title') or course_ref.get('title') cid = course_ref.get('_id') or course_ref.get('id') return (str(cid) if cid else None), title cid = None if isinstance(course_ref, ObjectId): cid = str(course_ref) elif isinstance(course_ref, str): cid = course_ref else: try: cid = ( str(course_ref.get('$id')) if isinstance(course_ref, dict) and course_ref.get('$id') else None ) except Exception: cid = None if not cid: return None, 'Uncategorized' if cid in course_title_cache: return cid, course_title_cache[cid] try: cobj = Course.objects(id=cid).only('course_title').first() if cobj: title = getattr(cobj, 'course_title', None) or ('Course ' + cid) course_title_cache[cid] = title return cid, title except Exception: try: db2 = OldPapers._get_collection().database course_doc = ( db2['courses'].find_one({'_id': ObjectId(cid)}) if ObjectId.is_valid(cid) else None ) if course_doc: title = course_doc.get('course_title') or ('Course ' + cid) course_title_cache[cid] = title return cid, title except Exception: pass title = 'Course ' + cid course_title_cache[cid] = title return cid, title # --- robust collection lookup: include exam_list and other common names --- def pick_collection(db_, candidates): if not db_: return None for nm in candidates: if nm in db_.list_collection_names(): return db_[nm] return None exams_coll = pick_collection(db, ['exams', 'exam_list', 'exam', 'examList']) subexams_coll = pick_collection(db, ['sub_exams_list', 'subexams', 'sub_exam', 'sub_exams']) subject_coll = pick_collection( db, ['videocoursesubjects', 'video_course_subjects', 'video_course_subject', 'ebook_subjects', 'ebooksubjects'] ) chapter_coll = pick_collection(db, ['chapters', 'chapter_list', 'chapter']) topic_coll = pick_collection(db, ['topics', 'topic_list', 'topic']) def resolve_simple(val, fields, raw_coll): """ Resolve id/embedded doc to a simple label string. """ if not val: return None if isinstance(val, dict): for f in fields: if val.get(f): return val.get(f) val = val.get('$id') or val.get('_id') or val.get('id') or val idstr = str(val) if raw_coll: try: q = {'_id': ObjectId(idstr)} if ObjectId.is_valid(idstr) else {'_id': idstr} r = raw_coll.find_one(q) if r: for f in fields: if r.get(f): return r.get(f) except Exception: pass return None groups = {} for doc in items_raw: c_ref = doc.get('course_id') subj_id, subj_name = resolve_course_title(c_ref) subj_id = subj_id or 'uncategorized' subj_name = subj_name or 'Uncategorized' raw_id = doc.get('_id') or doc.get('id') id_str = str(raw_id) if raw_id is not None else '' item = { 'id': id_str, 'title': doc.get('paper_title') or doc.get('title') or doc.get('name'), 'duration': doc.get('duration'), 'number_of_questions': doc.get('number_of_questions'), 'content_file_path': ( doc.get('content_file_path') or doc.get('path') or doc.get('file_path') ), 'status': doc.get('status') or '' } # Resolve sub_exam first se_val = doc.get('sub_exam_id') or doc.get('sub_exam') se_label = ( resolve_simple( se_val, ['sub_exam_title', 'title', 'name', 'sub_exam_name'], subexams_coll ) if se_val else None ) item['sub_exam'] = se_label if se_label else '' # Resolve exam: from doc, or derived from subexam ex_val = doc.get('exam_id') or doc.get('exam') ex_label = ( resolve_simple(ex_val, ['exam_title', 'title', 'name'], exams_coll) if ex_val else None ) if not ex_label and se_val and subexams_coll: try: se_idstr = norm_id_like(se_val) if se_idstr: q = ( {'_id': ObjectId(se_idstr)} if ObjectId.is_valid(se_idstr) else {'_id': se_idstr} ) se_doc = subexams_coll.find_one(q) if se_doc: cand = ( se_doc.get('exam_id') or se_doc.get('exam') or se_doc.get('parent_exam') or se_doc.get('exam_ref') ) if cand: ex_label = resolve_simple( cand, ['exam_title', 'title', 'name'], exams_coll ) if not ex_label: ex_label = ( se_doc.get('exam_title') or se_doc.get('exam_name') or None ) except Exception: pass item['exam'] = ex_label if ex_label else '' # Resolve subject/chapter/topic s_val = doc.get('subject_id') or doc.get('subject') s_label = ( resolve_simple(s_val, ['subject_name', 'title', 'name'], subject_coll) if s_val else None ) item['subject'] = s_label if s_label else '' ch_val = doc.get('chapter_id') or doc.get('chapter') ch_label = ( resolve_simple(ch_val, ['chapter_title', 'title', 'name'], chapter_coll) if ch_val else None ) item['chapter'] = ch_label if ch_label else '' tp_val = doc.get('topic_id') or doc.get('topic') tp_label = ( resolve_simple(tp_val, ['topic_title', 'title', 'name'], topic_coll) if tp_val else None ) item['topic'] = tp_label if tp_label else '' groups.setdefault( subj_id, {'id': subj_id, 'subject_name': subj_name, 'items': []} ) groups[subj_id]['items'].append(item) subjects = list(groups.values()) # ---------- attached oldpaper ids from ContentMapping ---------- attached = [] try: if ContentMapping is not None: course_id_str = str(course_id) # normalize course_id if it is an ObjectId-like string if ObjectId.is_valid(course_id_str): course_id_str = str(ObjectId(course_id_str)) cms = ContentMapping.objects( course_id=course_id_str, content_type='oldpaper' ).only('content_id') attached = [str(m.content_id) for m in cms] else: attached = [] except Exception: current_app.logger.exception("Failed to load attached oldpaper mappings") attached = [] return jsonify( success=True, subjects=subjects, attached_item_ids=attached, attached_oldpaper_ids=attached ) except Exception: current_app.logger.exception('Error fetching old paper subjects') return jsonify(success=False, message='Server error'), 500 # ========================================================================================== @user_bp.route('/user/import-question', methods=['GET', 'POST']) def import_questions(): errors = {} general_error = None mock_tests = MockTest.objects(status=1).order_by('-created_date') if request.method == 'POST': mock_test_id = request.form.get('mock_test_id') file = request.files.get('questions_file') if not mock_test_id: errors['mock_test_id'] = 'Please select a mock test.' if not file or file.filename == '': errors['questions_file'] = 'Please upload a questions file.' mock_test = None if mock_test_id: mock_test = MockTest.objects(id=mock_test_id).first() if not mock_test: errors['mock_test_id'] = 'Invalid mock test selected.' if not errors: try: filename = file.filename.lower() if filename.endswith('.csv'): df = pd.read_csv(file) else: df = pd.read_excel(file) # ✅ REQUIRED HEADERS CHECK (exact names) required_cols = [ "question_type", "topic", "question", "option_1", "option_2", "option_3", "option_4", "answer", "marks" ] missing = [c for c in required_cols if c not in df.columns] if missing: # show clear error to user raise ValueError( "Missing required columns: " + ", ".join(missing) ) # ---- normalize column names ---- norm_map = {} for col in df.columns: key = col.strip().lower().replace(' ', '').replace('_', '') norm_map[key] = col def get_val(row, key): col = norm_map.get(key) if col is None: return '' v = row.get(col) return '' if pd.isna(v) else str(v).strip() ALLOWED_TYPES = { 'mcq': 'MCQ', 'factual': 'Factual', 'analytical': 'Analytical', 'currentaffairs': 'CurrentAffairs', 'assertionreason': 'AssertionReason', 'matchthefollowing': 'MatchTheFollowing', 'chronology': 'Chronology', 'qa': 'QA', 'essay': 'Essay', 'descriptive': 'Descriptive' } created_count = 0 option_cols_norm = [] for key, orig in norm_map.items(): if key.startswith('option') or key in ('a', 'b', 'c', 'd', 'e'): option_cols_norm.append(key) for _, row in df.iterrows(): raw_qtype = get_val(row, 'questiontype') key_qtype = raw_qtype.lower().replace(' ', '') qtype = ALLOWED_TYPES.get(key_qtype, None) question = get_val(row, 'question') topic = get_val(row, 'topic') answer = get_val(row, 'answer') raw_marks = get_val(row, 'marks') try: marks = int(raw_marks) if raw_marks else 1 except Exception: marks = 1 options = [] for opt_key in option_cols_norm: col_name = norm_map[opt_key] v = row.get(col_name) if v is not None and not pd.isna(v) and str(v).strip(): options.append(str(v).strip()) if not qtype or not question or not answer: continue mtq = MockTestQuestion( mock_test=mock_test, question_type=qtype, topic=topic, question=question, options=options, answer=answer, marks=marks ) mtq.save() created_count += 1 flash(f'{created_count} questions imported successfully.', 'success') return redirect(url_for('user.import_questions')) except Exception as e: import traceback traceback.print_exc() general_error = f"Error: {e}" return render_template( 'user/import-questions.html', mock_tests=mock_tests, errors=errors, general_error=general_error ) def _norm_lang(s: str) -> str: """Lowercase + remove all whitespace. Used for duplicate detection.""" return re.sub(r'\s+', '', (s or '')).casefold() @user_bp.route('/user/add-language', methods=['GET', 'POST']) def add_language(): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None if request.method == 'POST': language_name = (request.form.get('language_name') or '').strip() status = (request.form.get('status') or '').strip() if not language_name: errors['language_name'] = "Enter language name" if status not in ('0', '1'): errors['status'] = "Select status" # --- Duplicate check (case/space insensitive) --- norm_new = _norm_lang(language_name) if not errors: try: for existing in Language.objects.only('language_name'): if _norm_lang(existing.language_name) == norm_new: errors['language_name'] = "Language already exists" break except Exception: # Fail-safe: if query fails, don’t block user with false duplicate pass if errors: return render_template( 'user/add-language.html', errors=errors, general_error=general_error, request=request ) try: Language( language_name=language_name.strip(), status=int(status), created_date=datetime.utcnow() ).save() flash("Language added successfully", "success") return redirect(url_for('user.add_language')) except Exception: general_error = "Something went wrong while saving" return render_template( 'user/add-language.html', errors=errors, general_error=general_error, request=request ) return render_template('user/add-language.html') # ----------------------------- # VIEW LANGUAGES (filters identical style) # ----------------------------- @user_bp.route('/user/view-languages', methods=['GET']) def languages_list(): if 'user' not in session: return redirect(url_for('user.user_login')) q = (request.args.get('q') or '').strip() status = (request.args.get('status') or '').strip() # '1', '0', or '' qry = Language.objects if q: qry = qry.filter(language_name__icontains=q) if status in ('0', '1'): try: qry = qry.filter(status=int(status)) except Exception: pass langs = qry.order_by('-created_date') rows = [] for lg in langs: rows.append({ 'id': str(lg.id), 'language_name': lg.language_name, 'status': lg.status, 'created_iso': lg.created_date.isoformat() if getattr(lg, 'created_date', None) else '', 'created_pretty': lg.created_date.strftime('%Y-%m-%d %H:%M') if getattr(lg, 'created_date', None) else '-' }) filters = { 'q': q, 'status': status } return render_template( 'user/view-languages.html', rows=rows, filters=filters ) # ----------------------------- # EDIT LANGUAGE # ----------------------------- @user_bp.route('/user/edit-language/', methods=['GET', 'POST']) def edit_language(lang_id): if 'user' not in session: return redirect(url_for('user.user_login')) lang = Language.objects(id=lang_id).first() if not lang: flash("Language not found", "danger") return redirect(url_for('user.languages_list')) errors = {} general_error = None if request.method == 'POST': language_name = (request.form.get('language_name') or '').strip() status = (request.form.get('status') or '').strip() if not language_name: errors['language_name'] = "Enter language name" if status not in ('0', '1'): errors['status'] = "Select status" # --- Duplicate check (case/space insensitive), exclude current doc --- norm_new = _norm_lang(language_name) if not errors: try: for existing in Language.objects(id__ne=lang.id).only('language_name'): if _norm_lang(existing.language_name) == norm_new: errors['language_name'] = "Language already exists" break except Exception: pass if errors: return render_template( 'user/edit-language.html', language=lang, errors=errors, general_error=general_error, request=request ) try: lang.update( set__language_name=language_name.strip(), set__status=int(status) ) lang.reload() flash("Language updated successfully", "success") return redirect(url_for('user.edit_language', lang_id=lang.id)) except Exception as e: general_error = f"Save failed: {type(e).__name__}: {e}" return render_template( 'user/edit-language.html', language=lang, errors=errors, general_error=general_error, request=request ) return render_template( 'user/edit-language.html', language=lang, errors=errors, general_error=general_error, request=request ) # ----------------------------- # DELETE LANGUAGE # ----------------------------- @user_bp.route('/user/delete-language/', methods=['POST']) def delete_language(lang_id): if 'user' not in session: return jsonify(success=False, message="Unauthorized"), 401 lg = Language.objects(id=lang_id).first() if not lg: return jsonify(success=False, message="Language not found"), 404 try: lg.delete() return jsonify(success=True, message="Language deleted") except Exception: return jsonify(success=False, message="Delete failed"), 500 # ---------------------------- # Add FAQ # ---------------------------- @user_bp.route('/user/add-faq', methods=['GET', 'POST']) def add_faq(): 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') or '').strip() description = (request.form.get('description') or '').strip() status = (request.form.get('status') or '').strip() # Validations if not title: errors['title'] = "Enter FAQ title" if not description: errors['description'] = "Enter FAQ description" if not status: errors['status'] = "Select status" if errors: return render_template( 'user/add-faq.html', errors=errors, general_error=general_error, request=request ) try: Faq( title=title, description=description, status=int(status) ).save() except Exception as e: general_error = "Something went wrong" return render_template( 'user/add-faq.html', general_error=general_error, request=request ) flash("FAQ added successfully", "success") return redirect(url_for('user.add_faq')) return render_template('user/add-faq.html') # ---------------------------- # View FAQs list # ---------------------------- @user_bp.route('/user/view-faqs', methods=['GET']) def faqs_list(): if 'user' not in session: return redirect(url_for('user.user_login')) faqs = Faq.objects().order_by('-created_date') # latest first return render_template('user/view-faqs.html', faqs=faqs) # ---------------------------- # Delete FAQ # ---------------------------- @user_bp.route('/user/delete-faq/', methods=['POST']) def delete_faq(faq_id): if 'user' not in session: return jsonify({'success': False, 'message': 'Unauthorized access'}), 401 try: faq = Faq.objects(id=faq_id).first() if not faq: return jsonify({'success': False, 'message': 'FAQ not found'}), 404 faq.delete() return jsonify({'success': True, 'message': 'FAQ deleted successfully'}) except Exception as e: return jsonify({'success': False, 'message': 'Failed to delete FAQ'}), 500 # ---------------------------- # Edit FAQ # ---------------------------- @user_bp.route('/user/edit-faq/', methods=['GET', 'POST']) def edit_faq(faq_id): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None faq = Faq.objects(id=faq_id).first() if not faq: flash('FAQ not found.', 'danger') return redirect(url_for('user.faqs_list')) if request.method == 'POST': title = (request.form.get('title') or '').strip() description = (request.form.get('description') or '').strip() status = (request.form.get('status') or '').strip() # Validations if not title: errors['title'] = "Enter FAQ title" if not description: errors['description'] = "Enter FAQ description" if not status: errors['status'] = "Select status" if errors: return render_template( 'user/edit-faq.html', errors=errors, faq=faq, general_error=general_error, request=request ) try: faq.update( title=title, description=description, status=int(status) ) flash("FAQ updated successfully", "success") # Reload document to reflect latest data in the template faq.reload() return render_template('user/edit-faq.html', faq=faq) except Exception as e: general_error = "Something went wrong" return render_template( 'user/edit-faq.html', general_error=general_error, faq=faq, request=request ) return render_template('user/edit-faq.html', faq=faq) @user_bp.route('/user/view-subscribers', methods=['GET']) def subscriptions_list(): if 'user' not in session: return redirect(url_for('user.user_login')) subscribers = Subscription.objects().order_by('-created_date') # latest first return render_template('user/view-subscribers.html', subscribers=subscribers) @user_bp.route('/user/view-educoins-topup', methods=['GET']) def educoins_topup_list(): if 'user' not in session: return redirect(url_for('user.user_login')) topups = EducoinsTopup.objects().order_by('-created_date') return render_template('user/view-educoins-topup.html', topups=topups) @user_bp.route('/user/delete-educoins-topup/', methods=['POST']) def delete_educoins_topup(topup_id): if 'user' not in session: return jsonify({'success': False, 'message': 'Unauthorized access'}), 401 try: topup = EducoinsTopup.objects(id=topup_id).first() if not topup: return jsonify({'success': False, 'message': 'Topup not found'}), 404 topup.delete() return jsonify({'success': True, 'message': 'Educoins topup deleted successfully'}) except Exception as e: return jsonify({'success': False, 'message': 'Failed to delete topup'}), 500 @user_bp.route('/user/add-educoins-topup', methods=['GET', 'POST']) def add_educoins_topup(): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None if request.method == 'POST': topup_coins = (request.form.get('topup_coins') or '').strip() topup_coin_value = (request.form.get('topup_coin_value') or '').strip() status = (request.form.get('status') or '').strip() # --- VALIDATIONS --- # Topup coins validation if not topup_coins: errors['topup_coins'] = "Enter number of coins" else: try: topup_coins_int = int(topup_coins) if topup_coins_int <= 0: errors['topup_coins'] = "Coins must be a positive number" except ValueError: errors['topup_coins'] = "Coins must be numeric" # Topup coin value validation + uniqueness if not topup_coin_value: errors['topup_coin_value'] = "Enter coin value" else: try: topup_coin_value_int = int(topup_coin_value) if topup_coin_value_int <= 0: errors['topup_coin_value'] = "Coin value must be a positive number" else: # 🔹 Uniqueness check for topup_coin_value existing = EducoinsTopup.objects(topup_coin_value=topup_coin_value_int).first() if existing: errors['topup_coin_value'] = "Coins are already top up for the entered value" except ValueError: errors['topup_coin_value'] = "Coin value must be numeric" # Status validation if not status: errors['status'] = "Select status" # If any errors, re-render form with errors if errors: return render_template( 'user/add-educoins-topup.html', errors=errors, general_error=general_error, request=request ) # --- SAVE TO DB --- try: EducoinsTopup( topup_coins=topup_coins_int, topup_coin_value=topup_coin_value_int, status=int(status) ).save() except Exception as e: general_error = "Something went wrong" return render_template( 'user/add-educoins-topup.html', general_error=general_error, request=request ) flash("Educoins topup added successfully", "success") return redirect(url_for('user.add_educoins_topup')) # GET request return render_template('user/add-educoins-topup.html') @user_bp.route('/user/edit-educoins-topup/', methods=['GET', 'POST']) def edit_educoins_topup(topup_id): if 'user' not in session: return redirect(url_for('user.user_login')) errors = {} general_error = None topup = EducoinsTopup.objects(id=topup_id).first() if not topup: flash('Educoins topup not found.', 'danger') return redirect(url_for('user.educoins_topup_list')) if request.method == 'POST': topup_coins = (request.form.get('topup_coins') or '').strip() topup_coin_value = (request.form.get('topup_coin_value') or '').strip() status = (request.form.get('status') or '').strip() # --- VALIDATIONS --- # Topup coins validation if not topup_coins: errors['topup_coins'] = "Enter number of coins" else: try: topup_coins_int = int(topup_coins) if topup_coins_int <= 0: errors['topup_coins'] = "Coins must be a positive number" except ValueError: errors['topup_coins'] = "Coins must be numeric" # Topup coin value validation + uniqueness (excluding current record) if not topup_coin_value: errors['topup_coin_value'] = "Enter coin value" else: try: topup_coin_value_int = int(topup_coin_value) if topup_coin_value_int <= 0: errors['topup_coin_value'] = "Coin value must be a positive number" else: # 🔹 Check if another record already has this value existing = EducoinsTopup.objects(topup_coin_value=topup_coin_value_int).first() if existing and str(existing.id) != str(topup.id): errors['topup_coin_value'] = "Coins are already top up for the entered value" except ValueError: errors['topup_coin_value'] = "Coin value must be numeric" # Status validation if not status: errors['status'] = "Select status" # If any errors, re-render form with errors if errors: return render_template( 'user/edit-educoins-topup.html', errors=errors, topup=topup, general_error=general_error, request=request ) # --- UPDATE IN DB --- try: topup.update( topup_coins=topup_coins_int, topup_coin_value=topup_coin_value_int, status=int(status) ) topup.reload() flash("Educoins topup updated successfully", "success") return render_template('user/edit-educoins-topup.html', topup=topup) except Exception as e: general_error = "Something went wrong" return render_template( 'user/edit-educoins-topup.html', general_error=general_error, topup=topup, request=request ) # GET request return render_template('user/edit-educoins-topup.html', topup=topup)