import datetime import os from typing import Annotated, Dict, List, Optional from uuid import uuid4 from fastapi import ( APIRouter, Body, Depends, File, Header, HTTPException, Request, UploadFile, status, ) from fastapi.responses import JSONResponse from models.team import CreateUserRequest, Owner, Team, Theme from models.user import AccountType, UserBase from services.auth_service import create_user_cognito_pool, validate_jwt_token from services.team_services import TeamService from services.user_service import UserService from utils.auth import generate_secure_password from utils.client_check import ClientConfig from utils.s3_storage import S3BucketStorage team = APIRouter(prefix="/teams", tags=["teams"]) async def create_team_class( name: str = Body(...), company_name: str = Body(...), created_by: str = Body(...), company_url: Optional[str] = Body(None), is_franchise: bool = Body(False), description: Optional[str] = Body(None), logo_url: Optional[str] = Body(None), theme: Optional[Theme] = Body(None), owners: Optional[List[Owner]] = Body(None), team_id: Optional[str] = Body(None), ): """Create a team class with the provided parameters""" client_config = ClientConfig("default") default_config = client_config.get_default_dict() team = Team( id=team_id, name=name, company_name=company_name, company_url=company_url, is_franchise=is_franchise, description=description, logo_url=logo_url, theme=theme, owners=owners if owners else [], created_by=created_by, members=[], client_config=default_config, ) return team TeamDependency = Annotated[Team, Depends(create_team_class)] @team.post("/", response_model=Team) async def create_team( team: TeamDependency, request: Request, ): """Create a new team or update existing team with the provided information""" try: current_user = request.state.current_user team_service = TeamService() if team.id: # Update existing team existing_team = team_service.get_team_by_id(str(team.id)) if not existing_team: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Team not found" ) # Update existing team with new team data existing_team.name = team.name existing_team.company_name = team.company_name existing_team.company_url = team.company_url existing_team.is_franchise = team.is_franchise existing_team.description = team.description existing_team.logo_url = team.logo_url existing_team.theme = team.theme existing_team.owners = team.owners existing_team.client_config = team.client_config existing_team.add_member(user_id=current_user.user_id) existing_team.save_to_db() return existing_team else: team.id = str(uuid4()) print(team) user_service = UserService() user_service.update_user_team(user_id=current_user.email, team_id=team.id) team.add_member(user_id=current_user.email) team.save_to_db() return team except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to create team: {str(e)}") @team.post("/logo/upload") async def upload_team_logo(file: UploadFile = File(...)): """Upload a team logo before team creation""" 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"]: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Only jpg, jpeg, png, gif, and svg files are allowed", ) # Generate temporary path for the logo s3_file_path = f"team-logos/{uuid4()}.{file_ext}" xcap_s3_storage = os.environ["xcap_s3_storage"] try: 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 ) except Exception as s3_error: if "ExpiredToken" in str(s3_error): # You might want to refresh credentials here if you have a mechanism raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="S3 credentials have expired. Please try again.", ) raise # Re-raise other S3-related errors s3_url = s3_storage.s3_file_location return JSONResponse( status_code=200, content={"message": "Logo uploaded successfully", "logo_url": s3_url}, ) 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 upload logo: {str(e)}", ) @team.get("/{team_id}", response_model=Team) async def get_team_details(team_id: str): """Get team details by team ID""" try: team_service = TeamService() team = team_service.get_team_by_id(team_id) if not team: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Team not found" ) return team 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 get team details: {str(e)}", ) @team.post("/{team_id}/member", status_code=status.HTTP_201_CREATED) async def create_user( user_data: CreateUserRequest, team_id: str, request: Request, ): """Create a new user in both Cognito and the user table, and add them to the team""" try: current_user = request.state.current_user temp_password = generate_secure_password() cognito_response = await create_user_cognito_pool(user_data, temp_password) user_service = UserService() team_service = TeamService() new_user = UserBase( user_id=cognito_response.get("email"), username=cognito_response.get("username"), email=cognito_response.get("email"), first_name=cognito_response.get("first_name"), last_name=cognito_response.get("last_name"), created_by=current_user.user_id, account_type=user_data.role, created_at=datetime.datetime.utcnow(), team_id=team_id, ) user_service.save_user(new_user) team = team_service.get_team_by_id(team_id) if not team: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Team not found" ) team.add_member(user_id=new_user.email) team.save_to_db() return JSONResponse( status_code=status.HTTP_201_CREATED, content={ "message": "User created successfully and added to team", }, ) 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 create user: {str(e)}", ) @team.get("/{team_id}/members", response_model=List[UserBase]) async def get_team_members( team_id: str, ): """Get all members of a team by team ID""" try: # First verify the team exists team_service = TeamService() team = team_service.get_team_by_id(team_id) if not team: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Team not found" ) user_service = UserService() member_ids = list(team.members) team_members = user_service.get_active_users_by_ids(member_ids) return team_members 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 get team members: {str(e)}", )