from datetime import datetime, timedelta from typing import Literal from fastapi import HTTPException, UploadFile from fastapi.security import OAuth2PasswordRequestForm import pandas as pd import app.api.student.schemas as schemas from sqlalchemy.orm import Session from app.dependency.authantication import JWTManager, JWTPayloadSchema from app.locale.messages import Messages from app.models.main.group import TblGroup from app.models.main.student import StudentBase, TblStudent from app.utils.common_utils import generate_secure_password from app.utils.email import send_email from app.utils.schemas_utils import CustomResponse VALID_ROLES: tuple[Literal["student", "admin", "superadmin"], ...] = ( "student", "admin", "superadmin" ) class StudentService: def __init__(self, db: Session,token:JWTPayloadSchema): self.db = db self.token = token async def create_student(self, request: schemas.StudentCreate): created_student = StudentBase.model_validate(request) TblStudent.create(created_student, self.db) self.db.commit() return CustomResponse(status="1", message=Messages.STUDENT_CREAT) async def get_student(self, student_id: int): student = TblStudent.get_by_id(student_id, self.db) return schemas.StudentResponse.model_validate(student) async def get_student_get(self, student_id: int): student= TblStudent.get_by_id(student_id, self.db) if not student: raise HTTPException(status_code=404, detail=Messages.STUDENT_NOT_FOUND) return schemas.StudentResponse.model_validate(student, from_attributes=True) async def get_student_group(self, student_id: int): student = TblStudent.get_by_id(student_id, self.db) if not student: raise HTTPException(status_code=404, detail=Messages.STUDENT_NOT_FOUND) if not student.group: raise HTTPException(status_code=404, detail=Messages.GROUP_NOT_ASSIGNED) return schemas.GroupResponse.model_validate(student.group, from_attributes=True) async def update_student(self, request: schemas.StudentUpdate): updated_student = StudentBase.model_validate(request) if updated_student.student_id is None: return CustomResponse(status="-1", message=Messages.SEGMENT_NOT_FOUND) TblStudent.update(updated_student.student_id,updated_student, self.db) self.db.commit() return CustomResponse(status="1", message=Messages.STUDENT_UPDATE) @staticmethod async def student_login(credentials: OAuth2PasswordRequestForm, db: Session): student = db.query(TblStudent).filter(TblStudent.student_email == credentials.username).first() if not student: raise HTTPException(status_code=404, detail=Messages.STUDENT_NOT_FOUND) if credentials.password != student.password: raise HTTPException(status_code=401, detail=Messages.INCORRECT_PASSWORD) if student.group_code: active = db.query(TblStudent).filter( TblStudent.group_code == student.group_code, TblStudent.login_status == "1", TblStudent.student_id != student.student_id ).first() if active: raise HTTPException( status_code=403, detail=f"Student '{active.first_name}' from your group is already logged in." ) student.login_status = "1" db.commit() payload = JWTPayloadSchema( student_id=student.student_id, user_type="student", user_role="student", exp=datetime.utcnow() + timedelta(minutes=30) ) access_token = JWTManager.create_access_token(payload) group_id = student.group.group_id if student.group else None return { "access_token": access_token, "token_type": "bearer", "email_id": student.student_email, "role": "student", "student_id":student.student_id, "group_id": group_id } def student_logout(self): if self.token.student_id is None: raise HTTPException(status_code=400, detail=Messages.STUDENT_ID_REQUIED) student = self.db.query(TblStudent).filter(TblStudent.student_id == self.token.student_id ).first() if not student: raise HTTPException(status_code=404, detail=Messages.CATCHMENT_NOT_FOUND) if student.login_status != "1": raise HTTPException(status_code=409, detail=Messages.STUDENT_NOT_LOGIN) student.login_status = "0" self.db.commit() return CustomResponse(status="1", message=Messages.STUDENT_LOGOUT) async def upload_students_from_excel(self, file: UploadFile): try: file.file.seek(0) df = pd.read_excel(file.file, engine="openpyxl", header=None) expected_headers = [ "first_name", "last_name", "student_email", "group_code", "simulation_code", "group_name", "group_email", "number_of_members", "category_assigned" ] header_row_index = None for i, row in df.iterrows(): normalized = [str(col).strip().lower().replace(" ", "_") for col in row] if normalized == expected_headers: header_row_index = i break if header_row_index is None: return { "status": "-1", "error": "Missing required headers in Excel file", "expected_headers": expected_headers, "sample_rows": df.head(5).values.tolist() } file.file.seek(0) df = pd.read_excel(file.file, engine="openpyxl", header=header_row_index) df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_") df = df.where(pd.notnull(df), None) if "password" in df.columns: df.drop(columns=["password"], inplace=True) group_student_map = {} for _, row in df.iterrows(): group_code = row.get("group_code") if group_code: group_student_map.setdefault(group_code, []).append(row) error_messages = [] warning_messages = [] for group_code, students in group_student_map.items(): first = students[0] student_count = len(students) group = self.db.query(TblGroup).filter_by(group_code=group_code).first() if not group: number_of_members = int(first["number_of_members"]) if first["number_of_members"] else student_count group = TblGroup( group_code=group_code, simulation_code=first["simulation_code"], group_name=first["group_name"], group_email=first["group_email"], number_of_members=number_of_members, category_assigned=first.get("category_assigned") ) self.db.add(group) self.db.commit() self.db.refresh(group) else: number_of_members = group.number_of_members or 0 # Validation Conditions if number_of_members < student_count: error_messages.append( f"Group {group_code}: You can only add {number_of_members} students, but you tried to add {student_count}." ) continue # skip this group elif number_of_members > student_count: remaining = number_of_members - student_count warning_messages.append( f"Group {group_code}: You added {student_count} students, but {remaining} spots are still available." ) # Update fields if necessary updated = False if not group.number_of_members or group.number_of_members == 0: group.number_of_members = student_count updated = True for field in ["group_name", "group_email", "simulation_code", "category_assigned"]: if first.get(field) and getattr(group, field) != first[field]: setattr(group, field, first[field]) updated = True if updated: self.db.commit() for row in students: if self.db.query(TblStudent).filter_by(student_email=row["student_email"]).first(): continue password_plain = generate_secure_password() student = TblStudent( first_name=row["first_name"], last_name=row["last_name"], student_email=row["student_email"], password=password_plain, role="student", group_code=group_code ) self.db.add(student) self.db.commit() if error_messages: return { "status": "0", "message": "Some groups were not uploaded due to exceeding the allowed number of members.", "errors": error_messages, "warnings": warning_messages } return { "status": "1", "message": "Upload successful", "warnings": warning_messages } except Exception as e: raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}") def send_password_to_email(self, student_email: str): student = self.db.query(TblStudent).filter_by(student_email=student_email).first() if not student: raise HTTPException(status_code=404, detail="Student not found") subject = "Your Student Portal Password" body = f"Hello {student.first_name},\n\nYour password is: {student.password}\n\nPlease keep it secure." success = send_email(to=student_email, subject=subject, body=body) if not success: raise HTTPException(status_code=500, detail="Failed to send email") return {"status": "1", "message": "Password sent to email"} def send_passwords_to_all_students(self): students = self.db.query(TblStudent).all() if not students: raise HTTPException(status_code=404, detail="No students found") failures = [] for student in students: subject = "Your Student Portal Password" body = f"Hello {student.first_name},\n\nYour password is: {student.password}\n\nPlease keep it secure." success = send_email(to=student.student_email, subject=subject, body=body) if not success: failures.append(student.student_email) if failures: return { "status": "0", "message": "Some emails failed to send", "failed_emails": failures } return {"status": "1", "message": "Passwords sent to all student emails"} # async def upload_students_from_excel(self, file: UploadFile): # try: # file.file.seek(0) # df = pd.read_excel(file.file, engine="openpyxl", header=None) # expected_headers = [ # "first_name", "last_name", "student_email", "group_code", # "simulation_code", "group_name", "group_email", "number_of_members", # "category_assigned" # ] # header_row_index = None # for i, row in df.iterrows(): # normalized = [str(col).strip().lower().replace(" ", "_") for col in row] # if normalized == expected_headers: # header_row_index = i # break # if header_row_index is None: # return { # "error": "Missing required headers in Excel file", # "expected_headers": expected_headers, # "sample_rows": df.head(5).values.tolist() # } # file.file.seek(0) # df = pd.read_excel(file.file, engine="openpyxl", header=header_row_index) # df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_") # if "password" in df.columns: # df.drop(columns=["password"], inplace=True) # for _, row in df.iterrows(): # # Fetch or create group # group = self.db.query(TblGroup).filter_by(group_code=row["group_code"]).first() # if not group: # group = TblGroup( # group_code=row["group_code"], # sim_code=row["simulation_code"], # group_name=row["group_name"], # group_email=row["group_email"], # number_of_members=row["number_of_members"], # category_assigned=row.get("category_assigned") # ) # self.db.add(group) # self.db.commit() # self.db.refresh(group) # else: # # Update existing group if data has changed # updated = False # if row.get("group_name") and group.group_name != row["group_name"]: # group.group_name = row["group_name"] # updated = True # if row.get("group_email") and group.group_email != row["group_email"]: # group.group_email = row["group_email"] # updated = True # if row.get("number_of_members") and group.number_of_members != row["number_of_members"]: # group.number_of_members = row["number_of_members"] # updated = True # if row.get("category_assigned") and group.category_assigned != row["category_assigned"]: # group.category_assigned = row["category_assigned"] # updated = True # if row.get("simulation_code") and group.sim_code != row["simulation_code"]: # group.sim_code = row["simulation_code"] # updated = True # if updated: # self.db.commit() # # Add student if not already exists # student_exists = self.db.query(TblStudent).filter_by(student_email=row["student_email"]).first() # if not student_exists: # password_plain = generate_secure_password() # student = TblStudent( # first_name=row["first_name"], # last_name=row["last_name"], # student_email=row["student_email"], # password=password_plain, # role="student", # group_code=group.group_code # ) # self.db.add(student) # self.db.commit() # return { # "status": "1", # "message": "Upload successful" # } # except Exception as e: # raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")