from datetime import datetime import shutil from fastapi import HTTPException, UploadFile from sqlalchemy.orm import Session import app.api.simulation.schemas as schemas from app.api.student.service import StudentService from app.dependency.authantication import JWTPayloadSchema from app.locale.messages import Messages from app.models.main.category_simulation import TblCategorySimulation from app.models.main.group import TblGroup from app.models.main.simulation import SimulationBase, TblSimulation from app.models.main.user import TblUser from app.models.main.user_simulation import TblUserSimulation from app.utils.schemas_utils import CustomResponse class SimulationService: def __init__(self, db: Session, token: JWTPayloadSchema): self.db = db self.token = token def generate_simulation_code(self): last_simulation = ( self.db.query(TblSimulation) .order_by(TblSimulation.simulation_id.desc()) .first() ) if not last_simulation or not last_simulation.simulation_code: return "SIM-001" last_code = last_simulation.simulation_code parts = last_code.split("-") if len(parts) != 2 or not parts[1].isdigit(): return "SIM-001" current_number = int(parts[1]) + 1 return f"SIM-{current_number:03d}" async def create_simulation(self, request: schemas.SimulationCreate): simulation_code = self.generate_simulation_code() data = request.model_dump() data["simulation_code"] = simulation_code created_simulation = SimulationBase.model_validate(data) new_simulation = TblSimulation(**created_simulation.model_dump()) self.db.add(new_simulation) self.db.commit() self.db.refresh(new_simulation) from app.api.user.service import UserService user_service = UserService(self.db, self.token) await user_service.admin_link( admin_id=request.admin_id, simulation_id=new_simulation.simulation_id ) user = self.db.query(TblUser).order_by(TblUser.user_id.desc()).first() new_link = TblUserSimulation( simulation_id=new_simulation.simulation_id, user_id=user.user_id, ) self.db.add(new_link) self.db.commit() self.db.refresh(new_link) return CustomResponse( status="1", message=Messages.SIMULATION_CREAT ) async def get_simulation(self, simulation_id: int): simulation = TblSimulation.get_by_id(simulation_id, self.db) return schemas.SimulationResponse.model_validate(simulation) async def get_groups_by_simulation_code(self, simulation_code: str): groups = ( self.db.query(TblGroup) .filter(TblGroup.simulation_code == simulation_code) .all() ) if not groups: return {"group": []} student_service = StudentService(self.db, self.token) result = [] for g in groups: student = await student_service.get_simulation_group_student(g.group_code) result.append({ "group_id": g.group_id, "group_name": g.group_name, "group_email": g.group_email, "category_assigned": g.category_assigned, "category": g.category, "number_of_members": g.number_of_members, "simulation_code": g.simulation_code, "group_code": g.group_code, "student": student }) return {"group": result} # async def get_groups_by_simulation_id(self, simulation_code: str): # group = self.db.query(TblGroup).filter(TblGroup.simulation_code == simulation_code).all() # if not group: # raise HTTPException(status_code=404, detail="Group not found") # return [schemas.GroupResponse.model_validate(get_data.__dict__) for get_data in group] async def get_groups_by_simulation_id(self, simulation_code: int): groups = ( self.db.query(TblGroup) .filter(TblGroup.simulation_code == simulation_code) .all() ) return [ schemas.GroupResponse( group_id=g.group_id, group_name=g.group_name, group_email=g.group_email, category_assigned=g.category_assigned, category=g.category, number_of_members=g.number_of_members, simulation_code=g.simulation_code, group_code=g.group_code ) for g in groups ] async def update_end_date(self, simulation_id:int, end_date:datetime): end_dates = self.db.query(TblSimulation).filter(TblSimulation.simulation_id == simulation_id).first() if not end_dates: return CustomResponse(status="-1", message="Simulation not found") end_dates.end_date = end_date self.db.commit() return CustomResponse(status="1", message="End Date Updated Successfully") async def run_simulation(self, simulation_id:int): simulation = self.db.query(TblSimulation).filter(TblSimulation.simulation_id == simulation_id).first() if not simulation: return CustomResponse(status="-1", message="Simulation not found") if simulation.status == "Pending": simulation.status = "Active" self.db.commit() self.db.refresh(simulation) return CustomResponse(status="1", message="The Simulation is Active") elif simulation.status == "Active": return CustomResponse(status="-1", message="The Simulation is Already Active") elif simulation.status == "Completed": return CustomResponse(status="-1", message="The Simulation is Already Complated") else: return CustomResponse(status="-1", message=f"Unknown status: {simulation.status}") async def end_simulation(self, simulation_id:int): simulation = self.db.query(TblSimulation).filter(TblSimulation.simulation_id == simulation_id).first() if not simulation: return CustomResponse(status="-1", message="Simulation not found") if simulation.status == "Pending": return CustomResponse(status="-1", message="The Simulation is Pending") elif simulation.status == "Active": simulation.status = "Completed" self.db.commit() self.db.refresh(simulation) return CustomResponse(status="1", message="The Simulation is Complated") elif simulation.status == "Completed": return CustomResponse(status="-1", message="The Simulation is Already Complated") else: return CustomResponse(status="-1", message=f"Unknown status: {simulation.status}") async def delete_simulation(self, simulation_id: int): deleted = TblSimulation.delete(simulation_id, self.db) if not deleted: return CustomResponse(status="-1", message=Messages.SIMULATION_NOT_FOUND) return CustomResponse(status="1", message=Messages.SIMULATION_DELETED)