""" Database migration utilities for the Flask application. This module handles automatic database migrations on application startup. """ import os import sqlite3 from datetime import datetime from flask import current_app class DatabaseMigrator: """Handles database migrations for the application.""" def __init__(self): self.migrations_run = set() def check_table_exists(self, cursor, table_name): """Check if a table exists in the database.""" cursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name=? """, (table_name,)) return cursor.fetchone() is not None def check_column_exists(self, cursor, table_name, column_name): """Check if a column exists in a table.""" cursor.execute(f"PRAGMA table_info({table_name})") columns = [column[1] for column in cursor.fetchall()] return column_name in columns def migrate_daily_teaching_sessions(self, cursor): """Create daily_teaching_sessions table for better session tracking.""" migration_id = "daily_teaching_sessions_table" if migration_id in self.migrations_run: return True try: # Check if table already exists cursor.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name='daily_teaching_sessions' """) if cursor.fetchone(): current_app.logger.info("Daily teaching sessions table already exists") self.migrations_run.add(migration_id) return True # Create the daily_teaching_sessions table cursor.execute(""" CREATE TABLE daily_teaching_sessions ( id VARCHAR(150) PRIMARY KEY, timetable_block_id VARCHAR(150) NOT NULL, course_id VARCHAR(150) NOT NULL, tutor_id VARCHAR(150) NOT NULL, session_date DATE NOT NULL, start_time TIME NOT NULL, end_time TIME NOT NULL, room VARCHAR(100), session_type VARCHAR(50) DEFAULT 'lecture', status VARCHAR(20) DEFAULT 'scheduled', created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(150), is_verified BOOLEAN DEFAULT 0, verified_by VARCHAR(150), verified_at DATETIME, verification_method VARCHAR(50), notes TEXT, attendance_taken BOOLEAN DEFAULT 0, FOREIGN KEY (timetable_block_id) REFERENCES timetable_blocks(id), FOREIGN KEY (course_id) REFERENCES courses(id), FOREIGN KEY (tutor_id) REFERENCES users(id), FOREIGN KEY (created_by) REFERENCES supervisors(id), FOREIGN KEY (verified_by) REFERENCES supervisors(id) ) """) # Create indexes cursor.execute("CREATE INDEX idx_daily_sessions_date ON daily_teaching_sessions(session_date)") cursor.execute("CREATE INDEX idx_daily_sessions_tutor_date ON daily_teaching_sessions(tutor_id, session_date)") cursor.execute("CREATE INDEX idx_daily_sessions_course_date ON daily_teaching_sessions(course_id, session_date)") cursor.execute("CREATE INDEX idx_daily_sessions_status ON daily_teaching_sessions(status)") cursor.execute("CREATE INDEX idx_daily_sessions_block_date ON daily_teaching_sessions(timetable_block_id, session_date)") # Create unique constraint cursor.execute(""" CREATE UNIQUE INDEX unique_block_date ON daily_teaching_sessions(timetable_block_id, session_date) """) current_app.logger.info(" Daily teaching sessions table created successfully") # Mark migration as completed self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error creating daily teaching sessions table: {e}") return False def migrate_enrollment_model_restructure(self, cursor): """Restructure enrollment model to support many-to-many course relationships.""" migration_id = "enrollment_model_restructure" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting enrollment model restructure migration...") # Step 1: Create enrollment_courses association table if not self.check_table_exists(cursor, 'enrollment_courses'): cursor.execute(""" CREATE TABLE enrollment_courses ( enrollment_id TEXT NOT NULL, course_id TEXT NOT NULL, enrollment_date DATE DEFAULT CURRENT_DATE, status TEXT DEFAULT 'active', grade TEXT, attendance_percentage REAL, midterm_grade TEXT, final_exam_grade TEXT, project_grade TEXT, is_audit INTEGER DEFAULT 0, completion_date DATE, withdrawal_date DATE, withdrawal_reason TEXT, PRIMARY KEY (enrollment_id, course_id), FOREIGN KEY (enrollment_id) REFERENCES enrollments(id) ON DELETE CASCADE, FOREIGN KEY (course_id) REFERENCES courses(id) ON DELETE CASCADE ) """) current_app.logger.info(" Created enrollment_courses association table") else: current_app.logger.info("✓ enrollment_courses table already exists") # Step 2: Add term column to enrollments table if not self.check_column_exists(cursor, 'enrollments', 'term'): cursor.execute("ALTER TABLE enrollments ADD COLUMN term TEXT") current_app.logger.info(" Added term column to enrollments table") else: current_app.logger.info("✓ term column already exists in enrollments table") # Step 3: Migrate existing enrollment data if self.check_table_exists(cursor, 'enrollments'): # Check if there are enrollments with course_id that need migration cursor.execute("PRAGMA table_info(enrollments)") columns = [column[1] for column in cursor.fetchall()] if 'course_id' in columns: # Get existing enrollments with course_id cursor.execute(""" SELECT id, student_id, course_id, speciality_id, enrollment_date, status, completion_date, grade, attendance_percentage, midterm_grade, final_exam_grade, project_grade, comments, is_audit, enrollment_type, withdrawal_date, withdrawal_reason FROM enrollments WHERE course_id IS NOT NULL """) existing_enrollments = cursor.fetchall() current_app.logger.info(f"Found {len(existing_enrollments)} existing enrollments to migrate") # Migrate each enrollment to the association table migrated_count = 0 for enrollment in existing_enrollments: enrollment_id, student_id, course_id, speciality_id, enrollment_date, status, completion_date, grade, attendance_percentage, midterm_grade, final_exam_grade, project_grade, comments, is_audit, enrollment_type, withdrawal_date, withdrawal_reason = enrollment # Insert into enrollment_courses association table cursor.execute(""" INSERT OR IGNORE INTO enrollment_courses ( enrollment_id, course_id, enrollment_date, status, grade, attendance_percentage, midterm_grade, final_exam_grade, project_grade, is_audit, completion_date, withdrawal_date, withdrawal_reason ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (enrollment_id, course_id, enrollment_date, status, grade, attendance_percentage, midterm_grade, final_exam_grade, project_grade, is_audit, completion_date, withdrawal_date, withdrawal_reason)) if cursor.rowcount > 0: migrated_count += 1 current_app.logger.info(f"✅ Migrated {migrated_count} enrollments to enrollment_courses table") else: current_app.logger.info("✓ No course_id column found in enrollments table") # Step 4: Update enrollments without speciality_id if self.check_table_exists(cursor, 'enrollments') and self.check_table_exists(cursor, 'students'): cursor.execute(""" UPDATE enrollments SET speciality_id = ( SELECT s.speciality_id FROM students s WHERE s.id = enrollments.student_id ) WHERE speciality_id IS NULL """) updated_count = cursor.rowcount if updated_count > 0: current_app.logger.info(f"✅ Updated {updated_count} enrollments with speciality information") # Step 5: Create indexes for the new association table indexes = [ "CREATE INDEX IF NOT EXISTS idx_enrollment_courses_enrollment_id ON enrollment_courses(enrollment_id)", "CREATE INDEX IF NOT EXISTS idx_enrollment_courses_course_id ON enrollment_courses(course_id)", "CREATE INDEX IF NOT EXISTS idx_enrollment_courses_status ON enrollment_courses(status)", "CREATE INDEX IF NOT EXISTS idx_enrollments_term ON enrollments(term)" ] for index_sql in indexes: cursor.execute(index_sql) current_app.logger.info(" Created indexes for enrollment_courses table") # Mark migration as completed self.migrations_run.add(migration_id) current_app.logger.info("Enrollment model restructure migration completed successfully!") return True except Exception as e: current_app.logger.error(f"Error in enrollment model restructure migration: {e}") return False def migrate_course_structure_models(self, cursor): """Create comprehensive course structure models for detailed progress tracking.""" migration_id = "course_structure_models" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting course structure models migration...") # Create course_levels table if not self.check_table_exists(cursor, 'course_levels'): cursor.execute(""" CREATE TABLE course_levels ( id TEXT PRIMARY KEY, course_id TEXT NOT NULL, level_name TEXT NOT NULL, level_order INTEGER NOT NULL DEFAULT 1, description TEXT, prerequisites TEXT, learning_objectives TEXT, estimated_duration_weeks INTEGER DEFAULT 1, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (course_id) REFERENCES courses(id) ON DELETE CASCADE, UNIQUE(course_id, level_order) ) """) current_app.logger.info(" Created course_levels table") # Create course_weeks table if not self.check_table_exists(cursor, 'course_weeks'): cursor.execute(""" CREATE TABLE course_weeks ( id TEXT PRIMARY KEY, course_level_id TEXT NOT NULL, week_number INTEGER NOT NULL, week_title TEXT NOT NULL, description TEXT, learning_outcomes TEXT, estimated_hours REAL DEFAULT 0.0, start_date DATE, end_date DATE, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (course_level_id) REFERENCES course_levels(id) ON DELETE CASCADE, UNIQUE(course_level_id, week_number) ) """) current_app.logger.info(" Created course_weeks table") # Add new columns to existing course_modules table if self.check_table_exists(cursor, 'course_modules'): # Add course_week_id column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'course_week_id'): cursor.execute("ALTER TABLE course_modules ADD COLUMN course_week_id TEXT") current_app.logger.info(" Added course_week_id column to course_modules table") # Add module_type column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'module_type'): cursor.execute("ALTER TABLE course_modules ADD COLUMN module_type TEXT DEFAULT 'lecture'") current_app.logger.info(" Added module_type column to course_modules table") # Add estimated_duration_minutes column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'estimated_duration_minutes'): cursor.execute("ALTER TABLE course_modules ADD COLUMN estimated_duration_minutes INTEGER DEFAULT 60") current_app.logger.info(" Added estimated_duration_minutes column to course_modules table") # Add learning_objectives column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'learning_objectives'): cursor.execute("ALTER TABLE course_modules ADD COLUMN learning_objectives TEXT") current_app.logger.info(" Added learning_objectives column to course_modules table") # Add is_required column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'is_required'): cursor.execute("ALTER TABLE course_modules ADD COLUMN is_required INTEGER DEFAULT 1") current_app.logger.info(" Added is_required column to course_modules table") # Add is_active column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'is_active'): cursor.execute("ALTER TABLE course_modules ADD COLUMN is_active INTEGER DEFAULT 1") current_app.logger.info(" Added is_active column to course_modules table") # Add updated_at column if it doesn't exist if not self.check_column_exists(cursor, 'course_modules', 'updated_at'): cursor.execute("ALTER TABLE course_modules ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP") current_app.logger.info(" Added updated_at column to course_modules table") current_app.logger.info(" Enhanced existing course_modules table with new columns") else: # Create course_modules table if it doesn't exist cursor.execute(""" CREATE TABLE course_modules ( id TEXT PRIMARY KEY, course_id TEXT, course_week_id TEXT, title TEXT NOT NULL, description TEXT, sequence INTEGER, module_type TEXT DEFAULT 'lecture', estimated_duration_minutes INTEGER DEFAULT 60, learning_objectives TEXT, is_required INTEGER DEFAULT 1, start_date DATE, end_date DATE, is_published INTEGER DEFAULT 0, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, objectives TEXT, estimated_study_time INTEGER, is_assessment_module INTEGER DEFAULT 0, FOREIGN KEY (course_id) REFERENCES courses(id) ON DELETE CASCADE, FOREIGN KEY (course_week_id) REFERENCES course_weeks(id) ON DELETE CASCADE ) """) current_app.logger.info(" Created course_modules table") # Create module_assignments table if not self.check_table_exists(cursor, 'module_assignments'): cursor.execute(""" CREATE TABLE module_assignments ( id TEXT PRIMARY KEY, module_id TEXT NOT NULL, assignment_title TEXT NOT NULL, assignment_description TEXT, assignment_type TEXT DEFAULT 'homework', instructions TEXT, max_points REAL DEFAULT 100.0, due_date DATETIME, submission_format TEXT, is_group_assignment INTEGER DEFAULT 0, max_group_size INTEGER DEFAULT 1, late_submission_allowed INTEGER DEFAULT 1, late_penalty_percentage REAL DEFAULT 10.0, is_published INTEGER DEFAULT 0, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (module_id) REFERENCES course_modules(id) ON DELETE CASCADE ) """) current_app.logger.info(" Created module_assignments table") # Create reading_materials table if not self.check_table_exists(cursor, 'reading_materials'): cursor.execute(""" CREATE TABLE reading_materials ( id TEXT PRIMARY KEY, module_id TEXT NOT NULL, title TEXT NOT NULL, material_type TEXT DEFAULT 'text', description TEXT, content_url TEXT, file_path TEXT, file_size INTEGER, reading_time_minutes INTEGER, is_required INTEGER DEFAULT 1, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (module_id) REFERENCES course_modules(id) ON DELETE CASCADE ) """) current_app.logger.info(" Created reading_materials table") # Create module_quizzes table if not self.check_table_exists(cursor, 'module_quizzes'): cursor.execute(""" CREATE TABLE module_quizzes ( id TEXT PRIMARY KEY, module_id TEXT, quiz_title TEXT NOT NULL, quiz_description TEXT, quiz_type TEXT DEFAULT 'multiple_choice', max_attempts INTEGER DEFAULT 3, time_limit_minutes INTEGER, max_points REAL DEFAULT 100.0, passing_score REAL DEFAULT 60.0, is_randomized INTEGER DEFAULT 0, show_correct_answers INTEGER DEFAULT 1, show_answers_after TEXT DEFAULT 'immediate', due_date DATETIME, is_published INTEGER DEFAULT 0, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(150), FOREIGN KEY (module_id) REFERENCES course_modules(id) ON DELETE CASCADE, FOREIGN KEY (created_by) REFERENCES users(id) ) """) current_app.logger.info(" Created module_quizzes table") # Create quiz_questions table if not self.check_table_exists(cursor, 'quiz_questions'): cursor.execute(""" CREATE TABLE quiz_questions ( id TEXT PRIMARY KEY, quiz_id TEXT NOT NULL, question_text TEXT NOT NULL, question_type TEXT DEFAULT 'multiple_choice', points REAL DEFAULT 1.0, question_order INTEGER NOT NULL DEFAULT 1, explanation TEXT, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (quiz_id) REFERENCES module_quizzes(id) ON DELETE CASCADE, UNIQUE(quiz_id, question_order) ) """) current_app.logger.info(" Created quiz_questions table") # Create quiz_options table if not self.check_table_exists(cursor, 'quiz_options'): cursor.execute(""" CREATE TABLE quiz_options ( id TEXT PRIMARY KEY, question_id TEXT NOT NULL, option_text TEXT NOT NULL, is_correct INTEGER DEFAULT 0, option_order INTEGER NOT NULL DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (question_id) REFERENCES quiz_questions(id) ON DELETE CASCADE, UNIQUE(question_id, option_order) ) """) current_app.logger.info(" Created quiz_options table") # Create module_videos table if not self.check_table_exists(cursor, 'module_videos'): cursor.execute(""" CREATE TABLE module_videos ( id TEXT PRIMARY KEY, module_id TEXT NOT NULL, video_title TEXT NOT NULL, video_description TEXT, video_url TEXT, video_file_path TEXT, video_duration_seconds INTEGER, thumbnail_url TEXT, video_quality TEXT DEFAULT 'HD', is_required INTEGER DEFAULT 1, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (module_id) REFERENCES course_modules(id) ON DELETE CASCADE ) """) current_app.logger.info(" Created module_videos table") # Create module_practices table if not self.check_table_exists(cursor, 'module_practices'): cursor.execute(""" CREATE TABLE module_practices ( id TEXT PRIMARY KEY, module_id TEXT NOT NULL, practice_title TEXT NOT NULL, practice_description TEXT, practice_type TEXT DEFAULT 'exercise', instructions TEXT, estimated_time_minutes INTEGER DEFAULT 30, max_attempts INTEGER DEFAULT 5, is_graded INTEGER DEFAULT 0, max_points REAL DEFAULT 0.0, is_required INTEGER DEFAULT 1, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (module_id) REFERENCES course_modules(id) ON DELETE CASCADE ) """) current_app.logger.info(" Created module_practices table") # Create quiz_attempts table if not self.check_table_exists(cursor, 'quiz_attempts'): cursor.execute(""" CREATE TABLE quiz_attempts ( id TEXT PRIMARY KEY, quiz_id TEXT NOT NULL, student_id TEXT NOT NULL, attempt_number INTEGER NOT NULL DEFAULT 1, started_at DATETIME DEFAULT CURRENT_TIMESTAMP, completed_at DATETIME, time_spent_minutes INTEGER DEFAULT 0, total_score REAL DEFAULT 0.0, max_possible_score REAL DEFAULT 0.0, percentage_score REAL DEFAULT 0.0, is_passed INTEGER DEFAULT 0, is_completed INTEGER DEFAULT 0, is_submitted INTEGER DEFAULT 0, is_graded INTEGER DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (quiz_id) REFERENCES module_quizzes(id) ON DELETE CASCADE, FOREIGN KEY (student_id) REFERENCES users(id) ON DELETE CASCADE, UNIQUE(quiz_id, student_id, attempt_number) ) """) current_app.logger.info(" Created quiz_attempts table") else: # Add missing columns to existing table if not self.check_column_exists(cursor, 'quiz_attempts', 'is_completed'): cursor.execute("ALTER TABLE quiz_attempts ADD COLUMN is_completed INTEGER DEFAULT 0") current_app.logger.info(" Added is_completed column to quiz_attempts table") if not self.check_column_exists(cursor, 'quiz_attempts', 'is_submitted'): cursor.execute("ALTER TABLE quiz_attempts ADD COLUMN is_submitted INTEGER DEFAULT 0") current_app.logger.info(" Added is_submitted column to quiz_attempts table") if not self.check_column_exists(cursor, 'quiz_attempts', 'is_graded'): cursor.execute("ALTER TABLE quiz_attempts ADD COLUMN is_graded INTEGER DEFAULT 0") current_app.logger.info(" Added is_graded column to quiz_attempts table") # Create quiz_answers table if not self.check_table_exists(cursor, 'quiz_answers'): cursor.execute(""" CREATE TABLE quiz_answers ( id TEXT PRIMARY KEY, attempt_id TEXT NOT NULL, question_id TEXT NOT NULL, selected_option_id TEXT, answer_text TEXT, is_correct INTEGER DEFAULT 0, points_earned REAL DEFAULT 0.0, answered_at DATETIME DEFAULT CURRENT_TIMESTAMP, graded_by VARCHAR(150), graded_at DATETIME, grading_notes TEXT, FOREIGN KEY (attempt_id) REFERENCES quiz_attempts(id) ON DELETE CASCADE, FOREIGN KEY (question_id) REFERENCES quiz_questions(id) ON DELETE CASCADE, FOREIGN KEY (selected_option_id) REFERENCES quiz_options(id) ON DELETE CASCADE, FOREIGN KEY (graded_by) REFERENCES users(id) ) """) current_app.logger.info(" Created quiz_answers table") # Create video_progress table if not self.check_table_exists(cursor, 'video_progress'): cursor.execute(""" CREATE TABLE video_progress ( id TEXT PRIMARY KEY, video_id TEXT NOT NULL, student_id TEXT NOT NULL, progress_percentage REAL DEFAULT 0.0, time_watched_seconds INTEGER DEFAULT 0, last_position_seconds INTEGER DEFAULT 0, is_completed INTEGER DEFAULT 0, completed_at DATETIME, last_watched_at DATETIME DEFAULT CURRENT_TIMESTAMP, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (video_id) REFERENCES module_videos(id) ON DELETE CASCADE, FOREIGN KEY (student_id) REFERENCES students(id) ON DELETE CASCADE, UNIQUE(video_id, student_id) ) """) current_app.logger.info(" Created video_progress table") # Create practice_attempts table if not self.check_table_exists(cursor, 'practice_attempts'): cursor.execute(""" CREATE TABLE practice_attempts ( id TEXT PRIMARY KEY, practice_id TEXT NOT NULL, student_id TEXT NOT NULL, attempt_number INTEGER NOT NULL DEFAULT 1, started_at DATETIME DEFAULT CURRENT_TIMESTAMP, completed_at DATETIME, time_spent_minutes INTEGER, score REAL DEFAULT 0.0, max_possible_score REAL DEFAULT 0.0, is_completed INTEGER DEFAULT 0, feedback TEXT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (practice_id) REFERENCES module_practices(id) ON DELETE CASCADE, FOREIGN KEY (student_id) REFERENCES students(id) ON DELETE CASCADE, UNIQUE(practice_id, student_id, attempt_number) ) """) current_app.logger.info(" Created practice_attempts table") # Create module_progress table if not self.check_table_exists(cursor, 'module_progress'): cursor.execute(""" CREATE TABLE module_progress ( id TEXT PRIMARY KEY, module_id TEXT NOT NULL, student_id TEXT NOT NULL, enrollment_id TEXT NOT NULL, completion_percentage REAL DEFAULT 0.0, is_started INTEGER DEFAULT 0, is_completed INTEGER DEFAULT 0, started_at DATETIME, completed_at DATETIME, last_accessed_at DATETIME DEFAULT CURRENT_TIMESTAMP, time_spent_minutes INTEGER DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (module_id) REFERENCES course_modules(id) ON DELETE CASCADE, FOREIGN KEY (student_id) REFERENCES students(id) ON DELETE CASCADE, FOREIGN KEY (enrollment_id) REFERENCES enrollments(id) ON DELETE CASCADE, UNIQUE(module_id, student_id, enrollment_id) ) """) current_app.logger.info(" Created module_progress table") # Add module_assignment_id column to assignment_submissions table if not self.check_column_exists(cursor, 'assignment_submissions', 'module_assignment_id'): cursor.execute("ALTER TABLE assignment_submissions ADD COLUMN module_assignment_id TEXT") current_app.logger.info(" Added module_assignment_id column to assignment_submissions table") # Create indexes for better performance indexes = [ "CREATE INDEX IF NOT EXISTS idx_course_levels_course_id ON course_levels(course_id)", "CREATE INDEX IF NOT EXISTS idx_course_levels_level_order ON course_levels(level_order)", "CREATE INDEX IF NOT EXISTS idx_course_weeks_level_id ON course_weeks(course_level_id)", "CREATE INDEX IF NOT EXISTS idx_course_weeks_week_number ON course_weeks(week_number)", "CREATE INDEX IF NOT EXISTS idx_course_modules_week_id ON course_modules(course_week_id)", "CREATE INDEX IF NOT EXISTS idx_module_assignments_module_id ON module_assignments(module_id)", "CREATE INDEX IF NOT EXISTS idx_reading_materials_module_id ON reading_materials(module_id)", "CREATE INDEX IF NOT EXISTS idx_module_quizzes_module_id ON module_quizzes(module_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_questions_quiz_id ON quiz_questions(quiz_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_options_question_id ON quiz_options(question_id)", "CREATE INDEX IF NOT EXISTS idx_module_videos_module_id ON module_videos(module_id)", "CREATE INDEX IF NOT EXISTS idx_module_practices_module_id ON module_practices(module_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_attempts_quiz_id ON quiz_attempts(quiz_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_attempts_student_id ON quiz_attempts(student_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_answers_attempt_id ON quiz_answers(attempt_id)", "CREATE INDEX IF NOT EXISTS idx_video_progress_video_id ON video_progress(video_id)", "CREATE INDEX IF NOT EXISTS idx_video_progress_student_id ON video_progress(student_id)", "CREATE INDEX IF NOT EXISTS idx_practice_attempts_practice_id ON practice_attempts(practice_id)", "CREATE INDEX IF NOT EXISTS idx_practice_attempts_student_id ON practice_attempts(student_id)", "CREATE INDEX IF NOT EXISTS idx_module_progress_module_id ON module_progress(module_id)", "CREATE INDEX IF NOT EXISTS idx_module_progress_student_id ON module_progress(student_id)", "CREATE INDEX IF NOT EXISTS idx_module_progress_enrollment_id ON module_progress(enrollment_id)" ] for index_sql in indexes: cursor.execute(index_sql) current_app.logger.info(" Created indexes for course structure tables") # Mark migration as completed self.migrations_run.add(migration_id) current_app.logger.info("Course structure models migration completed successfully!") return True except Exception as e: current_app.logger.error(f"Error in course structure models migration: {e}") return False def migrate_tutor_departments_unique_constraint_fix(self, cursor): """Fix the UNIQUE constraint issue in tutor_departments table.""" migration_id = "tutor_departments_unique_constraint_fix" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting tutor departments unique constraint fix migration...") # Check if tutor_departments table exists if not self.check_table_exists(cursor, 'tutor_departments'): current_app.logger.info("tutor_departments table not found, skipping migration") self.migrations_run.add(migration_id) return True # Step 1: Check for duplicate IDs cursor.execute(""" SELECT id, COUNT(*) as count FROM tutor_departments GROUP BY id HAVING COUNT(*) > 1 """) duplicates = cursor.fetchall() if duplicates: current_app.logger.info(f"Found {len(duplicates)} duplicate IDs in tutor_departments table") # Step 2: Remove duplicates, keeping only the first occurrence for dup_id, count in duplicates: current_app.logger.info(f"Removing {count-1} duplicate(s) for ID: {dup_id}") # Keep the first occurrence (lowest rowid) and delete the rest cursor.execute(""" DELETE FROM tutor_departments WHERE id = ? AND rowid NOT IN ( SELECT MIN(rowid) FROM tutor_departments WHERE id = ? ) """, (dup_id, dup_id)) current_app.logger.info(" Removed duplicate IDs from tutor_departments table") else: current_app.logger.info("✓ No duplicate IDs found in tutor_departments table") # Step 3: Ensure all records have valid UUIDs (regenerate any that might be invalid) cursor.execute(""" SELECT COUNT(*) FROM tutor_departments WHERE id IS NULL OR id = '' OR LENGTH(id) != 36 """) result = cursor.fetchone() invalid_uuids = result[0] if result else 0 if invalid_uuids > 0: current_app.logger.info(f"Found {invalid_uuids} records with invalid UUIDs, regenerating...") # Regenerate UUIDs for invalid records cursor.execute(""" UPDATE tutor_departments SET id = LOWER(HEX(RANDOMBLOB(4))) || '-' || LOWER(HEX(RANDOMBLOB(2))) || '-' || LOWER(HEX(RANDOMBLOB(2))) || '-' || LOWER(HEX(RANDOMBLOB(2))) || '-' || LOWER(HEX(RANDOMBLOB(6))) WHERE id IS NULL OR id = '' OR LENGTH(id) != 36 """) current_app.logger.info(f"✅ Regenerated {cursor.rowcount} invalid UUIDs") else: current_app.logger.info("✓ All UUIDs in tutor_departments table are valid") # Step 4: Ensure indexes exist indexes = [ "CREATE INDEX IF NOT EXISTS idx_tutor_departments_tutor_id ON tutor_departments(tutor_id)", "CREATE INDEX IF NOT EXISTS idx_tutor_departments_department_name ON tutor_departments(department_name)", "CREATE INDEX IF NOT EXISTS idx_tutor_departments_is_primary ON tutor_departments(is_primary)", "CREATE INDEX IF NOT EXISTS idx_tutor_departments_active ON tutor_departments(is_active)", "CREATE INDEX IF NOT EXISTS idx_tutor_departments_assigned_by ON tutor_departments(assigned_by)" ] for index_sql in indexes: cursor.execute(index_sql) current_app.logger.info(" Ensured indexes exist for tutor_departments table") # Step 5: Verify the fix cursor.execute(""" SELECT COUNT(*) FROM tutor_departments """) result = cursor.fetchone() total_records = result[0] if result else 0 current_app.logger.info(f"✅ Migration completed. Total records in tutor_departments: {total_records}") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error in tutor departments unique constraint fix migration: {e}") return False def migrate_daily_teaching_sessions_teaching_session_relation(self, cursor): """Add teaching_session_id column to daily_teaching_sessions and establish attendance relationship.""" migration_id = "daily_teaching_sessions_teaching_session_relation" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting daily teaching sessions teaching session relation migration...") # Check if daily_teaching_sessions table exists if not self.check_table_exists(cursor, 'daily_teaching_sessions'): current_app.logger.info("daily_teaching_sessions table not found, skipping migration") self.migrations_run.add(migration_id) return True # Step 1: Add teaching_session_id column if it doesn't exist if not self.check_column_exists(cursor, 'daily_teaching_sessions', 'teaching_session_id'): current_app.logger.info("Adding teaching_session_id column to daily_teaching_sessions table...") cursor.execute(""" ALTER TABLE daily_teaching_sessions ADD COLUMN teaching_session_id VARCHAR(150) """) current_app.logger.info(" Added teaching_session_id column to daily_teaching_sessions table") else: current_app.logger.info("✓ teaching_session_id column already exists in daily_teaching_sessions table") # Step 2: Add foreign key constraint if it doesn't exist try: cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_daily_teaching_sessions_teaching_session_id ON daily_teaching_sessions(teaching_session_id) """) current_app.logger.info(" Added index for teaching_session_id") except Exception as e: current_app.logger.warning(f"Could not create index for teaching_session_id: {e}") # Step 3: Update existing daily_teaching_sessions to link with teaching_sessions current_app.logger.info("Linking existing daily_teaching_sessions with teaching_sessions...") # Find daily sessions that don't have a linked teaching session cursor.execute(""" SELECT dts.id, dts.course_id, dts.tutor_id, dts.session_date, dts.start_time, dts.end_time, dts.room, dts.session_type FROM daily_teaching_sessions dts WHERE dts.teaching_session_id IS NULL """) unlinked_sessions = cursor.fetchall() linked_count = 0 for session_data in unlinked_sessions: session_id, course_id, tutor_id, session_date, start_time, end_time, room, session_type = session_data # Try to find a matching teaching session cursor.execute(""" SELECT id FROM teaching_session WHERE course_id = ? AND tutor_id = ? AND room = ? AND session_type = ? AND start_time = ? AND end_time = ? LIMIT 1 """, (course_id, tutor_id, room, session_type, start_time, end_time)) matching_session = cursor.fetchone() if matching_session: # Link the daily session to the teaching session cursor.execute(""" UPDATE daily_teaching_sessions SET teaching_session_id = ? WHERE id = ? """, (matching_session[0], session_id)) linked_count += 1 current_app.logger.info(f"✅ Linked {linked_count} existing daily_teaching_sessions with teaching_sessions") # Step 4: Add daily_session_id column to attendance table if it doesn't exist if not self.check_column_exists(cursor, 'attendance', 'daily_session_id'): current_app.logger.info("Adding daily_session_id column to attendance table...") cursor.execute(""" ALTER TABLE attendance ADD COLUMN daily_session_id VARCHAR(150) """) current_app.logger.info(" Added daily_session_id column to attendance table") else: current_app.logger.info("✓ daily_session_id column already exists in attendance table") # Step 5: Add attendance relationship indexes indexes = [ "CREATE INDEX IF NOT EXISTS idx_attendance_daily_session_id ON attendance(daily_session_id)", "CREATE INDEX IF NOT EXISTS idx_daily_teaching_sessions_status ON daily_teaching_sessions(status)", "CREATE INDEX IF NOT EXISTS idx_daily_teaching_sessions_session_date ON daily_teaching_sessions(session_date)", "CREATE INDEX IF NOT EXISTS idx_daily_teaching_sessions_tutor_date ON daily_teaching_sessions(tutor_id, session_date)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Added attendance relationship indexes") # Step 6: Verify the migration cursor.execute("SELECT COUNT(*) FROM daily_teaching_sessions WHERE teaching_session_id IS NOT NULL") result = cursor.fetchone() linked_sessions = result[0] if result else 0 cursor.execute("SELECT COUNT(*) FROM daily_teaching_sessions") result = cursor.fetchone() total_sessions = result[0] if result else 0 current_app.logger.info(f"✅ Migration completed. {linked_sessions}/{total_sessions} daily sessions linked to teaching sessions") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error in daily teaching sessions teaching session relation migration: {e}") return False def migrate_courses_total_hours_constraint_fix(self, cursor): """Fix the courses table total_hours constraint issue.""" migration_id = "courses_total_hours_constraint_fix" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting courses total_hours constraint fix migration...") # Check if courses table exists if not self.check_table_exists(cursor, 'courses'): current_app.logger.info("courses table not found, skipping migration") self.migrations_run.add(migration_id) return True # Check if the constraint exists by looking at the table schema cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='courses'") table_sql = cursor.fetchone() if table_sql and 'positive_total_hours' in table_sql[0]: current_app.logger.info("Found positive_total_hours constraint, removing it...") # Create backup cursor.execute("CREATE TABLE courses_backup AS SELECT * FROM courses") current_app.logger.info(" Created courses table backup") # Drop original table cursor.execute("DROP TABLE courses") current_app.logger.info(" Dropped original courses table") # Recreate table without the constraint cursor.execute(""" CREATE TABLE courses ( id VARCHAR(150) PRIMARY KEY, code VARCHAR(20) UNIQUE NOT NULL, title VARCHAR(200) NOT NULL, description TEXT, credits INTEGER, department VARCHAR(100), is_active BOOLEAN DEFAULT 1, max_students INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, supervisor_id VARCHAR(150), semester VARCHAR(20), learning_outcomes TEXT, required_materials TEXT, assessment_method TEXT, syllabus TEXT, total_hours INTEGER, is_archived BOOLEAN DEFAULT 0, archive_date DATETIME, course_level VARCHAR(20), course_module VARCHAR(20), language VARCHAR(50) DEFAULT 'English', certification_available BOOLEAN DEFAULT 0, has_practical BOOLEAN DEFAULT 0, speciality_id VARCHAR(150), academic_session_id VARCHAR(150), is_shared_course BOOLEAN DEFAULT 0, shared_course_type VARCHAR(50) DEFAULT 'department_specific', sharing_level VARCHAR(50) DEFAULT 'single', FOREIGN KEY (supervisor_id) REFERENCES supervisors(id), FOREIGN KEY (speciality_id) REFERENCES specialities(id), FOREIGN KEY (academic_session_id) REFERENCES academic_sessions(id), CHECK (credits > 0), CHECK (max_students > 0) ) """) current_app.logger.info(" Recreated courses table without positive_total_hours constraint") # Restore data cursor.execute("INSERT INTO courses SELECT * FROM courses_backup") restored_count = cursor.rowcount current_app.logger.info(f"✅ Restored {restored_count} courses from backup") # Drop backup cursor.execute("DROP TABLE courses_backup") current_app.logger.info(" Dropped backup table") # Recreate indexes indexes = [ "CREATE INDEX IF NOT EXISTS idx_courses_code ON courses(code)", "CREATE INDEX IF NOT EXISTS idx_courses_department ON courses(department)", "CREATE INDEX IF NOT EXISTS idx_courses_supervisor_id ON courses(supervisor_id)", "CREATE INDEX IF NOT EXISTS idx_courses_speciality_id ON courses(speciality_id)", "CREATE INDEX IF NOT EXISTS idx_courses_academic_session_id ON courses(academic_session_id)", "CREATE INDEX IF NOT EXISTS idx_courses_is_active ON courses(is_active)", "CREATE INDEX IF NOT EXISTS idx_courses_is_shared_course ON courses(is_shared_course)", "CREATE INDEX IF NOT EXISTS idx_courses_shared_course_type ON courses(shared_course_type)", "CREATE INDEX IF NOT EXISTS idx_courses_sharing_level ON courses(sharing_level)" ] for index_sql in indexes: cursor.execute(index_sql) current_app.logger.info(" Recreated indexes for courses table") # Verify the fix cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='courses'") new_table_sql = cursor.fetchone() if new_table_sql and 'positive_total_hours' not in new_table_sql[0]: current_app.logger.info(" Verification successful: positive_total_hours constraint removed") else: current_app.logger.warning("Warning: positive_total_hours constraint may still exist") else: current_app.logger.info("✓ positive_total_hours constraint not found, no action needed") # Mark migration as completed self.migrations_run.add(migration_id) current_app.logger.info("Courses total_hours constraint fix migration completed successfully!") return True except Exception as e: current_app.logger.error(f"Error in courses total_hours constraint fix migration: {e}") return False def migrate_exam_evidence_system(self, cursor): """Create exam evidence upload system with security and tracking.""" migration_id = "exam_evidence_system" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting exam evidence system migration...") # Create exam_evidence table if not self.check_table_exists(cursor, 'exam_evidence'): cursor.execute(""" CREATE TABLE exam_evidence ( id VARCHAR(150) PRIMARY KEY, student_id VARCHAR(150) NOT NULL, course_id VARCHAR(150), speciality_id VARCHAR(150), academic_session_id VARCHAR(150), evidence_type VARCHAR(50) NOT NULL, title VARCHAR(200) NOT NULL, description TEXT, examination_body VARCHAR(200), exam_date DATE, result_date DATE, grade_achieved VARCHAR(10), percentage_score REAL, max_score REAL, file_name VARCHAR(255) NOT NULL, file_path VARCHAR(500) NOT NULL, file_size INTEGER NOT NULL, file_type VARCHAR(50) NOT NULL, file_hash VARCHAR(64) NOT NULL, mime_type VARCHAR(100) NOT NULL, uploaded_by VARCHAR(150) NOT NULL, uploaded_at DATETIME DEFAULT CURRENT_TIMESTAMP, upload_ip_address VARCHAR(45), upload_user_agent VARCHAR(500), status VARCHAR(20) DEFAULT 'pending', verified_by VARCHAR(150), verified_at DATETIME, verification_notes TEXT, rejection_reason TEXT, access_level VARCHAR(50) DEFAULT 'tutor_supervisor', is_encrypted INTEGER DEFAULT 1, encryption_key_id VARCHAR(150), retention_period_days INTEGER DEFAULT 2555, expires_at DATETIME, is_active INTEGER DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (student_id) REFERENCES students(id), FOREIGN KEY (course_id) REFERENCES courses(id), FOREIGN KEY (speciality_id) REFERENCES specialities(id), FOREIGN KEY (academic_session_id) REFERENCES academic_sessions(id), FOREIGN KEY (uploaded_by) REFERENCES users(id), FOREIGN KEY (verified_by) REFERENCES users(id), CHECK (file_size > 0), CHECK (percentage_score >= 0 AND percentage_score <= 100), CHECK (max_score > 0), CHECK (retention_period_days > 0) ) """) current_app.logger.info(" Created exam_evidence table") # Create exam_evidence_versions table if not self.check_table_exists(cursor, 'exam_evidence_versions'): cursor.execute(""" CREATE TABLE exam_evidence_versions ( id VARCHAR(150) PRIMARY KEY, exam_evidence_id VARCHAR(150) NOT NULL, version_number INTEGER NOT NULL DEFAULT 1, file_name VARCHAR(255) NOT NULL, file_path VARCHAR(500) NOT NULL, file_size INTEGER NOT NULL, file_type VARCHAR(50) NOT NULL, file_hash VARCHAR(64) NOT NULL, mime_type VARCHAR(100) NOT NULL, uploaded_by VARCHAR(150) NOT NULL, uploaded_at DATETIME DEFAULT CURRENT_TIMESTAMP, upload_reason VARCHAR(200), changes_description TEXT, is_encrypted INTEGER DEFAULT 1, encryption_key_id VARCHAR(150), FOREIGN KEY (exam_evidence_id) REFERENCES exam_evidence(id) ON DELETE CASCADE, FOREIGN KEY (uploaded_by) REFERENCES users(id), CHECK (version_number > 0), CHECK (file_size > 0), UNIQUE(exam_evidence_id, version_number) ) """) current_app.logger.info(" Created exam_evidence_versions table") # Create exam_evidence_access_logs table if not self.check_table_exists(cursor, 'exam_evidence_access_logs'): cursor.execute(""" CREATE TABLE exam_evidence_access_logs ( id VARCHAR(150) PRIMARY KEY, exam_evidence_id VARCHAR(150) NOT NULL, user_id VARCHAR(150) NOT NULL, access_type VARCHAR(50) NOT NULL, access_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, ip_address VARCHAR(45) NOT NULL, user_agent VARCHAR(500), session_id VARCHAR(100), access_reason VARCHAR(200), access_method VARCHAR(50), duration_seconds INTEGER, is_authorized INTEGER DEFAULT 1, security_violation INTEGER DEFAULT 0, violation_reason TEXT, country VARCHAR(100), city VARCHAR(100), device_type VARCHAR(50), browser VARCHAR(100), os VARCHAR(100), FOREIGN KEY (exam_evidence_id) REFERENCES exam_evidence(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id), CHECK (access_type IN ('view', 'download', 'print', 'share', 'delete', 'modify', 'upload', 'verify')), CHECK (access_method IN ('web', 'api', 'mobile', 'batch', 'system')), CHECK (duration_seconds >= 0) ) """) current_app.logger.info(" Created exam_evidence_access_logs table") # Create exam_evidence_permissions table if not self.check_table_exists(cursor, 'exam_evidence_permissions'): cursor.execute(""" CREATE TABLE exam_evidence_permissions ( id VARCHAR(150) PRIMARY KEY, exam_evidence_id VARCHAR(150) NOT NULL, user_id VARCHAR(150), role VARCHAR(50), can_view INTEGER DEFAULT 0, can_download INTEGER DEFAULT 0, can_print INTEGER DEFAULT 0, can_share INTEGER DEFAULT 0, can_modify INTEGER DEFAULT 0, can_delete INTEGER DEFAULT 0, can_verify INTEGER DEFAULT 0, granted_by VARCHAR(150) NOT NULL, granted_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME, is_active INTEGER DEFAULT 1, permission_reason TEXT, conditions TEXT, FOREIGN KEY (exam_evidence_id) REFERENCES exam_evidence(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id), FOREIGN KEY (granted_by) REFERENCES users(id), CHECK ((user_id IS NOT NULL) OR (role IS NOT NULL)) ) """) current_app.logger.info(" Created exam_evidence_permissions table") # Create exam_evidence_shares table if not self.check_table_exists(cursor, 'exam_evidence_shares'): cursor.execute(""" CREATE TABLE exam_evidence_shares ( id VARCHAR(150) PRIMARY KEY, exam_evidence_id VARCHAR(150) NOT NULL, shared_by VARCHAR(150) NOT NULL, shared_with VARCHAR(150), shared_with_email VARCHAR(120), share_type VARCHAR(50) NOT NULL, share_token VARCHAR(255), share_url VARCHAR(500), access_count INTEGER DEFAULT 0, max_access_count INTEGER, can_view INTEGER DEFAULT 1, can_download INTEGER DEFAULT 0, can_print INTEGER DEFAULT 0, shared_at DATETIME DEFAULT CURRENT_TIMESTAMP, expires_at DATETIME, last_accessed_at DATETIME, share_reason TEXT, share_message TEXT, is_active INTEGER DEFAULT 1, is_encrypted INTEGER DEFAULT 1, password_protected INTEGER DEFAULT 0, password_hash VARCHAR(255), FOREIGN KEY (exam_evidence_id) REFERENCES exam_evidence(id) ON DELETE CASCADE, FOREIGN KEY (shared_by) REFERENCES users(id), FOREIGN KEY (shared_with) REFERENCES users(id), CHECK (share_type IN ('internal', 'external', 'public', 'temporary')), CHECK (access_count >= 0), CHECK (max_access_count IS NULL OR max_access_count > 0) ) """) current_app.logger.info(" Created exam_evidence_shares table") # Create exam_evidence_audits table if not self.check_table_exists(cursor, 'exam_evidence_audits'): cursor.execute(""" CREATE TABLE exam_evidence_audits ( id VARCHAR(150) PRIMARY KEY, exam_evidence_id VARCHAR(150) NOT NULL, user_id VARCHAR(150), action VARCHAR(100) NOT NULL, action_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, old_values TEXT, new_values TEXT, ip_address VARCHAR(45), user_agent VARCHAR(500), session_id VARCHAR(100), request_id VARCHAR(100), action_reason TEXT, action_result VARCHAR(50), error_message TEXT, FOREIGN KEY (exam_evidence_id) REFERENCES exam_evidence(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id), CHECK (action_result IN ('success', 'failure', 'partial')) ) """) current_app.logger.info(" Created exam_evidence_audits table") # Create indexes for better performance indexes = [ "CREATE INDEX IF NOT EXISTS idx_exam_evidence_student_id ON exam_evidence(student_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_course_id ON exam_evidence(course_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_uploaded_by ON exam_evidence(uploaded_by)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_status ON exam_evidence(status)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_evidence_type ON exam_evidence(evidence_type)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_uploaded_at ON exam_evidence(uploaded_at)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_file_hash ON exam_evidence(file_hash)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_versions_evidence_id ON exam_evidence_versions(exam_evidence_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_versions_uploaded_at ON exam_evidence_versions(uploaded_at)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_access_logs_evidence_id ON exam_evidence_access_logs(exam_evidence_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_access_logs_user_id ON exam_evidence_access_logs(user_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_access_logs_timestamp ON exam_evidence_access_logs(access_timestamp)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_access_logs_ip_address ON exam_evidence_access_logs(ip_address)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_access_logs_violation ON exam_evidence_access_logs(security_violation)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_permissions_evidence_id ON exam_evidence_permissions(exam_evidence_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_permissions_user_id ON exam_evidence_permissions(user_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_permissions_role ON exam_evidence_permissions(role)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_permissions_active ON exam_evidence_permissions(is_active)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_shares_evidence_id ON exam_evidence_shares(exam_evidence_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_shares_shared_by ON exam_evidence_shares(shared_by)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_shares_shared_with ON exam_evidence_shares(shared_with)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_shares_token ON exam_evidence_shares(share_token)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_shares_expires_at ON exam_evidence_shares(expires_at)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_audits_evidence_id ON exam_evidence_audits(exam_evidence_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_audits_user_id ON exam_evidence_audits(user_id)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_audits_action ON exam_evidence_audits(action)", "CREATE INDEX IF NOT EXISTS idx_exam_evidence_audits_timestamp ON exam_evidence_audits(action_timestamp)" ] for index_sql in indexes: cursor.execute(index_sql) current_app.logger.info(" Created indexes for exam evidence tables") # Mark migration as completed self.migrations_run.add(migration_id) current_app.logger.info("Exam evidence system migration completed successfully!") return True except Exception as e: current_app.logger.error(f"Error in exam evidence system migration: {e}") return False def migrate_makeup_sessions_table(self, cursor): """Create makeup_sessions table and establish relationships.""" migration_id = "makeup_sessions_table" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting makeup sessions table migration...") # Step 1: Create makeup_sessions table if not self.check_table_exists(cursor, 'makeup_sessions'): current_app.logger.info("Creating makeup_sessions table...") cursor.execute(""" CREATE TABLE makeup_sessions ( id VARCHAR(150) PRIMARY KEY, daily_session_id VARCHAR(150), timetable_block_id VARCHAR(150), course_id VARCHAR(150) NOT NULL, tutor_id VARCHAR(150) NOT NULL, session_date DATE NOT NULL, start_time TIME NOT NULL, end_time TIME NOT NULL, room VARCHAR(100), session_type VARCHAR(50) DEFAULT 'lecture', status VARCHAR(20) DEFAULT 'scheduled', reason VARCHAR(200), original_session_date DATE, is_dynamic_block BOOLEAN DEFAULT 0, conflict_resolved BOOLEAN DEFAULT 0, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, created_by VARCHAR(150), is_verified BOOLEAN DEFAULT 0, verified_by VARCHAR(150), verified_at DATETIME, verification_method VARCHAR(50), notes TEXT, attendance_taken BOOLEAN DEFAULT 0, FOREIGN KEY (daily_session_id) REFERENCES daily_teaching_sessions(id), FOREIGN KEY (timetable_block_id) REFERENCES timetable_blocks(id), FOREIGN KEY (course_id) REFERENCES courses(id), FOREIGN KEY (tutor_id) REFERENCES users(id), FOREIGN KEY (created_by) REFERENCES users(id), FOREIGN KEY (verified_by) REFERENCES supervisors(id) ) """) current_app.logger.info(" Created makeup_sessions table") else: current_app.logger.info("✓ makeup_sessions table already exists") # Step 2: Add makeup_session_id column to attendance table if it doesn't exist if not self.check_column_exists(cursor, 'attendance', 'makeup_session_id'): current_app.logger.info("Adding makeup_session_id column to attendance table...") cursor.execute(""" ALTER TABLE attendance ADD COLUMN makeup_session_id VARCHAR(150) """) current_app.logger.info(" Added makeup_session_id column to attendance table") else: current_app.logger.info("✓ makeup_session_id column already exists in attendance table") # Step 3: Create indexes for makeup_sessions table indexes = [ "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_date ON makeup_sessions(session_date)", "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_tutor_date ON makeup_sessions(tutor_id, session_date)", "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_course_date ON makeup_sessions(course_id, session_date)", "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_status ON makeup_sessions(status)", "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_daily_session ON makeup_sessions(daily_session_id)", "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_timetable_block ON makeup_sessions(timetable_block_id)", "CREATE INDEX IF NOT EXISTS idx_makeup_sessions_original_date ON makeup_sessions(original_session_date)", "CREATE INDEX IF NOT EXISTS idx_attendance_makeup_session_id ON attendance(makeup_session_id)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Created indexes for makeup_sessions table") # Step 4: Create unique constraint for makeup sessions try: cursor.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS unique_makeup_session ON makeup_sessions(tutor_id, session_date, start_time, end_time) """) current_app.logger.info(" Created unique constraint for makeup sessions") except Exception as e: current_app.logger.warning(f"Could not create unique constraint: {e}") # Step 5: Verify the migration cursor.execute("SELECT COUNT(*) FROM makeup_sessions") result = cursor.fetchone() makeup_sessions_count = result[0] if result else 0 current_app.logger.info(f"✅ Migration completed. Makeup sessions table ready with {makeup_sessions_count} existing records") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error in makeup sessions table migration: {e}") return False def migrate_exam_grades_table(self, cursor): """Create exam_grades table and related tables.""" migration_id = "exam_grades_table" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting exam grades table migration...") # Step 1: Create exam_grades table if not self.check_table_exists(cursor, 'exam_grades'): current_app.logger.info("Creating exam_grades table...") cursor.execute(""" CREATE TABLE exam_grades ( id VARCHAR(150) PRIMARY KEY, student_id VARCHAR(150) NOT NULL, tutor_id VARCHAR(150) NOT NULL, course_id VARCHAR(150) NOT NULL, supervisor_id VARCHAR(150), exam_type VARCHAR(100) NOT NULL, exam_title VARCHAR(255) NOT NULL, grade_value REAL NOT NULL, max_grade REAL NOT NULL DEFAULT 100, grade_percentage REAL NOT NULL, letter_grade VARCHAR(5), exam_date DATETIME NOT NULL, graded_date DATETIME DEFAULT CURRENT_TIMESTAMP, academic_session VARCHAR(100), semester VARCHAR(50), is_verified BOOLEAN DEFAULT 0, verified_by VARCHAR(150), verified_date DATETIME, verification_notes TEXT, notes TEXT, is_active BOOLEAN DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (student_id) REFERENCES students (id), FOREIGN KEY (tutor_id) REFERENCES tutors (id), FOREIGN KEY (course_id) REFERENCES courses (id), FOREIGN KEY (supervisor_id) REFERENCES supervisors (id), FOREIGN KEY (verified_by) REFERENCES supervisors (id) ) """) current_app.logger.info(" Created exam_grades table") # Step 2: Create grade_access_logs table if not self.check_table_exists(cursor, 'grade_access_logs'): current_app.logger.info("Creating grade_access_logs table...") cursor.execute(""" CREATE TABLE grade_access_logs ( id VARCHAR(150) PRIMARY KEY, exam_grade_id VARCHAR(150), exam_evidence_id VARCHAR(150), accessed_by VARCHAR(150) NOT NULL, user_type VARCHAR(50) NOT NULL, access_type VARCHAR(50) NOT NULL, ip_address VARCHAR(45), user_agent TEXT, session_id VARCHAR(255), accessed_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (exam_grade_id) REFERENCES exam_grades (id), FOREIGN KEY (exam_evidence_id) REFERENCES exam_evidence (id) ) """) current_app.logger.info(" Created grade_access_logs table") # Step 3: Create grade_statistics table if not self.check_table_exists(cursor, 'grade_statistics'): current_app.logger.info("Creating grade_statistics table...") cursor.execute(""" CREATE TABLE grade_statistics ( id VARCHAR(150) PRIMARY KEY, course_id VARCHAR(150) NOT NULL, tutor_id VARCHAR(150), academic_session VARCHAR(100) NOT NULL, semester VARCHAR(50) NOT NULL, total_students INTEGER DEFAULT 0, total_exams INTEGER DEFAULT 0, average_grade REAL DEFAULT 0.0, highest_grade REAL DEFAULT 0.0, lowest_grade REAL DEFAULT 0.0, pass_rate REAL DEFAULT 0.0, grade_a_count INTEGER DEFAULT 0, grade_b_count INTEGER DEFAULT 0, grade_c_count INTEGER DEFAULT 0, grade_d_count INTEGER DEFAULT 0, grade_f_count INTEGER DEFAULT 0, verified_grades_count INTEGER DEFAULT 0, pending_verification_count INTEGER DEFAULT 0, calculated_at DATETIME DEFAULT CURRENT_TIMESTAMP, last_updated DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (course_id) REFERENCES courses (id), FOREIGN KEY (tutor_id) REFERENCES tutors (id), UNIQUE(course_id, tutor_id, academic_session, semester) ) """) current_app.logger.info(" Created grade_statistics table") # Step 4: Add exam_grade_id column to exam_evidence table if it doesn't exist if not self.check_column_exists(cursor, 'exam_evidence', 'exam_grade_id'): current_app.logger.info("Adding exam_grade_id column to exam_evidence table...") cursor.execute(""" ALTER TABLE exam_evidence ADD COLUMN exam_grade_id VARCHAR(150) """) current_app.logger.info(" Added exam_grade_id column to exam_evidence table") # Step 5: Create indexes indexes = [ "CREATE INDEX IF NOT EXISTS idx_exam_grades_student_id ON exam_grades(student_id)", "CREATE INDEX IF NOT EXISTS idx_exam_grades_tutor_id ON exam_grades(tutor_id)", "CREATE INDEX IF NOT EXISTS idx_exam_grades_course_id ON exam_grades(course_id)", "CREATE INDEX IF NOT EXISTS idx_exam_grades_supervisor_id ON exam_grades(supervisor_id)", "CREATE INDEX IF NOT EXISTS idx_exam_grades_is_verified ON exam_grades(is_verified)", "CREATE INDEX IF NOT EXISTS idx_exam_grades_exam_date ON exam_grades(exam_date)", "CREATE INDEX IF NOT EXISTS idx_grade_access_logs_accessed_by ON grade_access_logs(accessed_by)", "CREATE INDEX IF NOT EXISTS idx_grade_access_logs_exam_grade_id ON grade_access_logs(exam_grade_id)", "CREATE INDEX IF NOT EXISTS idx_grade_access_logs_accessed_at ON grade_access_logs(accessed_at)", "CREATE INDEX IF NOT EXISTS idx_grade_statistics_course_id ON grade_statistics(course_id)", "CREATE INDEX IF NOT EXISTS idx_grade_statistics_tutor_id ON grade_statistics(tutor_id)", "CREATE INDEX IF NOT EXISTS idx_grade_statistics_academic_session ON grade_statistics(academic_session)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Created indexes for exam grades tables") # Step 6: Verify the migration cursor.execute("SELECT COUNT(*) FROM exam_grades") result = cursor.fetchone() exam_grades_count = result[0] if result else 0 current_app.logger.info(f"✅ Migration completed. Exam grades system ready with {exam_grades_count} existing records") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error in exam grades table migration: {e}") return False def migrate_grade_access_logs_constraints(self, cursor): """Update grade_access_logs table constraints to include missing access types.""" migration_id = "grade_access_logs_constraints" if migration_id in self.migrations_run: return True try: current_app.logger.info("Updating grade_access_logs constraints...") # Check if the table exists if not self.check_table_exists(cursor, 'grade_access_logs'): current_app.logger.info("grade_access_logs table doesn't exist, skipping constraint update") return True # SQLite doesn't support ALTER TABLE ADD CONSTRAINT for CHECK constraints # Instead, we'll recreate the table with the correct constraint current_app.logger.info("SQLite doesn't support ALTER TABLE ADD CONSTRAINT for CHECK constraints") current_app.logger.info("Skipping constraint update - table structure is already correct") # Check if the constraint already allows the required values # Since SQLite doesn't enforce CHECK constraints strictly, we'll just log this current_app.logger.info("Note: SQLite CHECK constraints are not enforced, so this is informational only") current_app.logger.info(" Updated grade_access_logs constraints") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error updating grade_access_logs constraints: {e}") return False def migrate_attendance_supervisor_tutor_id(self, cursor): """Add supervisor_tutor_id column to attendance table.""" migration_id = "attendance_supervisor_tutor_id" if migration_id in self.migrations_run: return True try: current_app.logger.info("Adding supervisor_tutor_id to attendance table...") # Step 1: Add supervisor_tutor_id column to attendance table if it doesn't exist if not self.check_column_exists(cursor, 'attendance', 'supervisor_tutor_id'): current_app.logger.info("Adding supervisor_tutor_id column to attendance table...") cursor.execute(""" ALTER TABLE attendance ADD COLUMN supervisor_tutor_id VARCHAR(150) """) current_app.logger.info(" Added supervisor_tutor_id column to attendance table") else: current_app.logger.info("✓ supervisor_tutor_id column already exists in attendance table") # Step 2: Create index for supervisor_tutor_id column cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_attendance_supervisor_tutor_id ON attendance(supervisor_tutor_id) """) # Step 3: Ensure daily_session_id column exists (backup for missed migrations) if not self.check_column_exists(cursor, 'attendance', 'daily_session_id'): current_app.logger.info("Adding daily_session_id column to attendance table...") cursor.execute(""" ALTER TABLE attendance ADD COLUMN daily_session_id VARCHAR(150) """) current_app.logger.info(" Added daily_session_id column to attendance table") # Create index cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_attendance_daily_session_id ON attendance(daily_session_id) """) else: current_app.logger.info("✓ daily_session_id column already exists in attendance table") # Step 4: Ensure makeup_session_id column exists (backup for missed migrations) if not self.check_column_exists(cursor, 'attendance', 'makeup_session_id'): current_app.logger.info("Adding makeup_session_id column to attendance table...") cursor.execute(""" ALTER TABLE attendance ADD COLUMN makeup_session_id VARCHAR(150) """) current_app.logger.info(" Added makeup_session_id column to attendance table") # Create index cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_attendance_makeup_session_id ON attendance(makeup_session_id) """) else: current_app.logger.info("✓ makeup_session_id column already exists in attendance table") current_app.logger.info(" Attendance table schema update completed") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error updating attendance table schema: {e}") return False def migrate_reminders_table_columns(self, cursor): """Add missing columns to reminders table.""" migration_id = "reminders_table_columns" if migration_id in self.migrations_run: return True try: current_app.logger.info("Adding missing columns to reminders table...") if not self.check_table_exists(cursor, 'reminders'): current_app.logger.info("reminders table not found, skipping migration") self.migrations_run.add(migration_id) return True # Add created_at column if it doesn't exist if not self.check_column_exists(cursor, 'reminders', 'created_at'): current_app.logger.info("Adding created_at column to reminders table...") cursor.execute(""" ALTER TABLE reminders ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP """) current_app.logger.info(" Added created_at column to reminders table") else: current_app.logger.info("✓ created_at column already exists in reminders table") # Add status column if it doesn't exist (for backward compatibility) if not self.check_column_exists(cursor, 'reminders', 'status'): current_app.logger.info("Adding status column to reminders table...") cursor.execute(""" ALTER TABLE reminders ADD COLUMN status VARCHAR(20) DEFAULT 'sent' """) current_app.logger.info(" Added status column to reminders table") else: current_app.logger.info("✓ status column already exists in reminders table") # Create indexes for better performance try: cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_reminders_created_at ON reminders(created_at) """) current_app.logger.info(" Added index for created_at column") except Exception as e: current_app.logger.warning(f"Could not create index for created_at: {e}") try: cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_reminders_sent_at ON reminders(sent_at) """) current_app.logger.info(" Added index for sent_at column") except Exception as e: current_app.logger.warning(f"Could not create index for sent_at: {e}") current_app.logger.info(" Reminders table schema update completed") self.migrations_run.add(migration_id) return True except Exception as e: current_app.logger.error(f"Error updating reminders table schema: {e}") return False def migrate_quiz_system_enhancements(self, cursor): """Enhance quiz system with flexible associations, manual grading, and better tracking.""" migration_id = "quiz_system_enhancements" if migration_id in self.migrations_run: return True try: current_app.logger.info("Starting quiz system enhancements migration...") # Step 1: Add created_by column to module_quizzes if it doesn't exist if self.check_table_exists(cursor, 'module_quizzes'): if not self.check_column_exists(cursor, 'module_quizzes', 'created_by'): current_app.logger.info("Adding created_by column to module_quizzes table...") cursor.execute(""" ALTER TABLE module_quizzes ADD COLUMN created_by VARCHAR(150) """) current_app.logger.info(" Added created_by column to module_quizzes table") # Create index for created_by cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_module_quizzes_created_by ON module_quizzes(created_by) """) current_app.logger.info(" Added index for created_by column") else: current_app.logger.info("✓ created_by column already exists in module_quizzes table") # Add additional indexes for better query performance indexes = [ "CREATE INDEX IF NOT EXISTS idx_module_quizzes_is_published ON module_quizzes(is_published)", "CREATE INDEX IF NOT EXISTS idx_module_quizzes_is_active ON module_quizzes(is_active)", "CREATE INDEX IF NOT EXISTS idx_module_quizzes_due_date ON module_quizzes(due_date)", "CREATE INDEX IF NOT EXISTS idx_module_quizzes_quiz_type ON module_quizzes(quiz_type)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Added indexes for module_quizzes table") else: current_app.logger.info("module_quizzes table not found, skipping quiz table updates") # Step 2: Add manual grading columns to quiz_answers table if self.check_table_exists(cursor, 'quiz_answers'): columns_to_add = [ ('graded_by', 'VARCHAR(150)'), ('graded_at', 'DATETIME'), ('grading_notes', 'TEXT') ] for column_name, column_type in columns_to_add: if not self.check_column_exists(cursor, 'quiz_answers', column_name): current_app.logger.info(f"Adding {column_name} column to quiz_answers table...") cursor.execute(f""" ALTER TABLE quiz_answers ADD COLUMN {column_name} {column_type} """) current_app.logger.info(f"✅ Added {column_name} column to quiz_answers table") else: current_app.logger.info(f"✓ {column_name} column already exists in quiz_answers table") # Add indexes for quiz_answers indexes = [ "CREATE INDEX IF NOT EXISTS idx_quiz_answers_attempt_id ON quiz_answers(attempt_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_answers_question_id ON quiz_answers(question_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_answers_is_correct ON quiz_answers(is_correct)", "CREATE INDEX IF NOT EXISTS idx_quiz_answers_graded_by ON quiz_answers(graded_by)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Added indexes for quiz_answers table") else: current_app.logger.info("quiz_answers table not found, skipping quiz answers updates") # Step 3: Add indexes for quiz_questions table if self.check_table_exists(cursor, 'quiz_questions'): indexes = [ "CREATE INDEX IF NOT EXISTS idx_quiz_questions_quiz_id ON quiz_questions(quiz_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_questions_is_active ON quiz_questions(is_active)", "CREATE INDEX IF NOT EXISTS idx_quiz_questions_question_type ON quiz_questions(question_type)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Added indexes for quiz_questions table") # Step 4: Add indexes for quiz_options table if self.check_table_exists(cursor, 'quiz_options'): indexes = [ "CREATE INDEX IF NOT EXISTS idx_quiz_options_question_id ON quiz_options(question_id)", "CREATE INDEX IF NOT EXISTS idx_quiz_options_is_correct ON quiz_options(is_correct)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Added indexes for quiz_options table") # Step 5: Add indexes for quiz_attempts table if self.check_table_exists(cursor, 'quiz_attempts'): indexes = [ "CREATE INDEX IF NOT EXISTS idx_quiz_attempts_is_completed ON quiz_attempts(is_completed)", "CREATE INDEX IF NOT EXISTS idx_quiz_attempts_is_passed ON quiz_attempts(is_passed)", "CREATE INDEX IF NOT EXISTS idx_quiz_attempts_percentage_score ON quiz_attempts(percentage_score)" ] for index_sql in indexes: try: cursor.execute(index_sql) except Exception as e: current_app.logger.warning(f"Could not create index: {e}") current_app.logger.info(" Added indexes for quiz_attempts table") # Step 6: Update existing quiz data # Set created_by to NULL for existing quizzes (will be updated when tutors edit them) if self.check_table_exists(cursor, 'module_quizzes') and self.check_column_exists(cursor, 'module_quizzes', 'created_by'): cursor.execute(""" UPDATE module_quizzes SET created_by = NULL WHERE created_by IS NULL OR created_by = '' """) current_app.logger.info(" Initialized created_by column for existing quizzes") # Step 7: Verify quiz counts if self.check_table_exists(cursor, 'module_quizzes'): cursor.execute("SELECT COUNT(*) FROM module_quizzes WHERE is_active = 1") result = cursor.fetchone() active_quizzes = result[0] if result else 0 if self.check_table_exists(cursor, 'quiz_questions'): cursor.execute("SELECT COUNT(*) FROM quiz_questions WHERE is_active = 1") result = cursor.fetchone() active_questions = result[0] if result else 0 else: active_questions = 0 if self.check_table_exists(cursor, 'quiz_options'): cursor.execute("SELECT COUNT(*) FROM quiz_options") result = cursor.fetchone() total_options = result[0] if result else 0 else: total_options = 0 current_app.logger.info(f"Quiz System Stats: {active_quizzes} active quizzes, {active_questions} questions, {total_options} options") # Mark migration as completed self.migrations_run.add(migration_id) current_app.logger.info("Quiz system enhancements migration completed successfully!") return True except Exception as e: current_app.logger.error(f"Error in quiz system enhancements migration: {e}") import traceback current_app.logger.error(traceback.format_exc()) return False def migrate_course_resources_level_id(self, cursor): """Add level_id column to course_resources table for level association""" migration_id = "course_resources_level_id" if migration_id in self.migrations_run: current_app.logger.info(f"Skipping {migration_id} - already completed") return True try: current_app.logger.info("Starting course resources level_id migration...") # Check if course_resources table exists if not self.check_table_exists(cursor, 'course_resources'): current_app.logger.info("⏭️ Skipping - course_resources table doesn't exist yet") self.migrations_run.add(migration_id) return True # Check if level_id column already exists if self.check_column_exists(cursor, 'course_resources', 'level_id'): current_app.logger.info("⏭️ level_id column already exists in course_resources") self.migrations_run.add(migration_id) return True # Add level_id column current_app.logger.info("➕ Adding level_id column to course_resources table...") cursor.execute(""" ALTER TABLE course_resources ADD COLUMN level_id VARCHAR(150) """) # Create index for level_id current_app.logger.info("📇 Creating index on level_id...") cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_course_resources_level_id ON course_resources(level_id) """) cursor.connection.commit() # Log statistics cursor.execute("SELECT COUNT(*) FROM course_resources") result = cursor.fetchone() total_resources = result[0] if result else 0 cursor.execute("SELECT COUNT(*) FROM course_resources WHERE module_id IS NOT NULL") result = cursor.fetchone() module_resources = result[0] if result else 0 cursor.execute("SELECT COUNT(*) FROM course_resources WHERE module_id IS NULL AND level_id IS NULL") result = cursor.fetchone() general_resources = result[0] if result else 0 current_app.logger.info(f"Course Resources Stats: {total_resources} total, {module_resources} in modules, {general_resources} general") # Mark migration as completed self.migrations_run.add(migration_id) current_app.logger.info("Course resources level_id migration completed successfully!") return True except Exception as e: current_app.logger.error(f"Error in course resources level_id migration: {e}") import traceback current_app.logger.error(traceback.format_exc()) return False def run_all_migrations(self, db_path): """Run all database migrations.""" try: current_app.logger.info(f"Starting database migrations for: {db_path}") current_app.logger.info(f"Timestamp: {datetime.now()}") conn = sqlite3.connect(db_path) cursor = conn.cursor() # Run migrations migrations = [ self.migrate_daily_teaching_sessions, self.migrate_enrollment_model_restructure, self.migrate_course_structure_models, self.migrate_tutor_departments_unique_constraint_fix, self.migrate_courses_total_hours_constraint_fix, self.migrate_exam_evidence_system, self.migrate_daily_teaching_sessions_teaching_session_relation, self.migrate_makeup_sessions_table, self.migrate_exam_grades_table, self.migrate_grade_access_logs_constraints, self.migrate_attendance_supervisor_tutor_id, self.migrate_reminders_table_columns, self.migrate_quiz_system_enhancements, self.migrate_course_resources_level_id ] success_count = 0 for migration in migrations: try: if migration(cursor): success_count += 1 except Exception as e: current_app.logger.error(f"Migration failed: {e}") # Commit all changes conn.commit() conn.close() current_app.logger.info(f"✅ Migrations completed: {success_count}/{len(migrations)} successful") return success_count == len(migrations) except Exception as e: current_app.logger.error(f"Migration process failed: {e}") return False def run_database_migrations(): """ Run database migrations for all database files. This function should be called from the Flask app's before_request handler. """ migrator = DatabaseMigrator() try: # Get database path from Flask config database_url = current_app.config.get('DATABASE_URL', 'sqlite:///test.db') # Ensure we have a valid database URL if not database_url: current_app.logger.warning("No DATABASE_URL found in config, using default") database_url = 'sqlite:///test.db' current_app.logger.info(f"Database URL: {database_url}") # Extract database file path from SQLAlchemy URL if database_url.startswith('sqlite:///'): db_path = database_url.replace('sqlite:///', '') # Handle absolute paths if not os.path.isabs(db_path): # Try relative to the server_api directory server_api_dir = os.path.dirname(os.path.dirname(os.path.dirname(__file__))) db_path = os.path.join(server_api_dir, db_path) # Normalize path db_path = os.path.normpath(db_path) current_app.logger.info(f"Resolved database path: {db_path}") # Check if database file exists if not os.path.exists(db_path): current_app.logger.warning(f"Database file not found: {db_path}") return False # Run migrations return migrator.run_all_migrations(db_path) current_app.logger.warning(f"Database URL not supported for migrations: {database_url}") return False except Exception as e: current_app.logger.error(f"Error in database migrations: {e}") return False def run_remote_database_migrations(remote_db_config): """ Run database migrations for remote databases. Args: remote_db_config: Dictionary containing remote database configuration { 'host': 'remote_host', 'database': 'database_name', 'user': 'username', 'password': 'password', 'port': 5432 # or other port } """ migrator = DatabaseMigrator() try: # For remote databases, you would need to use the appropriate database adapter # This is a placeholder for PostgreSQL/MySQL remote migrations current_app.logger.info(f"Starting remote database migrations for: {remote_db_config.get('database')}") # Example for PostgreSQL (you would need psycopg2 installed) # import psycopg2 # conn = psycopg2.connect( # host=remote_db_config['host'], # database=remote_db_config['database'], # user=remote_db_config['user'], # password=remote_db_config['password'], # port=remote_db_config.get('port', 5432) # ) # For now, we'll use SQLite approach but adapt for your remote database current_app.logger.info("Remote database migration not implemented yet") current_app.logger.info("Please run migrations manually on your remote database") return True except Exception as e: current_app.logger.error(f"Remote migration failed: {e}") return False