User Registration and Email Verification
User Registration and Email Verification
[Link]("confirm_password")
def register_user():
try:
"""
db = next(get_db())
User registration with email verification
"""
# Check if email already exists
data = request.get_json()
existing_user =
[Link](User).filter([Link] == email).first()
# Extract and validate user data
if existing_user:
full_name = [Link]("full_name")
if existing_user.is_verified:
email = [Link]("email")
return jsonify({"status": "error",
password = [Link]("password") "message": "Email already registered"}), 409
else:
# Resend verification email for [Link](new_user)
unverified users
[Link]()
send_verification_email(existing_user,
db) [Link](new_user)
if not token:
return jsonify({"status": "error", except Exception as e:
"message": "Verification token is missing"}), if db:
400
[Link]()
current_app.[Link](f"Email
# Find user by verification token verification error: {str(e)}")
user = return jsonify({"status": "error", "message":
[Link](User).filter(User.verification_token == "Verification failed"}), 500
token).first()
finally:
if db: return jsonify({"status": "error",
"message": "Invalid credentials"}), 401
[Link]()
# Role-based validation
if [Link] == "admin" and
def login_user(): is_teacher_login:
""" return jsonify({"status": "error",
User login with role-based validation "message": "Admin users must log in through
admin portal"}), 403
"""
data = request.get_json()
if [Link] == "program_chair" and not
is_teacher_login:
identifier = data["email"] # Can be email or return jsonify({"status": "error",
username "message": "Course selection required for
Program Chair"}), 400
password = data["password"]
course_id_str = [Link]("course") # For
program chair login # Check user status
if not user.is_verified:
is_teacher_login = bool(course_id_str) return jsonify({"status": "error",
"message": "Account not verified"}), 403
try:
if not user.is_active:
db = next(get_db())
return jsonify({"status": "error",
"message": "Account is inactive"}), 403
# Find user by email or username
if "@" in identifier:
# Validate course assignment for program
user = [Link](User).filter([Link] chairs
== identifier).first()
if [Link] == "program_chair":
else:
if not course_id_str:
user =
return jsonify({"status": "error",
[Link](User).filter([Link] ==
"message": "Course selection required"}), 400
identifier).first()
try:
if not user or not
check_password_hash([Link], course_id = int(course_id_str)
password):
except ValueError:
return jsonify({"status": "error",
"message": "Invalid course ID"}), 400
try:
db = next(get_db())
if user.category_id != course_id:
user = [Link](User).filter([Link] ==
return jsonify({"status": "error", email).first()
"message": "Access denied to selected
program"}), 403
if not user:
# Email configuration
smtp_host = "[Link]-
def forgot_password(): [Link]"
try:
except Exception as e:
# Send reset email
[Link]()
message =
MIMEMultipart("alternative") current_app.[Link](f"Password reset
error: {str(e)}")
message["Subject"] = "Password Reset
Request" return jsonify({"status": "error", "message":
f"Server error: {str(e)}"}), 500
message["From"] = f"Password Reset
<{sender_email}>"
message["To"] = email
def reset_password():
//=======[Link]======//
category =
from flask import jsonify, request, session [Link](Category).filter([Link] ==
from [Link] import category_id).first()
Category if not category:
from [Link] import Subject return jsonify({"status": "error",
from [Link] import SQLAlchemyError "message": "Category not found"}), 404
new_category = try:
Category(name=data["name"], color=color) # Only admin can update
[Link](new_category) user_role = [Link]("role")
[Link]() if user_role not in ("admin",):
[Link](new_category) return jsonify({"status": "error",
"message": "Only admin can update
categories"}), 403
# Process subject assignments if provided
if "subjects" in data and
isinstance(data["subjects"], list): category =
[Link](Category).filter([Link] ==
process_subject_assignments(db, category_id).first()
new_category.id, data["subjects"])
if not category:
return jsonify({"status": "error",
return jsonify({ "message": "Category not found"}), 404
# Update fields if provided - Removes category from all subjects
if "name" in data: - Deletes the category
[Link] = data["name"] """
if "color" in data: try:
[Link] = data["color"] # Only admin can delete
user_role = [Link]("role")
# Process subject assignments if provided if user_role not in ("admin",):
if "subjects" in data and return jsonify({"status": "error",
isinstance(data["subjects"], list): "message": "Only admin can delete
categories"}), 403
process_subject_assignments(db,
category_id, data["subjects"])
category =
[Link](Category).filter([Link] ==
[Link]() category_id).first()
[Link](category) if not category:
return jsonify({"status": "error",
return jsonify({ "message": "Category not found"}), 404
"status": "success",
"message": "Category updated # Remove category from all subjects
successfully", subjects =
"data": category.to_dict(), [Link](Subject).filter(Subject.category_id ==
category_id).all()
}), 200
for subject in subjects:
subject.category_id = None
except SQLAlchemyError as e:
[Link] = 1.0
[Link]()
return jsonify({"status": "error", "message":
str(e)}), 500 [Link](category)
[Link]()
"""
Helper function to process subject [Link]()
assignments to categories
- Clears existing category assignments //=======[Link]========//
- Assigns new subjects with weights
""" from flask import jsonify, request
# Get all subjects and create lookup dict from [Link] import Class,
subjects = [Link](Subject).all() Student
"name": data["name"],
"average_score": None, return jsonify({"status": "success", "data":
class_dict}), 200
"exams_taken": 0,
"passed": None,
def create_class(db: Session):
})
"""
student_dict["status"] = {
Create a new class for the current user.
"passed": passed if
any(s["exams_taken"] > 0 for s in """
subject_details) else None, data = [Link]
"subjects": subject_details, if "name" not in data:
} return jsonify({"status": "error", "message":
else: "Missing required field: name"}), 400
[Link](student)
[Link]() def update_student(db: Session, class_id: int,
student_id: int):
if student not in class_obj.students:
"""
class_obj.[Link](student)
Update student info in a class (name,
[Link]() student_id, scores, etc.).
[Link](student) """
return jsonify({"status": "success", data = [Link]
"message": "Student added to class
successfully", "student": student.to_dict()}), 200 class_obj = [Link](Class).filter([Link] ==
class_id).first()
if not class_obj:
def remove_student_from_class(db: Session,
class_id: int, student_id: int): return jsonify({"status": "error", "message":
"Class not found"}), 404
"""
student = [Link](Student).filter([Link] import sys
== student_id).first()
import subprocess
if not student:
from sqlalchemy import desc, inspect, text
return jsonify({"status": "error", "message":
f"Student with ID {student_id} not found"}), from [Link] import joinedload
404 from [Link] import get_db
if student not in class_obj.students: from [Link] import Exam,
return jsonify({"status": "error", "message": exam_questions, ExamResult
"Student is not in this class"}), 400 from [Link] import Question,
# Update fields Subject, Option, SituationalQuestion
[Link](stmt) db = next(get_db())
user_id = [Link]("user_id") # Get questions with images
query = [Link](Exam) query =
[Link](Question).join(exam_questions).filter(
exam_questions.c.exam_id ==
if user_id: exam_id).order_by(exam_questions.c.question_
order)
query = [Link]((Exam.user_id ==
user_id) | (Exam.user_id == None)) questions = [Link]()
else:
query = [Link](Exam.user_id == result = {
None) "id": [Link],
"title": [Link],
if recent: "course": [Link],
query = "num_items": exam.num_items,
query.order_by(Exam.created_at.desc())
"question_bank": exam.question_bank,
if limit:
"question_bank_name":
query = [Link](limit) question_bank_name,
"class_id": exam.class_id,
exams = [Link]() "date_created":
return [exam_to_dict(exam) for exam in exam.created_at.isoformat() if exam.created_at
exams] else None,
"questions": [{
def select_questions_for_exam(course,
question_bank, total_items, class_id=None): # Standard fallback selection
""" available_questions =
[Link](Question).filter(
Select questions for exam with ML-based
selection fallback Question.subject_id == subject_id,
selected_questions = [] else:
[Link](situational_groups)
items_to_shuffle = list(temp_groups.values())
+ temp_standalone
# Add complete situational groups
[Link](items_to_shuffle)
for group in situational_groups:
if len(group) <= remaining_items:
for item in items_to_shuffle:
selected_questions.extend(group)
if isinstance(item, list):
remaining_items -= len(group)
final_ordered_selection.extend(item)
if remaining_items == 0:
else:
break
final_ordered_selection.append(item)
# Determine format
sample_row = sample_rows[0] # Generate answer key from sequential
questions
is_test_format = "studname" in sample_row
and "idnum" in sample_row and any(f"item{i}" label_to_number = {"A": 1, "B": 2, "C": 3,
in sample_row for i in range(1, 10)) "D": 4}
for idx, q in enumerate(ml_questions):
answer_key = {} correct_option_num = 0
if exam.is_for_ml_training: correct_option_num =
label_to_number.get(option.option_label, 0)
# ML Mode: Use sequential questions from
subject break
ml_questions = [] answer_key[question_num] =
correct_option_num
custom_sequence_ids =
exam.ml_custom_sequence else:
# Standard Mode: Use exam-specific
questions
if custom_sequence_ids and
isinstance(custom_sequence_ids, list) and questions_for_exam =
len(custom_sequence_ids) == get_exam_questions(exam_id)
num_items_to_process:
label_to_number = {"A": 1, "B": 2, "C": 3, student_name = [Link]("studname",
"D": 4} "").strip()
actual_question_count = 0 student_id = [Link]("idnum",
"").strip()
for item in questions_for_exam:
responses = []
if not [Link]("is_situational_header",
False): detailed_results = {}
actual_question_count += 1
question_num = for i in range(1,
[Link]("display_number") num_items_to_process + 1):
if question_num: item_key = f"item{i}"
correct_option_num = 0 question_num_for_key = i
for option in [Link]("options", []):
if [Link]("is_correct"): if item_key in row:
correct_option_num = try:
label_to_number.get([Link]("option_label"),
0) answer = int(row[item_key]) if
row[item_key].strip() else 0
break
[Link](answer)
answer_key[question_num] =
correct_option_num correct_answer =
answer_key.get(question_num_for_key, 0)
is_correct = 1 if answer ==
# Ensure answer key has entries for all correct_answer else 0
expected items
detailed_results[str(i)] = {
for i in range(1, num_items_to_process +
1): "response": answer,
},
) ordered_questions_result =
ordered_questions_query.all()
if not ordered_questions_result:
results["processed"] += 1
return []
results["details"].append(f"Processed
{student_id}:
{raw_score}/{num_items_to_process}") processed_list = []
processed_situational_ids = set()
# Commit periodically display_counter = 1
if row_idx > 0 and row_idx % 10 == 0:
for question in ordered_questions_result: question.situational_question else "Context not
found",
question_data = {
"image_path":
"id": [Link], question.situational_question.image_path if
"display_number": display_counter, question.situational_question else None,
"question_text": question.question_text, })
"question_image_path":
question.question_image_path, processed_situational_ids.add([Link]
al_id)
"topic": [Link],
"options": [{
processed_list.append(question_data)
"option_label": option.option_label,
display_counter += 1
"option_text": option.option_text,
"option_image_path":
option.option_image_path, return processed_list
"is_correct": option.is_correct,
} for option in sorted([Link], def exam_to_dict(exam):
key=lambda o: o.option_label)], """Convert Exam object to dictionary"""
"correct_answer": next((opt.option_label questions_data = []
for opt in [Link] if opt.is_correct),
None), if not exam.is_for_ml_training:
"situational_id": question.situational_id, for question in [Link]:
"is_situational_subquestion": question_dict = {
bool(question.situational_id),
"id": [Link],
}
"question_text":
question.question_text,
# Add situational context once before first "subject_id": question.subject_id,
sub-question
"topic": [Link],
if question.situational_id and
question.situational_id not in "credit": [Link],
processed_situational_ids: "options": [{
processed_list.append({ "id": [Link],
"is_situational_header": True, "option_text": option.option_text,
"situational_id": "is_correct": option.is_correct,
question.situational_id,
"option_label": option.option_label,
"description":
question.situational_question.description if } for option in [Link]],
}
if question.situational_question and # Additional utility functions (update_exam,
question.situational_question.questions[0].id == delete_exam, get_alternative_question,
[Link]: replace_question)
# ... [similar summarization pattern for
question_dict["situational_description"] = remaining functions]
question.situational_question.description
//=====[Link]=======//
questions_data.append(question_dict)
import csv
import os
class_data = None
import uuid
if exam.class_relation:
import logging
class_data = {
from sqlalchemy import exc
"id": exam.class_relation.id,
from [Link] import secure_filename
"name": exam.class_relation.name,
from [Link] import get_db
"section": exam.class_relation.section,
from [Link] import Question,
} Option, Subject, SituationalQuestion
from flask import session
return {
"id": [Link], UPLOAD_FOLDER =
[Link]([Link]([Link](__
"title": [Link], file__)), "static", "uploads", "questions")
"course": [Link], ALLOWED_EXTENSIONS = {"png", "jpg",
"num_items": exam.num_items, "jpeg", "gif"}
"class_id": exam.class_id,
"class": class_data, def allowed_file(filename):
[Link](UPLOAD_FOLDER,
exist_ok=True) # Apply filters
filename = secure_filename([Link]) if category:
unique_filename = subjects =
f"{uuid.uuid4()}_{filename}" [Link](Subject).filter(Subject.category_id ==
category).all()
file_path = [Link](UPLOAD_FOLDER,
unique_filename) subject_ids = [[Link] for subject in
subjects]
db_path =
f"backend/static/uploads/questions/{unique_file query =
name}" [Link](Question.subject_id.in_(subject_ids)
)
elif subject:
try:
query = [Link](Question.subject_id ==
[Link](file_path) subject)
return db_path
except Exception as e: if topic:
[Link](f"Error saving file {filename}: query = [Link]([Link] ==
{str(e)}") topic)
return None if search:
query =
def get_all_questions(category=None, [Link](Question.question_text.like(f"%
subject=None, search=None, page=1, {search}%"))
page_size=20, topic=None):
"""Get questions with filtering, pagination, # Pagination
and user-based access control"""
total_items = [Link]()
db = next(get_db())
total_pages = (total_items + page_size - 1) //
query = [Link](Question) page_size
offset = (page - 1) * page_size
# Filter by user_id from session questions =
user_id = [Link]("user_id") query.order_by(Question.created_at.desc()).offse
t(offset).limit(page_size).all()
if user_id:
query = [Link]((Question.user_id ==
user_id) | (Question.user_id == None)) return {
}, option_image_path =
save_uploaded_image(option_images[str(i)])
}
option = Option(
def create_question(data, question_image=None,
option_images=None): question_id=[Link],
option_label=option_data.get("option_label",
question = Question( chr(65 + i))
question_text=data["question_text"], )
subject_id=data["subject_id"], [Link](option)
topic=[Link]("topic"),
credit=[Link]("credit"), [Link]()
status=[Link]("status", "pending"), [Link](question)
user_id=user_id return question_to_dict(question)
)
def create_group_question(form_data,
files_data):
if question_image and
question_image.filename: """Create situational question group with sub-
questions and options"""
question.question_image_path =
save_uploaded_image(question_image) db = next(get_db())
user_id = [Link]("user_id")
[Link](question)
[Link]() situational_description =
form_data.get("situational_description",
"").strip()
# Create options
subject_id = form_data.get("subject_id")
if not situational_description or not # Handle sub-question image
subject_id:
sub_question_image_path = None
raise ValueError("Situational description
and subject ID are required") sub_question_image_key =
f"sub_question_image_{i}"
if sub_question_image_key in files_data
# Create situational question and
files_data[sub_question_image_key].filename !=
situational_question = "":
SituationalQuestion(description=situational_des
cription, user_id=user_id) sub_question_image_path =
save_uploaded_image(files_data[sub_question_i
mage_key])
if 'situational_image' in files_data and
files_data['situational_image'].filename:
# Create sub-question
image_path =
save_uploaded_image(files_data['situational_im sub_question = Question(
age'])
question_text=sub_question_text,
situational_question.image_path =
image_path subject_id=subject_id,
topic=sub_topic,
[Link](situational_question) status="pending",
[Link]() situational_id=situational_question.id,
question_image_path=sub_question_image_path
# Process sub-questions ,
sub_question_count = user_id=user_id
int(form_data.get("sub_question_count", 0))
)
created_sub_question_ids = []
[Link](sub_question)
[Link]()
for i in range(sub_question_count):
sub_question_text = created_sub_question_ids.append(sub_question.i
form_data.get(f"sub_question_text_{i}", d)
"").strip()
sub_topic =
form_data.get(f"sub_topic_{i}", "") # Create options for sub-question
sub_option_count =
int(form_data.get(f"sub_option_count_{i}", 0))
if not sub_question_text:
continue
sub_correct_option_index = [Link]()
int(form_data.get(f"sub_correct_option_{i}", -
1)) return {
"situational_id": situational_question.id,
[Link]() }
} else:
raise ValueError(f"Row {row_num}:
Invalid question_type")
option_count = 0
correct_option_added = False
except Exception as e:
for label in possible_options:
error_count += 1
text = options_map.get(label,
"").strip() error_details.append(f"Row {row_num}:
{str(e)}")
if not text:
[Link]()
continue
if error_count == 0:
is_correct = (label ==
correct_option_label) [Link]()
if is_correct: else:
option_count += 1 imported_single_count = 0
imported_group_count = 0
question_id=new_question.id,
option_text=text, return {
) "imported_subs": imported_sub_count,
"options": [],
} # Additional utility functions (update_question,
delete_question, revise_question, etc.)
# ... [similar summarization pattern for
if question.situational_question: remaining functions]
result["situational"] = { //=======[Link]=========//
"id": question.situational_question.id, from flask import jsonify, request, session
"description": from [Link] import Subject
question.situational_question.description,
from [Link] import
"image_path": Category
question.situational_question.image_path,
from [Link] import SQLAlchemyError
"created_at":
question.situational_question.created_at.isoform from [Link] import Session
at() if question.situational_question.created_at from [Link] import get_db
else None,
"updated_at":
question.situational_question.updated_at.isofor def get_all_subjects(db, course=None):
mat() if question.situational_question.updated_at
"""Get subjects filtered by user's category and
else None
optional course filter"""
}
query = [Link](Subject)
user_category_id = [Link]('category_id')
for option in [Link]:
if user_category_id: 'status': 'success',
query = [Link](Subject.category_id == 'message': 'Subject created successfully',
user_category_id)
'data': new_subject.to_dict()
if course:
}), 201
query = [Link](Subject.category_id ==
course)
def update_subject(db: Session, subject_id: int):
name=data['name'], if data['category_id']:
description=[Link]('description'), category =
[Link](Category).filter([Link] ==
category_id=[Link]('category_id'), data['category_id']).first()
weight=[Link]('weight', 1.0) if not category:
) return jsonify({'status': 'error',
'message': 'Category not found'}), 404
subject.category_id = data['category_id']
[Link](new_subject)
[Link]()
# Update weight
[Link](new_subject)
if 'weight' in data and data['weight'] is
not None:
return jsonify({ try:
[Link] = def get_subject_by_id(db: Session, subject_id:
float(data['weight']) int):
except (ValueError, TypeError): """Get specific subject by ID"""
[Link] = 1.0 subject = [Link](Subject).filter([Link]
== subject_id).first()
elif not [Link]:
if not subject:
[Link] = 1.0
return jsonify({'status': 'error', 'message':
else: 'Subject not found'}), 404
subject.category_id = None
[Link] = None return jsonify({'status': 'success', 'data':
subject.to_dict()}), 200
[Link]()
[Link](subject) //============middleware===========//
//=======[Link]=========//
[Link](subject) db = next(get_db())
[Link]() try:
# Check if user is logged in via session
first
return jsonify({'status': 'success', 'message':
'Subject deleted successfully'}), 200 user_id = [Link]("user_id")
if user_id:
# Validate user exists else auth_header
user = [Link](User).filter([Link] == )
user_id).first()
payload = [Link](
if user:
token,
# User is authenticated
return f(*args, **kwargs) [Link]("JWT_SECRET_KEY",
"default_secret_key"),
algorithms=["HS256"],
# If not using session auth, try token auth
)
auth_header =
[Link]("Authorization")
if not auth_header: # Check if token is expired
# Check if the request wants JSON if
response [Link](payload["exp"]) <
[Link]():
if (
# Redirect to home page for expired
token
request.accept_mimetypes.accept_json
return redirect("/")
and not
request.accept_mimetypes.accept_html
): user = [Link](User).filter([Link] ==
payload["sub"]).first()
return (
if not user:
jsonify(
# Redirect to login page if user not
{"status": "error", "message": found
"Authentication required"}
if (
),
401, request.accept_mimetypes.accept_json
) and not
# Otherwise redirect to login page for request.accept_mimetypes.accept_html
HTML requests ):
return redirect("/") return (
jsonify({"status": "error",
try: "message": "User not found"}),
token = ( 404,
auth_header.split(" ")[1] )
if ( def generate_token(user_id):
try:
request.accept_mimetypes.accept_json # Token expires in 24 hours
and not payload = {
request.accept_mimetypes.accept_html
"exp": [Link]() +
): timedelta(days=1),
return jsonify({"status": "error", "iat": [Link](),
"message": "Invalid token"}), 401
"sub": user_id,
return redirect("/")
}
except Exception as e:
# Redirect to login page for any
authentication error return [Link](
if ( payload,
[Link]("JWT_SECRET_KEY", if not [Link]("logged_in"):
"default_secret_key"),
return redirect("/")
algorithm="HS256",
# Check session for program_chair or
) admin role
except Exception as e: user_role = [Link]("role")
return str(e) if user_role not in ["program_chair",
"admin"]:
# Always return JSON for API routes on
def admin_required(f): insufficient privileges
@wraps(f) return (
def decorated_function(*args, **kwargs): jsonify(
# If not logged in, redirect to index {
if not [Link]("logged_in"): "status": "error",
return redirect("/") "message": "Program Chair or
# Check session for admin role only Administrator access required",
if [Link]("role") != "admin": }
# Updates existing subjects to have proper # Updates user roles to use consistent values
category associations # Converts old role formats to new standardized
//=010_add_user_id_to_categories.py==// ones
"""Add user_id to categories table""" # Handles role migration for existing users
//========[Link]===========// updated_at =
Column(DateTime(timezone=True))
from sqlalchemy import Column, Integer, String,
ForeignKey, DateTime, JSON, Table, Boolean
ml_custom_sequence = Column(JSON, student = relationship("Student",
nullable=True) foreign_keys=[student_db_id],
back_populates="results")
# Relationships
def to_dict(self): ...
class_relation = relationship("Class",
back_populates="exams") //======[Link]============//
questions = relationship("Question", from sqlalchemy import Column, Integer, String,
secondary=exam_questions, ForeignKey, Boolean, DateTime, Text
order_by=exam_questions.c.question_order,
lazy="joined") from [Link] import relationship
@[Link]("/api/algorithm/compare",
@[Link]("/api/auth/logout",
methods=["POST"])
methods=["POST"])
# Compares different ML algorithms on exam
# Logout (requires authentication)
data
//=======[Link]=======//
# CRUD endpoints for categories (admin only)
# Endpoints for exam CRUD, question selection,
OMR result upload, analytics, ML integration
@[Link]("/api/categories", methods=["GET",
"POST"])
# List all categories, create new category @[Link]("/api/exams", methods=["GET",
"POST"])
# List, create exams
@[Link]("/api/categories/<int:category_id>",
methods=["GET", "PUT", "DELETE"])
# Get, update, or delete a specific category @[Link]("/api/exams/<int:exam_id>",
methods=["GET", "PUT", "DELETE"])
# Get, update, delete specific exam
# Assign subjects to categories, manage
color/weight
//========[Link]============// @[Link]("/api/exams/<int:exam_id>/print",
methods=["GET", "POST"])
# CRUD endpoints for classes and student
management # Print exam
@[Link]("/api/classes/<int:class_id>", @[Link]("/api/exams/import-omr",
methods=["GET", "PUT", "DELETE"]) methods=["POST"])
# Get, update, or delete a specific class # Upload OMR results
@[Link]("/api/classes/<int:class_id>/ @[Link]("/api/exams/<int:exam_id>/ml-
students", methods=["POST"]) question-sequence", methods=["GET",
"POST"])
# Add student to class
# ML question sequence endpoints
@[Link]("/api/classes/<int:class_id>/
students/<int:student_id>", # Handles exam creation, question assignment,
methods=["DELETE", "PUT"]) result processing, reporting, ML analytics
# Remove or update student in class //======[Link]=========//
# Utility endpoints for fixing and validating
image paths
# Returns class details, student status, and exam
summary
//=======[Link]============// @[Link]("/api/image-fix/fix-paths",
methods=["POST"])
# Fixes image paths for questions/options @[Link]("/api/questions/import",
methods=["POST"])
# Import questions from CSV
@[Link]("/api/image-fix/check-paths",
methods=["GET"])
# Checks for missing/broken image paths @[Link]("/api/questions/group",
methods=["POST"])
//======[Link]==========//
# Create group (situational) questions
# Endpoints for ML model training, prediction,
and analysis
@[Link]("/api/questions/<int:question_id>/
revise", methods=["POST"])
@[Link]("/api/ml/train", methods=["POST"])
# Revise question
# Train ML model
@media print {
{% if not answer_key_only %} <h2>Answer Key -
{{ [Link] }}</h2>
<!-- Student Info Section -->
<div class="answer-grid">
<div class="student-info">
{% for item in
<div>Name: <span processed_items %}
class="line"></span></div>
{% if [Link] is
<div>Date: <span defined %}
class="line"></span></div>
{% for sub_q in
</div> item.sub_questions %}
<div class="student-info">
<div>Section: <span <div>{{ sub_q.display_number }}.
class="line"></span></div> {% for option in
<div>Score: <span sub_q.options %}
class="line"></span></div> {% if
</div> option.is_correct %}{{ option.option_label }}
{% endif %}
{% endfor %}
<!-- Instructions -->
</div>
<div class="instructions">
{% endfor %}
<h4>Instructions:</h4>
{% else %}
<ul>
<li>Read each question <div>{{ item.display_number }}.
carefully.</li>
{% for option in
<li>Choose the best answer for [Link] %}
multiple-choice questions.</li>
{% if
<li>Write your answers clearly option.is_correct %}{{ option.option_label }}
on the provided space or answer sheet.</li> {% endif %}
<li>Ensure your name and {% endfor %}
section are written correctly.</li>
</div>
</ul>
{% endif %}
</div>
{% endfor %}
</div>
<!-- Questions Section -->
</div>
<div class="questions-section">
{% else %}
{% if answer_key_only %}
{% for item in processed_items %}
<div class="answer-key">
{% if [Link] is defined {% for option in
%} sub_q.options %}
<div class="situational-group"> <div class="option">
<div class="situational-
description"> <span>{{ option.option_label }}.</span>
<div>
<p><strong>{{ [Link]
}}</strong></p> {{ option.option_text
}}
</div>
{% if
{% for sub_q in option.option_image_path %}
item.sub_questions %}
<div class="option-
<div class="question-item"> image-container">
<div class="question- <img
header"> src="/{{ option.option_image_path }}"
alt="Option Image" class="option-image"
<span class="question-
display-
number">{{ sub_q.display_number }}.</span> onerror="[Link]=null;
[Link]='/frontend/images/image-not-
<div class="question-text- [Link]';">
content">
</div>
<p>{{ sub_q.question_text }}</p> {% endif %}
{% if </div>
sub_q.question_image_path %}
</div>
<div class="question-
image-container"> {% endfor %}
<img </div>
src="/{{ sub_q.question_image_path }}" </div>
alt="Question Image" class="question-image"
{% endfor %}
onerror="[Link]=null; </div>
[Link]='/frontend/images/image-not-
{% else %}
[Link]';">
<div class="question-item">
</div>
<div class="question-header">
{% endif %}
<span class="question-
</div>
display-number">{{ item.display_number
</div> }}.</span>
<div class="options"> <div class="question-text-
content">
{% endif %}
<p>{{ item.question_text }}</p>
</div>
{% if
item.question_image_path %} </div>
[Link]('img').forEach(funct //========[Link]===========//
ion(img) { # (Empty or placeholder file)
[Link]("Image src:",
[Link]);
//=========scripts============//
if img_path:
# Check image paths for each option
# Check if path is absolute or relative
for j, opt in enumerate([Link]("options", [])):
opt_img_path = except ValueError:
[Link]("option_image_path")
print("Please provide a valid exam ID
[Link]( (integer)")
f" Option {[Link]('option_label')}: [Link](1)
Image path: {opt_img_path}"
except Exception as e:
)
print(f"Error: {str(e)}")
[Link](1)
if opt_img_path:
# Check if path is absolute or relative
if opt_img_path.startswith("/"):
full_path = opt_img_path
//==========util===================//
else:
//========[Link]========//
full_path = [Link](
# Database connection utilities for SQLAlchemy
[Link]([Link]([Link]
e(__file__))), from sqlalchemy import create_engine
opt_img_path, from [Link] import declarative_base,
) sessionmaker, scoped_session
from dotenv import load_dotenv
if __name__ == "__main__":
# Thread-safe session factory
if len([Link]) < 2:
SessionLocal =
print("Usage: python scoped_session(sessionmaker(autocommit=Fals
debug_exam_images.py <exam_id>") e, autoflush=False, bind=engine))
[Link](1)
# Declarative base for models
try: Base = declarative_base()
exam_id = int([Link][1])
debug_exam_images(exam_id) def get_db():
# Yields a database session (for use in
routes/controllers)
//=====[Link]========//
...
# Session management utilities for Flask
@contextmanager
from flask import session
def db_session():
# Context manager for DB sessions
(commit/rollback/close) def set_session_user(user):
import logging
def clear_session():
# Clears all session data (logout)
def setup_logging():
...
# Sets up root logger and ML logger
# Configures file and console handlers, log
format, and log level def get_current_user():
... ...
//=====migration_runner.py========// //========.env===========//
def run_specific_migration(migration_name):
# Runs a specific migration script by name
//======run_migrations.py=========//
# Returns True if successful, False otherwise
"""
...
Script to run database migrations manually
""" # If no args provided, run the
detailed_results migration
run_migration("add_detailed_results")
import importlib
import os
//======[Link]==============//
import sys
import sys
import os
def run_migration(migration_name):
import logging
"""Run a specific migration by name"""
import secrets
try:
import re
# Import the migration module
from datetime import timedelta
migration =
importlib.import_module(f"migrations.
{migration_name}")
# Add dotenv support
from dotenv import load_dotenv
# Run the upgrade function
print(f"Running migration:
{migration_name}") load_dotenv()
[Link]()
print(f"Migration {migration_name} # Add the parent directory to the path so we can
completed successfully") import modules properly
run_migration(migration_name) make_response,
else: render_template,
)
from [Link] import get_db, ) # Import the new image fix routes
Base, engine
from [Link].logging_config import
setup_logging from [Link]
import (
auth_required,
# Import models in the correct order to ensure
proper registration with SQLAlchemy admin_required,
) def log_request_info():
[Link]("Headers: %s",
[Link])
# Add routes for admin user and program
management pages [Link]("Body: %s",
request.get_data())
@[Link]("/admin/users", methods=["GET"])
[Link]("Session: %s",
@admin_required dict(session))
def user_management_page():
return send_from_directory( @[Link]("/")
[Link](app.static_folder, "pages"), def home():
"[Link]"
if [Link]("logged_in"):
)
# Redirect to stored next_url if it exists,
otherwise to dashboard
@[Link]("/admin/programs", next_url = [Link]("next_url",
methods=["GET"]) "/dashboard")
@admin_required return redirect(next_url)
def categories_management_page(): return send_from_directory(app.static_folder,
"[Link]")
# Serve from the static folder (frontend) with
the relative path
@[Link]("/dashboard") @program_chair_required
@program_chair_required def class_record():
def dashboard(): return send_from_directory(
return send_from_directory( [Link](app.static_folder, "pages"),
"[Link]"
[Link](app.static_folder, "pages"),
"[Link]" )
)
@[Link]("/dashboard/question-bank/items")
@[Link]("/dashboard/question-bank") @program_chair_required
@program_chair_required def question_bank_items():
def question_bank(): return send_from_directory(
return send_from_directory( [Link](app.static_folder, "pages"),
"[Link]"
[Link](app.static_folder, "pages"),
"[Link]" )
)
# Replace [Link] route with a
cleaner URL pattern
@[Link]("/dashboard/item-analysis")
@[Link]("/dashboard/question")
@program_chair_required
@program_chair_required
def item_analysis():
def question_detail():
return send_from_directory(
"""Serve the question details page with query
[Link](app.static_folder, "pages"), parameters"""
"[Link]"
[Link]("Serving question detail page
) with params: %s", [Link])
return send_from_directory(
@[Link]("/dashboard/prediction-tool") [Link](app.static_folder, "pages"),
@program_chair_required "[Link]"
def prediction_tool(): )
return send_from_directory(
[Link](app.static_folder, "pages"), # Keep the old route temporarily for backwards
"[Link]" compatibility
) @[Link]("/dashboard/
[Link]")
@program_chair_required
@[Link]("/dashboard/class-record")
def question_bank_items_direct(): return send_from_directory(
# Redirect to the new URL format [Link](app.static_folder, "pages"),
"[Link]"
exam = [Link]("exam", "")
)
subject_id = [Link]("subject_id", "")
return redirect(f"/dashboard/question?
exam={exam}&subject_id={subject_id}") @[Link]("/dashboard/classes")
@program_chair_required
# Keep both old and new routes temporarily for def classes_management():
backward compatibility
return send_from_directory(
@[Link]("/question-bank")
[Link](app.static_folder, "pages"),
@program_chair_required "[Link]"
def question_bank_redirect(): )
return redirect("/dashboard/question-bank")
@[Link]("/dashboard/class-management")
@[Link]("/item-analysis") @program_chair_required
@program_chair_required def class_management_redirect():
def item_analysis_redirect(): return redirect("/dashboard/classes")
return redirect("/dashboard/item-analysis")
# Add logout route
@[Link]("/prediction-tool") @[Link]("/api/auth/logout")
@program_chair_required def logout():
def prediction_tool_redirect(): [Link]()
return redirect("/dashboard/prediction-tool") return jsonify({"status": "success",
"message": "Logged out successfully"}), 200
@[Link]("/class-record")
# Session testing route
@program_chair_required
@[Link]("/test-session")
def class_record_redirect():
def test_session():
return redirect("/dashboard/class-record")
if "visit_count" in session:
session["visit_count"] =
@[Link]("/dashboard/categories") [Link]("visit_count") + 1
@program_chair_required else:
def categories_management(): session["visit_count"] = 1
@[Link]("/frontend/assets/css/
<path:filename>")
return jsonify(
def serve_css(filename):
{
return send_from_directory(
"status": "success",
[Link](app.static_folder, "assets",
"message": f"You've visited this page "css"), filename
{session['visit_count']} time(s)",
)
"session_data": {k: str(v) for k, v in
[Link]()},
} @[Link]("/frontend/images/
<path:filename>")
)
def serve_images(filename):
return
# Special route to handle thesis-preboard-exam- send_from_directory([Link](app.static_fol
gen paths der, "images"), filename)
@[Link]("/thesis-preboard-exam-gen/
frontend/<path:filename>")
# Add specific route for print CSS
def serve_thesis_files(filename):
@[Link]("/api/assets/css/[Link]")
return send_from_directory(app.static_folder,
filename) def serve_print_css():
response = send_from_directory(
# Add specific routes for components to make [Link](app.static_folder, "assets",
debugging easier "css"), "[Link]"
@[Link]("/frontend/components/ )
<path:filename>")
# Set cache control for CSS
def serve_components(filename):
[Link]["Cache-Control"] = "public,
return max-age=86400"
send_from_directory([Link](app.static_fol
der, "components"), filename) return response
{ )
"status": "success",
"directory": upload_dir, # Ensure the directory exists
) [Link](f"Looking in directory:
{upload_dir}")
except Exception as e:
[Link](f"Absolute path: """Serve frontend images including the
{[Link](upload_dir)}") fallback image"""
try:
# Check if the file exists images_dir = [Link](app.static_folder,
"images")
file_path = [Link](upload_dir,
filename) return send_from_directory(images_dir,
filename)
if not [Link](file_path):
except Exception as e:
[Link](f"Image file not
found: {file_path}") [Link](f"Error serving frontend
image {filename}: {str(e)}")
# Return the default image-not-found
image return jsonify({"error": "Error serving
image"}), 500
return send_from_directory(
[Link](app.static_folder,
"images"), "[Link]" @[Link]("/static/<path:filepath>")
) def serve_static_files(filepath):
[Link](f"Serving static file:
{filepath}")
[Link](f"Image found, serving
from: {file_path}") # Try to first serve from the backend/static
directory
response =
send_from_directory(upload_dir, filename) backend_static =
[Link]([Link](__file__), "static")
[Link]["Cache-Control"] = "no-
cache, no-store, must-revalidate" if [Link]([Link](backend_static,
filepath)):
[Link]["Pragma"] = "no-cache"
return send_from_directory(backend_static,
[Link]["Expires"] = "0" filepath)
return response
except Exception as e: # Fallback to the project root static directory
[Link](f"Error serving image static_dir =
{filename}: {str(e)}") [Link]([Link]([Link](__
return jsonify({"error": "Error serving file__)))
image"}), 500 return send_from_directory(static_dir,
f"static/{filepath}")
[Link] = [Link](data); };
}
[Link] = async function(email)
{
try {
return apiRequest('/auth/forgot-password',
const response = await fetch(`$ 'POST', { email });
{API_BASE_URL}${endpoint}`, options);
};
const result = await [Link]();
/* Styles for question bank management */ // - Manages event listeners for dashboard
widgets
.question-bank-list { ... }
// - Handles AJAX calls to fetch dashboard data
.[Link] { ... }
// - Updates DOM with user, class, and exam <li><a href="...">Dashboard</a></li>
summaries
<li><a href="...">User Management</a></li>
<li><a href="...">Class
Management</a></li>
//======components=================// <li><a href="...">Category
Management</a></li>
//=======[Link]==========//
<li><a href="...">Exam
<!-- Admin navigation bar --> Management</a></li>
<nav class="admin-navbar"> <li><a href="...">Analytics</a></li>
<!-- Brand/logo section --> <!-- ... -->
<div class="navbar-brand">...</div> </ul>
<!-- Navigation links: Dashboard, Users, <!-- Collapsible/expandable sections for sub-
Classes, Categories, Exams, Analytics --> menus (if any) -->
<ul class="navbar-links"> </aside>
<li><a href="...">Dashboard</a></li> //=====[Link]=================//
<li><a href="...">Users</a></li> <!-- General header for non-admin pages -->
<li><a href="...">Classes</a></li> <header class="main-header">
<!-- ... --> <!-- Brand/logo -->
</ul> <div class="header-logo">...</div>
<!-- User profile dropdown/logout --> <!-- Page title or breadcrumbs -->
<div class="navbar-user"> <div class="header-title">Page Title</div>
<span>Admin Name</span> <!-- User info/profile/logout -->
<a href="...">Logout</a> <div class="header-user">
</div> <span>User Name</span>
</nav> <a href="...">Logout</a>
//=======[Link]=======// </div>
<!-- Admin sidebar menu --> </header>
<aside class="admin-sidebar"> //
<!-- User info/profile (optional) --> =========[Link]===============//
<!-- User info/profile --> # - Asserts that the pipeline completes without
errors
<div class="sidebar-profile">...</div>
# - Checks that model accuracy and output meet
<!-- Navigation links: Dashboard, Exams, expected thresholds
Question Bank, Item Analysis, etc. -->
def test_full_pipeline(): ...
<ul class="sidebar-menu">
def test_pipeline_with_real_data(): ...
<li><a href="...">Dashboard</a></li>
//========test_integration.py=========//
<li><a href="...">Exams</a></li>
# Integration tests for ML components
<li><a href="...">Question Bank</a></li>
# - Tests interaction between data preparation,
<li><a href="...">Item Analysis</a></li> model training, and prediction modules
<li><a href="...">Prediction Tool</a></li> # - Ensures models can be trained and used for
inference with real or mock data
<li><a href="...">Class Record</a></li>
def test_model_training_and_prediction(): ...
<!-- ... -->
def test_feature_importance_output(): ...
</ul>
//===========test_models.py=======//
<!-- Collapsible for mobile/tablet (if any) -->
# Unit tests for individual ML models # Adds item analysis metrics (difficulty,
discrimination, distractor strength, etc.)
# - Tests model initialization, training, and
prediction for each supported algorithm
# - Checks model serialization/deserialization def enhance_features(df):
(joblib)
# Adds advanced text features (readability,
def test_decision_tree(): ... word count, technical terms, etc.)
def test_random_forest(): ...
def test_xgboost(): ... def prepare_data(question_bank_id=None,
use_examinee_data=True,
def test_model_save_and_load(): ... use_item_analysis=True):
//=========test_prediction.py=======// # Full pipeline: get data, add empirical/item
# Tests for prediction logic and API analysis features, define target, clean
//======data_preparation.py==========// def
prepare_features_for_prediction(question_data):
# Data extraction, feature engineering, and
dataset preparation for ML # Prepares features for a single question for
prediction
//===db_model_evaluation.py=========//
def
get_data_from_db(question_bank_id=None): # Loads student data from DB, preprocesses,
and evaluates XGBoost classifier
# Fetches questions (and options, subjects)
from DB, returns DataFrame with features
# 1. Fetches students with board results,
preboard scores, subject GPAs
def enrich_with_examinee_data(df=None):
# 2. Extracts/averages scores and GPAs from
# Adds empirical difficulty, discrimination,
JSON fields
and response stats from exam results
# 3. Balances classes (upsampling)
# Handles missing data with text-based
difficulty estimation # 4. Trains XGBoost with GridSearchCV,
evaluates with cross-validation
# 5. Prints confusion matrix, accuracy, precision,
def enrich_with_item_analysis_data(df,
recall, F1
question_bank_id=None):
//====difficulty_model.py========// # Checks for missing mappings, prints sample
detailed_results
# Class-based ML model for predicting question
difficulty (with NLP features)
if __name__ == "__main__":
class DifficultyModel: diagnose_exams()
def __init__(self): //======ml_prediction.py======//
# Loads model, vectorizer, scaler, metadata # Unified interface for predicting question
if available difficulty
if __name__ == "__main__":
# Runs sample evaluation with dummy data # Trains XGBoost model for question difficulty
prediction
//======model_integration.py========//
# Integrates ML models with the web app,
logging, and question selection def
train_xgboost_model(question_bank_id=None,
use_existing_models=False, timestamp=None):
def predict_question_difficulty(question): # Prepares data, splits train/test, tunes
# Predicts difficulty for a question, logs hyperparameters with GridSearchCV
prediction # Trains, evaluates, saves model and feature
importances
def get_ml_system_status():
# Returns ML system status, accuracy, feature if __name__ == "__main__":
importance # Runs training for all or specific question
bank
# Aggregates class performance for intelligent # Trains model (XGBoost, DecisionTree, etc.)
question selection with optional tuning
# Saves model training stats for monitoring # Saves model version info for tracking
//======model_training.py==========//
def get_prediction_explanation(model, X, # Selects questions maximizing topic
feature_names): diversity
# Returns top feature importances for a
prediction
def select_questions_randomly(questions,
//=======prediction_logger.py=======// num_items):
# Logs predictions for analysis and continuous # Fallback: random selection
learning
def
class PredictionLogger: load_question_selection_model(question_bank_i
d=None):
def log_prediction(self, question_id,
prediction_result, model_version="current"): # Loads latest XGBoost model and feature
names for a question bank
# Appends prediction to daily JSONL log
//========run_full_tests.py==========//
# Test runners for ML module
def get_recent_predictions(self, limit=100):
# Loads recent predictions from logs
def run_full_tests():
# Runs all tests, including end-to-end
def generate_prediction_report(self):
# Aggregates stats (avg difficulty, counts by
subject/topic) def run_tests():
//========question_selection.py========// # Runs standard ML tests
# Intelligent question selection for exams using
ML predictions
# run_single_end_to_end.py runs just the end-
to-end test
def select_questions_intelligently(db, course_id, //========run_tests.py=========//
question_bank_id, num_items,
class_performance=None): # Script to run model training with correct
import order
# Loads model, predicts difficulty, ranks and
selects diverse questions
if __name__ == "__main__":
def demonstrate_high_accuracy():
def schedule_model_update(): # Shows why high accuracy is legitimate
# Schedules model update if needed (plots MAE vs accuracy)
# Trains new model with latest data # Tests accuracy scaling for various MAE
values
def generate_question_analytics(question_id):
def test_synthetic_model_evaluation():
# Returns analytics for a specific question
# Tests accuracy scaling with synthetic
predictions
def automatic_model_update_check(): //===test_algorithm_comparison.py=====//
# Test scripts for accuracy explanation and # Debugging utility for class and student
algorithm comparison relationships
# Runs validation for all or specific question # Assigns them to a default class or creates
bank new class
def fix_duplicate_students():
""" charset="utf8mb4"
)
import os
import sys try:
if result:
# Database connection settings (without
database name) print(f"Database '{DB_NAME}'
already exists.")
DB_USER = [Link]("DB_USER",
"root") else:
# Create database
[Link](f"CREATE import subprocess
DATABASE {DB_NAME} CHARACTER SET
utf8mb4 COLLATE utf8mb4_unicode_ci") import sys
[Link]() import os
print(f"Database '{DB_NAME}'
created successfully.") def create_lock_file():
finally: print("Creating requirements lock file...")
[Link]()
if create_database(): capture_output=True,
except [Link] as e:
print("Installing NLTK packages...")
print(f"Error running pip freeze: {e}")
import nltk
return False
except IOError as e:
for package in ["punkt", "stopwords",
print(f"Error writing lock file: {e}") "wordnet"]:
return False print(f"Downloading {package}...")
[Link](package)
return True
print("Environment setup complete!")
if __name__ == "__main__":
create_lock_file() if __name__ == "__main__":
install_requirements()
//=====setup_environment.py===//
import subprocess
import sys
def install_requirements():
print("Installing main requirements...")
subprocess.check_call(
[[Link], "-m", "pip", "install", "-r",
"[Link]"]
)