from src import db from src.models import StudentTable, TeacherTable from sqlalchemy import Select import pandas as pd from sqlalchemy.orm.attributes import flag_modified import io from src.lib.bcrypt import encrypt_password ALLOWED_EXTENSIONS = {'xlsx', 'xls'} REQUIRED_HEADERS = { 'student': ['firstname', 'lastname', 'roll_no', 'grade', 'section'], 'teacher': ['firstname', 'lastname', 'grade', 'section'] } def allowed_file(filename): return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS def generate_excel_report(result, file_type, school_id): """Generate Excel file with processed data""" if file_type == "student": # Fetch all students for this school stmt = Select(StudentTable).where(StudentTable.school_id == school_id) all_records = db.session.execute(stmt).scalars().all() # Create DataFrame data = [] for student in all_records: data.append({ 'Firstname': student.firstname, 'Lastname': student.lastname, 'Roll No': student.roll_no, 'Grade': student.grade, 'Section': student.section, 'Username': student.username }) df = pd.DataFrame(data) else: # teacher # Fetch all teachers for this school stmt = Select(TeacherTable).where(TeacherTable.school_id == school_id) all_records = db.session.execute(stmt).scalars().all() # Create DataFrame data = [] for teacher in all_records: for grade_info in teacher.grades: data.append({ 'Firstname': teacher.firstname, 'Lastname': teacher.lastname, 'Grade': grade_info.get('class'), 'Section': grade_info.get('section'), 'Username': teacher.username }) df = pd.DataFrame(data) # Create Excel file in memory output = io.BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: # Write data sheet df.to_excel(writer, sheet_name='Data', index=False) # Write summary sheet summary_data = { 'Metric': ['Added', 'Skipped', 'Total'], 'Count': [ result.get('added', 0), result.get('skipped', 0), result.get('added', 0) + result.get('skipped', 0) ] } if file_type == 'teacher': summary_data['Metric'].insert(1, 'Updated') summary_data['Count'].insert(1, result.get('updated', 0)) summary_df = pd.DataFrame(summary_data) summary_df.to_excel(writer, sheet_name='Summary', index=False) output.seek(0) return output def process_student_file(file_path, school_id): """Process student Excel file and add students to database""" df = pd.read_excel(file_path) df.columns = [col.lower().strip() for col in df.columns] # Get all existing students for this school stmt_all_student = Select(StudentTable).where(StudentTable.school_id == school_id) all_students = db.session.execute(stmt_all_student).scalars().all() ranking_start = len(all_students) + 1 # Create set of existing students (firstname, lastname, grade, section, roll_no) existing_student_set = set( ( stu.firstname.lower().strip(), (stu.lastname or "").lower().strip(), str(stu.grade).strip(), str(stu.section).strip(), stu.roll_no ) for stu in all_students ) # Create set of existing roll numbers per grade-section existing_roll_nos = {} for stu in all_students: key = (str(stu.grade).strip(), str(stu.section).strip()) if key not in existing_roll_nos: existing_roll_nos[key] = set() existing_roll_nos[key].add(stu.roll_no) added_count = 0 skipped_count = 0 errors = [] for idx, row in df.iterrows(): try: firstname = str(row.get("firstname", "")).strip() lastname = str(row.get("lastname", "")).strip() grade = str(row.get("grade", "")).strip() section = str(row.get("section", "")).strip() roll_no = row.get("roll_no") # Validate required fields if not firstname or not lastname or not grade or not section or pd.isna(roll_no): skipped_count += 1 errors.append(f"Row {idx + 2}: Missing required fields") continue try: roll_no = int(roll_no) except (ValueError, TypeError): skipped_count += 1 errors.append(f"Row {idx + 2}: Invalid roll number") continue # Check if student already exists student_key = ( firstname.lower(), lastname.lower(), grade, section, roll_no ) if student_key in existing_student_set: skipped_count += 1 continue # Check if roll number is duplicate in same grade-section grade_section_key = (grade, section) if grade_section_key in existing_roll_nos and roll_no in existing_roll_nos[grade_section_key]: skipped_count += 1 errors.append(f"Row {idx + 2}: Roll number {roll_no} already exists in Grade {grade}, Section {section}") continue # Generate unique username username = f'{firstname.lower()}_{school_id}{roll_no}{grade}{section}' # Create default password (you can modify this logic) default_password = f'{firstname.lower()}{roll_no}' # Create new student new_student = StudentTable( username=username, firstname=firstname, lastname=lastname, grade=grade, section=section, roll_no=roll_no, school_id=school_id, overall_ranking=ranking_start, speed_ranking=ranking_start, accuracy_ranking=ranking_start ) new_student.set_password(default_password) db.session.add(new_student) # Update tracking sets existing_student_set.add(student_key) if grade_section_key not in existing_roll_nos: existing_roll_nos[grade_section_key] = set() existing_roll_nos[grade_section_key].add(roll_no) added_count += 1 except Exception as e: skipped_count += 1 errors.append(f"Row {idx + 2}: {str(e)}") # Commit all changes db.session.commit() message = f"Students processed: {added_count} added, {skipped_count} skipped" if errors and len(errors) <= 5: message += f". Errors: {'; '.join(errors)}" return { "success": True, "message": message, "added": added_count, "skipped": skipped_count } def process_teacher_file(file_path, school_id, stream): """Process teacher Excel file and add/update teachers in database""" df = pd.read_excel(file_path) df.columns = [col.lower().strip() for col in df.columns] # Get all existing teachers for this school stmt_all_teachers = Select(TeacherTable).where(TeacherTable.school_id == school_id) all_teachers = db.session.execute(stmt_all_teachers).scalars().all() # Create dictionary of existing teachers by name existing_teachers = {} for teacher in all_teachers: key = (teacher.firstname.lower().strip(), (teacher.lastname or "").lower().strip()) existing_teachers[key] = teacher added_count = 0 updated_count = 0 skipped_count = 0 errors = [] for idx, row in df.iterrows(): try: firstname = str(row.get("firstname", "")).strip() lastname = str(row.get("lastname", "")).strip() grade = str(row.get("grade", "")).strip() section = str(row.get("section", "")).strip() # Validate required fields if not firstname or not lastname or not grade or not section: skipped_count += 1 errors.append(f"Row {idx + 2}: Missing required fields") continue teacher_key = (firstname.lower(), lastname.lower()) # Check if teacher exists if teacher_key in existing_teachers: teacher = existing_teachers[teacher_key] # Check if grade-section combination already exists grade_section_exists = any( g.get("class") == grade and g.get("section") == section for g in teacher.grades ) if grade_section_exists: skipped_count += 1 continue else: # Add new grade-section to existing teacher teacher.grades.append({"class": grade, "section": section}) # CRITICAL FIX: Mark the 'grades' field as modified # This tells SQLAlchemy that the mutable list has changed flag_modified(teacher, 'grades') db.session.add(teacher) updated_count += 1 else: # Create new teacher username = f'{firstname.lower()}{lastname.lower()}_{school_id}' default_password = f'{firstname.lower()}{lastname.lower()}' new_teacher = TeacherTable( firstname=firstname, lastname=lastname, username=username, grades=[{"class": grade, "section": section}], school_id=school_id, password=encrypt_password(default_password), stream=stream ) db.session.add(new_teacher) existing_teachers[teacher_key] = new_teacher added_count += 1 except Exception as e: skipped_count += 1 errors.append(f"Row {idx + 2}: {str(e)}") # Commit all changes db.session.commit() message = f"Teachers processed: {added_count} added, {updated_count} updated, {skipped_count} skipped" if errors and len(errors) <= 5: message += f". Errors: {'; '.join(errors)}" return { "success": True, "message": message, "added": added_count, "updated": updated_count, "skipped": skipped_count } def validate_headers(file_path, file_type): try: df = pd.read_excel(file_path) file_headers = [col.lower().strip() for col in df.columns] required_headers = REQUIRED_HEADERS.get(file_type, []) missing_headers = [h for h in required_headers if h not in file_headers] if missing_headers: return False, f"Missing required headers: {', '.join(missing_headers)}" extra_headers = [h for h in file_headers if h not in required_headers] if extra_headers: return False, f"Extra invalid headers found: {', '.join(extra_headers)}" if len(file_headers) != len(required_headers): return False, f"Header count mismatch. Expected {len(required_headers)} headers but found {len(file_headers)}." return True, "Headers validated successfully" except Exception as e: return False, f"Error reading file: {str(e)}"