from datetime import datetime, timedelta import random from typing import Literal, Optional from fastapi import BackgroundTasks, HTTPException from fastapi.security import OAuth2PasswordRequestForm from fastapi_mail import ConnectionConfig, FastMail, MessageSchema from sqlalchemy.orm import Session from app.api.simulation.schemas import SimulationResponse import app.api.user.schemas as schemas from app.dependency.authantication import JWTManager, JWTPayloadSchema from app.locale.messages import Messages from app.models.main.institution import TblInstitution from app.models.main.simulation import TblSimulation from app.models.main.user import UserBase, TblUser from app.utils.email import RESET_OTP_EXPIRE_MINUTES, admin_account_created, send_email, send_email_async, send_otp_email from app.utils.schemas_utils import CustomResponse from app.models.main.user_simulation import TblUserSimulation from app.api.simulation import service from app.utils import email VALID_ROLES: tuple[Literal["student", "admin", "superadmin"], ...] = ("student", "admin", "superadmin") class UserService: def __init__(self, db: Session, token:JWTPayloadSchema): self.db = db self.token = token async def create_user(self, request: schemas.UserCreate): created_user = UserBase.model_validate(request) TblUser.create(created_user, self.db) self.db.commit() return CustomResponse(status="1", message=Messages.USER_CREAT) @staticmethod async def user_login(credentials: OAuth2PasswordRequestForm, db: Session): user_filter = {"email_id": credentials.username} user_log = TblUser.get_by_user_filter(user_filter, db) if not user_log: raise HTTPException(status_code=404, detail=Messages.USER_NOT_FOUND) if credentials.password != user_log.password: raise HTTPException(status_code=401, detail=Messages.INCORRECT_PASSWORD) role = user_log.role.strip() if role not in VALID_ROLES: raise HTTPException(status_code=400, detail=f"Invalid role: {role}") payload = JWTPayloadSchema( user_id=user_log.user_id, user_type="admin", # or determine dynamically if needed user_role=role, exp=datetime.utcnow() ) access_token = JWTManager.create_access_token(payload) return { "access_token": access_token, "token_type": "bearer", "email_id": user_log.email_id, "role": role, "user_id":user_log.user_id } async def get_demo_user(self): if self.token.user_id is None: raise HTTPException(status_code=400, detail=Messages.USER_ID_MISSING) user = TblUser.get_usr_id(self.token.user_id, self.db) if not user: return CustomResponse(status="-1", message=Messages.USER_NOT_FOUND) return user async def get_admin(self, user_id:int): get_data = self.db.query(TblUser).filter(TblUser.user_id == user_id).first() return get_data async def logout(self): if self.token.user_id is None: raise HTTPException(status_code=400, detail=Messages.USER_ID_MISSING) print(f"User ID during logout: {self.token.user_id}") user = self.db.query(TblUser).filter(TblUser.user_id == self.token.user_id).first() if not user: return CustomResponse(status="-1", message=Messages.USER_NOT_FOUND) self.db.commit() return CustomResponse(status="1", message=Messages.USER_LOGOUT) async def simulation(self, status: str): get_data = ( self.db.query(TblUser.role) .filter(TblUser.user_id == self.token.user_id) .first() ) if get_data is None: raise HTTPException(status_code=400, detail=Messages.USER_ID_MISSING) role = get_data[0] simulation_service = service.SimulationService(self.db, self.token) if role == "superadmin": if status in ["Active", "Pending"]: simulations = ( self.db.query(TblSimulation) .filter(TblSimulation.status.in_(["Active", "Pending"])) .all() ) else: simulations = ( self.db.query(TblSimulation) .filter(TblSimulation.status == status) .all() ) elif role == "admin": if status in ["Active", "Pending"]: simulations = ( self.db.query(TblSimulation) .join( TblUserSimulation, TblUserSimulation.simulation_id == TblSimulation.simulation_id, ) .filter( TblUserSimulation.user_id == self.token.user_id, TblSimulation.status.in_(["Active", "Pending"]), ) .all() ) else: simulations = ( self.db.query(TblSimulation) .join( TblUserSimulation, TblUserSimulation.simulation_id == TblSimulation.simulation_id, ) .filter( TblUserSimulation.user_id == self.token.user_id, TblSimulation.status == status, ) .all() ) else: simulations = [] result = [] for sim in simulations: sim_obj = await simulation_service.get_simulation(sim.simulation_id) sim_groups = await simulation_service.get_groups_by_simulation_id(sim.simulation_code) result.append({ "simulationId": sim_obj.simulation_id, "simulationName": sim_obj.simulation_name, "description": sim_obj.description, "status": sim_obj.status, "startDate": sim_obj.start_date, "endDate": sim_obj.end_date, "institution": sim_obj.institution, "location": sim_obj.location, "simulationCode": sim_obj.simulation_code, "groups": sim_groups or [] }) return { "role": role, "simulations": result } async def admin_link(self, admin_id: int, simulation_id: int, request: Optional[schemas.UserCreateLink] = None): if admin_id == 0: if not request: raise HTTPException(status_code=400, detail="Admin data is required when admin_id is 0") password = email.generate_random_password() created_user = UserBase.model_validate(request) created_user.password = password existing_user = ( self.db.query(TblUser) .filter(TblUser.email_id == request.email_id) .first() ) if existing_user: raise HTTPException( status_code=400, detail="Admin exists with another simulation" ) new_user = TblUser.create(created_user, self.db) self.db.commit() self.db.refresh(new_user) admin_id = new_user.user_id login_url = "http://206.72.206.188/retail/" subject, body = admin_account_created( name=new_user.first_name, email=new_user.email_id, password=password, login_url=login_url ) await send_email_async( to=new_user.email_id, subject=subject, body=body ) return CustomResponse(status="1", message=Messages.USER_CREAT) # async def admin_link(self, admin_id: int, simulation_id:int, request: Optional[schemas.UserCreateLink] = None): # if admin_id == 0: # if not request: # return HTTPException(status_code=400, detail="Admin data is required when admin_id is 0") # password = email.generate_random_password() # created_user = UserBase.model_validate(request) # created_user.password = password # # 1️⃣ Check duplicate email # existing_user = ( # self.db.query(TblUser) # .filter(TblUser.email_id == request.email_id) # .first() # ) # if existing_user: # raise HTTPException( # status_code=400, # detail="Admin exists with another simulation" # ) # new_user = TblUser.create(created_user, self.db) # self.db.commit() # self.db.refresh(new_user) # admin_id = new_user.user_id # 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 = "http://206.72.206.188/retail/" # subject = "Your New Account Details" # body = f""" # Hello {new_user.first_name}, # Your account has been created successfully. # Email: {new_user.email_id} # Password: {password} # Login using the link below: # {login_url} # Please log in and change your password immediately. # """ # message = MessageSchema( # subject=subject, # recipients=[new_user.email_id], # body=body, # subtype="plain" # ) # await fm.send_message(message) # return CustomResponse(status="1", message=Messages.USER_CREAT) async def update_pass(self, request: schemas.UserPassUpdate): user_id = self.token.user_id if not user_id: return CustomResponse(status="-1", message=Messages.USER_ID_MISSING) updated_user = schemas.UserPassUpdate.model_validate(request) user = TblUser.get_usr_id(user_id, self.db) if not user: return CustomResponse(status="-1", message=Messages.USER_NOT_FOUND) if updated_user.current_password != user.password: return CustomResponse(status="-1", message=Messages.CURRENT_PASSORD_INCORRECT) if updated_user.new_password != updated_user.confirm_password: return CustomResponse(status="-1", message=Messages.NEWPASSWORD_CONFIRMPASSWORD_NOT_MATCH) user.password = updated_user.new_password self.db.commit() return CustomResponse(status="1", message=Messages.PASSWORD_UPDATED) async def get_all_admin(self): users = self.db.query(TblUser).all() result = [] for user in users: institutions = ( self.db.query(TblInstitution) .filter(TblInstitution.admin_id == user.user_id) .all() ) institutions_response = [ { "institution_id": inst.institution_id, "institutions_name": inst.institutions_name, "members_count": inst.members_count, "simulation_count": inst.simulation_count } for inst in institutions ] result.append({ "user_id": user.user_id, "first_name": user.first_name, "last_name": user.last_name, "email_id": user.email_id, "mobile_number": user.mobile_number, "institutions": institutions_response }) return result async def get_admin_institution(self, user_id: int): users = ( self.db.query(TblUser) .filter(TblUser.user_id == user_id) .first() ) if not users: raise HTTPException(status_code=404, detail="User not found") institutions = ( self.db.query(TblInstitution) .filter(TblInstitution.admin_id == users.user_id) .all() ) simulation_rows = ( self.db.query(TblUserSimulation) .filter(TblUserSimulation.user_id == users.user_id) .all() ) simulation_ids = [row.simulation_id for row in simulation_rows] simulations = ( self.db.query(TblSimulation) .filter(TblSimulation.simulation_id.in_(simulation_ids)) .all() ) total_members = sum(sim.members for sim in simulations) institutions_response = [ { "institution_id": inst.institution_id, "institutions_name": inst.institutions_name, "members_count": inst.members_count, "simulation_count": inst.simulation_count } for inst in institutions ] simulations_response = [ { "simulation_id": sim.simulation_id, "simulation_name": sim.simulation_name, "description": sim.description, "status": sim.status, "start_date": sim.start_date, "end_date": sim.end_date, "institution": sim.institution, "location": sim.location, "simulation_code": sim.simulation_code, "members": sim.members } for sim in simulations ] return [{ "user_id": users.user_id, "first_name": users.first_name, "last_name": users.last_name, "email_id": users.email_id, "mobile_number": users.mobile_number, "simulation_count": len(simulations), "members_count": total_members, "institutions": institutions_response, "simulations": simulations_response }] def reset_password_service(db:Session, request: schemas.VerifyOTPRequest): """Verifies OTP and allows password reset.""" user = db.query(TblUser).filter(TblUser.email_id == request.email_id).first() if not user: raise HTTPException(status_code=404, detail=Messages.USER_NOT_FOUND) if ( user.otp != request.otp or user.otp_expiry is None or user.otp_expiry < datetime.utcnow() ): raise HTTPException(status_code=400, detail=Messages.INVALID_OTP) user.otp_expiry = None db.add(user) db.commit() return CustomResponse(status="1", message=Messages.OTP_IS_CORRECT) def new_password_service(db:Session, request: schemas.NewPass): """Verifies OTP and allows password reset.""" if TblUser.password == request.old_password: user = db.query(TblUser).filter(TblUser.email_id == request.email_id).first() user.password = request.new_password db.add(user) db.commit() else: user = db.query(TblUser).filter(TblUser.email_id == request.email_id).first() user.password = request.new_password db.add(user) db.commit() return CustomResponse(status="1", message=Messages.PASSWORD_UPDATED) # 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 # ) 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 ) async def forgot_password_service( db: Session, request: schemas.ForgotPasswordRequest, background_tasks: BackgroundTasks ): user = db.query(TblUser).filter(TblUser.email_id == request.email_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") otp = str(random.randint(100000, 999999)) user.otp = otp user.otp_expiry = datetime.utcnow() + timedelta(minutes=10) db.add(user) db.commit() subject = "Password Reset OTP - Retail Simulation" body = f""" Hello {user.first_name}, We received a request to reset your password. Your OTP is: {otp} This OTP is valid for 10 minutes. If you did not request a password reset, please ignore this email. """ message = MessageSchema( subject=subject, recipients=[user.email_id], body=body, subtype="plain" ) fm = FastMail(conf) background_tasks.add_task(fm.send_message, message) return CustomResponse(status="1", message="OTP sent successfully")