from datetime import datetime, timedelta import os from typing import List, Literal from fastapi import HTTPException, UploadFile from fastapi.responses import FileResponse from fastapi.security import OAuth2PasswordRequestForm from fastapi_mail import ConnectionConfig, FastMail, MessageSchema from openpyxl import Workbook, load_workbook 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.simulation import TblSimulation 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" ) # EXCEL_DIR = os.getenv("EXCEL_DIR", "download_excel") # EXCEL_FILENAME = "Updated_group_data.xlsx" # EXCEL_PATH = os.path.join(EXCEL_DIR,EXCEL_FILENAME) # EXCEL_DIR = r"download_excel" EXCEL_DIR = r"download_excel" 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 get_simulation_group_student(self, group_code: str): students = ( self.db.query(TblStudent) .filter(TblStudent.group_code == group_code) .all() ) # Don't throw an error for no students if not students: return [] return [ schemas.GetSimulationGroupStudent.model_validate(s) for s in students ] # async def get_simulation_group_student(self, group_code:str): # get_data = self.db.query(TblStudent).filter(TblStudent.group_code == group_code).all() # if not get_data: # raise HTTPException(status_code=404, detail="Group Code Not Found") # return [schemas.GetSimulationGroupStudent.model_validate(data) for data in get_data] async def update_student(self, request: List[schemas.StudentUpdate]): for req in request: updated_student = StudentBase.model_validate(req.model_dump()) 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 download_excel(self, simulation_code: str): """ Download an Excel file from the download_excel folder. """ # Ensure the filename ends with .xlsx if not simulation_code.lower().endswith(".xlsx"): simulation_code += ".xlsx" filepath = os.path.join(EXCEL_DIR, simulation_code) if not os.path.exists(filepath): raise HTTPException(status_code=404, detail=f"Excel file '{simulation_code}' not found.") return FileResponse( filepath, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", filename=simulation_code ) async def upload_students_from_excel(self, file: UploadFile): try: import numpy as np import pandas as pd # --------------------------------------------------------- # 1. READ EXCEL # --------------------------------------------------------- file.file.seek(0) df = pd.read_excel(file.file, engine="openpyxl", header=None) expected_headers = [ "first_name", "last_name", "student_email", "roll_no", "group_code", "simulation_code", "group_name", "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: df = df.replace({np.nan: 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(" ", "_") def clean_value(x): if x is None or (isinstance(x, float) and pd.isna(x)): return None if isinstance(x, str) and x.strip().lower() in ["nan", "none", "null", ""]: return None return x df = df.applymap(clean_value) df = df.where(pd.notna(df), None) def clean_group_code(x): x = clean_value(x) if isinstance(x, float): return str(int(x)) return x if "group_code" in df.columns: df["group_code"] = df["group_code"].apply(clean_group_code) def slugify(name: str): return name.strip().upper().replace(" ", "-") for index, row in df.iterrows(): if row["group_code"] is None: group_name = row.get("group_name") sim_code = row.get("simulation_code") if not group_name or not sim_code: continue group_slug = slugify(group_name) existing_group = ( self.db.query(TblGroup) .filter( TblGroup.group_name == group_name, TblGroup.simulation_code == sim_code ) .first() ) if existing_group: df.at[index, "group_code"] = existing_group.group_code continue prefix = group_slug existing_codes = ( self.db.query(TblGroup) .filter( TblGroup.group_code.like(f"{prefix}-%"), TblGroup.simulation_code == sim_code ) .all() ) nums = [] for g in existing_codes: try: nums.append(int(g.group_code.split("-")[-1])) except: pass next_num = (max(nums) + 1) if nums else 1 new_group_code = f"{prefix}-{next_num:03d}" df.at[index, "group_code"] = new_group_code group_student_map = {} for _, row in df.iterrows(): gc = row["group_code"] if gc: group_student_map.setdefault(gc, []).append(row) error_messages = [] warning_messages = [] for group_code, students in group_student_map.items(): first = {k: clean_value(v) for k, v in students[0].items()} student_count = len(students) sim_code = first.get("simulation_code") group_name = first.get("group_name") simulation = ( self.db.query(TblSimulation) .filter(TblSimulation.simulation_code == sim_code) .first() ) if not simulation: simulation = TblSimulation( simulation_code=sim_code, simulation_name=f"Auto Created {sim_code}" ) self.db.add(simulation) self.db.commit() group = ( self.db.query(TblGroup) .filter(TblGroup.group_code == group_code) .first() ) if not group: num_members = ( int(first["number_of_members"]) if first["number_of_members"] else student_count ) group = TblGroup( group_code=group_code, simulation_code=sim_code, group_name=group_name, number_of_members=num_members, category_assigned=first.get("category_assigned"), ) self.db.add(group) self.db.commit() else: for field in ["group_name", "simulation_code", "category_assigned", "number_of_members"]: if first.get(field) is not None: setattr(group, field, first[field]) self.db.commit() for row in students: row = {k: clean_value(v) for k, v in row.items()} existing = ( self.db.query(TblStudent) .filter_by(student_email=row["student_email"]) .first() ) if existing: existing.first_name = row["first_name"] existing.last_name = row["last_name"] existing.roll_no = row["roll_no"] self.db.commit() continue password_plain = generate_secure_password() student = TblStudent( first_name=row["first_name"], last_name=row["last_name"], student_email=row["student_email"], roll_no=row["roll_no"], password=password_plain, role="student", group_code=group_code ) self.db.add(student) self.db.commit() return { "status": "1", "message": "Upload successful", "warnings": warning_messages } except Exception as e: return {"status": "0", "error": 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 get_students_by_group(self, group_code: str): students = ( self.db.query(TblStudent) .filter(TblStudent.group_code == group_code) .all() ) return students async def send_group_mail(self, group_code: str): students = await self.get_students_by_group(group_code) if not students: raise HTTPException(status_code=404, detail="No students found for this group") conf = ConnectionConfig( MAIL_USERNAME="info@compunet.solutions", MAIL_PASSWORD="Jn193518f", MAIL_FROM="info@compunet.solutions", MAIL_SERVER="smtp.stackmail.com", MAIL_PORT=465, # SSL PORT MAIL_SSL_TLS=True, # SSL ON MAIL_STARTTLS=False, # STARTTLS OFF USE_CREDENTIALS=True, VALIDATE_CERTS=True ) # conf = ConnectionConfig( # MAIL_USERNAME="info@compunet.solutions", # MAIL_PASSWORD="Jn193518f", # MAIL_FROM="info@compunet.solutions", # MAIL_PORT=587, # MAIL_SERVER="smtp.stackmail.com", # MAIL_STARTTLS=True, # MAIL_SSL_TLS=False, # USE_CREDENTIALS=True # ) fm = FastMail(conf) login_url = "https://trilliant.compunet.in/retail/" for s in students: subject = "Your Retail Simulation Login Details" body = f""" Hello {s.first_name} {s.last_name}, Your account for the Retail Simulation Platform has been created. Here are your login details: First Name : {s.first_name} Last Name : {s.last_name} Email : {s.student_email} Password : {s.password} Group Code : {s.group_code} Login using the link below: {login_url} Please change your password after first login. Regards, College Professor / Admin """ message = MessageSchema( subject=subject, recipients=[s.student_email], body=body, subtype="plain" ) await fm.send_message(message) return { "status": "success", "message": "Emails sent to all students in this group" }