from typing import List from fastapi import HTTPException from sqlalchemy.orm import Session import app.api.group.schemas as schemas from app.api.student.service import StudentService from app.locale.messages import Messages from app.models.main.category import CategoryCreate, TblCategory from app.models.main.group import GroupBase,TblGroup from app.models.main.main_data import TblMainData from app.models.main.student import TblStudent from app.utils.schemas_utils import CustomResponse class GroupService: def __init__(self, db: Session, token): self.db = db self.token = token async def create_group(self, request: schemas.GroupCreate): created_user = GroupBase.model_validate(request.model_dump()) TblGroup.create(created_user, self.db) self.db.commit() return CustomResponse(status="1", message= Messages.GROUP_CREAT) async def get_group(self, group_id: int): group = TblGroup.get_group(group_id, self.db) if not group: raise HTTPException(status_code=404,detail=Messages.GROUP_NOT_FOUND) return schemas.GroupRespose.model_validate(group) async def get_group_category(self, group_id: int): result = TblGroup.get_group_category(group_id, self.db) if not result: raise HTTPException(status_code=404, detail=Messages.GROUP_NOT_FOUND) return schemas.GroupCategoryResponse(category_assigned=result[0], category=result[1]) async def get_simulation_group(self, group_code:str): get_data = self.db.query(TblGroup).filter(TblGroup.group_code == group_code).first() if not get_data: raise HTTPException(status_code=404,detail="Simulation Code Not Found") student_service = StudentService(self.db, self.token) result = [] student = await student_service.get_simulation_group_student(get_data.group_code) result.append({ "group_id": get_data.group_id, "group_name": get_data.group_name, # "group_email": get_data.group_email, "category_assigned": get_data.category_assigned, "category": get_data.category, "number_of_members": get_data.number_of_members, "simulation_code": get_data.simulation_code, "group_code": get_data.group_code, "student": student }) return {"group": result} # return [schemas.GetSimulationGroup.model_validate(data) for data in get_data] async def update_group(self, request: List[schemas.GroupUpdate]): for req in request: group = GroupBase.model_validate(req.model_dump()) if group.group_id is None: raise HTTPException(status_code=404, detail=Messages.GROUP_NOT_FOUND) updated_group = TblGroup.update(group.group_id, group, self.db) if not updated_group: raise HTTPException(status_code=404,detail=Messages.GROUP_NOT_FOUND) self.db.commit() return CustomResponse(status="1", message=Messages.GROUP_UPDATE) async def get_group_students(self, group_code: str): students = self.db.query(TblStudent).filter(TblStudent.group_code == group_code).all() return [schemas.StudentInfo.model_validate(student) for student in students] # def create_category(self, request:CategoryCreate): # group = TblGroup.get_group(request.group_id, self.db) # if not group: # raise HTTPException(status_code=404,detail=Messages.GROUP_NOT_FOUND) # if group.category_assigned: # return CustomResponse(status="0", message="Category already assigned by admin.") # TblCategory.create(CategoryCreate.model_validate(request.model_dump()), self.db) # group.category_assigned = request.choose_category # self.db.commit() # return CustomResponse(status="1", message=Messages.GROUP_CATEGORY) # def create_category(self, request: CategoryCreate): # group = TblGroup.get_group(request.group_id, self.db) # if not group: # raise HTTPException(status_code=404, detail=Messages.GROUP_NOT_FOUND) # sim_code = group.sim_code # existing_category = self.db.query(TblGroup).filter( # TblGroup.sim_code == sim_code, # TblGroup.category_assigned == request.choose_category # ).first() # if existing_category: # return CustomResponse(status="0", message="This category is already assigned to another group in the same simulation.") # if group.category_assigned: # return CustomResponse(status="0", message="Category already assigned to this group.") # TblCategory.create(CategoryCreate.model_validate(request.model_dump()), self.db) # group.category_assigned = request.choose_category # self.db.commit() # return CustomResponse(status="1", message=Messages.GROUP_CATEGORY) def create_category(self, request: CategoryCreate): group = TblGroup.get_group(request.group_id, self.db) if not group: raise HTTPException(status_code=404, detail=Messages.GROUP_NOT_FOUND) if group.category_assigned: return CustomResponse(status="1", message="Category already assigned by admin.") simulation_code = group.simulation_code existing_category = self.db.query(TblGroup).filter( TblGroup.simulation_code == simulation_code, TblGroup.category_assigned == request.choose_category ).first() if existing_category: return CustomResponse(status="1", message="This category is already assigned to another group in the same simulation.") if group.category_assigned: return CustomResponse(status="1", message="Category already assigned to this group.") TblCategory.create(CategoryCreate.model_validate(request.model_dump()), self.db) group.category_assigned = request.choose_category self.db.commit() return CustomResponse(status="1", message=Messages.GROUP_CATEGORY) def get_main_data_by_group_service(self,group_id: int): group = self.db.query(TblGroup).filter(TblGroup.group_id == group_id).first() if not group: raise HTTPException(status_code=404, detail="Group not found") category = group.category_assigned or group.category if not category: raise HTTPException(status_code=400, detail="Group has no category assigned") main_data = self.db.query(TblMainData).filter(TblMainData.format_type == category).first() if not main_data: raise HTTPException(status_code=404, detail="Main data not found for this category") return main_data def get_group_category_service(self, group_id: int) -> schemas.GroupCategoryResponse: group = self.db.query(TblGroup).filter(TblGroup.group_id == group_id).first() if not group: raise HTTPException(status_code=404, detail="Group not found") return schemas.GroupCategoryResponse( category_assigned=group.category_assigned, category=group.category )