import base64 # from utils.logger import ServiceLogger import logging import os from datetime import datetime import boto3 from fastapi import (APIRouter, Body, File, Header, HTTPException, Query, UploadFile, status) from fastapi.responses import JSONResponse from models.user import AccountType, UserBase from services.auth_service import validate_jwt_token from services.user_service import UserService from utils.s3_storage import S3BucketStorage from . import utilities XCM_logger = logging.getLogger() user = APIRouter(prefix="/users", tags=["Users"]) user_service = UserService() @user.get("/{user_id}", response_model=UserBase) async def get_user(user_id: str): """Get user details for the given user_id""" user_data = user_service.get_user_by_id(user_id) if not user_data: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with ID {user_id} not found", ) return user_data @user.put("/{user_id}", response_model=UserBase) async def update_user(user_id: str, user_data: UserBase): """Update user details for the given user_id""" # Validate user_id matches if user_id != user_data.user_id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="User ID in path does not match user ID in request body", ) # Check if user exists existing_user = user_service.db.get_item(user_service.db.user, user_id) if not existing_user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with ID {user_id} not found", ) try: # Update user in DynamoDB user_data.update_modify() # Update the modified timestamp XCM_logger.info(f"Updating user {user_data.user_id} in DynamoDB") user_service.db.upload_to_dynamodb(user_service.db.user, user_data.model_dump()) XCM_logger.info(f"Successfully updated user {user_data.user_id}") return user_data except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update user: {str(e)}", ) @user.patch("/{user_id}/account-type") async def update_account_type(user_id: str, account_type: AccountType): """Update user's account type""" user_data = user_service.db.get_item(user_service.db.user, user_id) if not user_data: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with ID {user_id} not found", ) try: user = UserBase(**user_data) user.update_account_type(account_type) user_service.db.upload_to_dynamodb(user_service.db.user, user.model_dump()) return {"message": "Account type updated successfully"} except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update account type: {str(e)}", ) @user.patch("/{user_id}/status") async def update_user_status(user_id: str, is_active: bool): """Activate or deactivate user account""" user_data = user_service.db.get_item(user_service.db.user, user_id) if not user_data: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with ID {user_id} not found", ) try: user = UserBase(**user_data) if is_active: user.activate() else: user.deactivate() user_service.db.upload_to_dynamodb(user_service.db.user, user.model_dump()) return { "message": f"User {'activated' if is_active else 'deactivated'} successfully" } except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to update user status: {str(e)}", ) @user.post("/{user_id}/profile-picture") async def upload_profile_picture(user_id: str, file: UploadFile = File(...)): """Upload user profile picture""" user_data = user_service.db.get_item(user_service.db.user, user_id) if not user_data: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"User with ID {user_id} not found", ) try: if not file.content_type.startswith("image/"): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="File must be an image" ) file_ext = file.filename.split(".")[-1].lower() if file_ext not in ["jpg", "jpeg", "png", "gif", "svg", "webp"]: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Only jpg, jpeg, png, webp, svg, and gif files are allowed", ) s3_file_path = f"profile-pictures/{user_id}/{datetime.now().strftime('%Y%m%d_%H%M%S')}.{file_ext}" xcap_s3_storage = os.environ["xcap_s3_storage"] s3_storage = S3BucketStorage(bucket_name=xcap_s3_storage) # Read file content file_content = await file.read() # Upload to S3 s3_storage.upload_file_to_s3_bucket( s3_file_path=s3_file_path, s3_file=file_content ) s3_url = s3_storage.s3_file_location # Get the image blob from S3 XCM_logger.info(f"Successfully uploaded profile picture for user {user_id}") user = UserBase(**user_data) user.update_profile(profile_pic=s3_url) user_service.db.upload_to_dynamodb(user_service.db.user, user.model_dump()) return JSONResponse( status_code=200, content={ "message": "Profile picture uploaded successfully", "profile_pic_url": s3_url, "image_blob": base64.b64encode(file_content).decode( "utf-8" ), # Convert bytes to base64 string }, ) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to upload profile picture: {str(e)}", ) @user.delete("/{user_id}") async def delete_user( user_id: str, deactivate_cognito: bool = Query( False, description="Whether to deactivate the user in Cognito as well" ), ): """Delete/deactivate a user""" try: # Get the user to be deleted user_service = UserService() user = user_service.get_user_by_id(user_id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Deactivate user in our database user.deactivate() user.save_to_db() # If requested, deactivate in Cognito as well if deactivate_cognito: try: cognito_client = boto3.client("cognito-idp") cognito_client.admin_disable_user( UserPoolId=os.environ["COGNITO_USER_POOL_ID"], Username=user.email ) except Exception as e: # Log the error but don't fail the whole request XCM_logger.error(f"Failed to deactivate user in Cognito: {str(e)}") return JSONResponse( status_code=status.HTTP_200_OK, content={ "message": "User successfully deactivated", "user_id": user_id, "cognito_deactivated": deactivate_cognito, }, ) except HTTPException as http_ex: raise http_ex except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to delete user: {str(e)}", )