from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, EmailStr, constr from datetime import datetime import os import sys import bcrypt import random from typing import Optional, List from twilio.rest import Client from bson import ObjectId import httpx import jwt import logging from datetime import timedelta # Configure logger logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) # Add root directory to Python path sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) from utils.database import DatabaseManager app = FastAPI(title="Auth Service") # Configure CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize database db = DatabaseManager() users_collection = None businesses_collection = None verification_collection = None business_types_collection = None @app.on_event("startup") async def startup_event(): global users_collection, businesses_collection, verification_collection, business_types_collection users_collection = db.get_collection("users") businesses_collection = db.get_collection("businesses") verification_collection = db.get_collection("verifications") business_types_collection = db.get_collection("business_types") # Twilio setup twilio_client = Client( os.getenv("TWILIO_ACCOUNT_SID"), os.getenv("TWILIO_AUTH_TOKEN") ) TWILIO_PHONE = os.getenv("TWILIO_PHONE_NUMBER") # Models class SignupStep1(BaseModel): name: str email: EmailStr phone: str password: str confirm_password: str class PhoneVerification(BaseModel): phone: str otp: str class BusinessDetails(BaseModel): business_name: str business_type: str street_address: str city: str postal_code: str # Business Type Models class BusinessType(BaseModel): name: str class BusinessTypeResponse(BaseModel): id: str name: str # Additional Models class AIAssistantSelection(BaseModel): template_id: str business_id: str user_email: str class ServiceDetails(BaseModel): service_name: str price: str duration: str class ServiceRequest(BaseModel): business_id: str services: List[ServiceDetails] class WorkHours(BaseModel): start: str end: str class Schedule(BaseModel): off_days: List[str] work_hours: WorkHours class FinalStepRequest(BaseModel): schedule: Schedule business_id: str class SignIn(BaseModel): email: str password: str # Step 1: Basic Info and Phone Verification @app.post("/signup/step1") async def signup_step1(user_data: SignupStep1): # Validate password length if len(user_data.password) < 8: raise HTTPException(status_code=400, detail="Password must be at least 8 characters") # Validate passwords match if user_data.password != user_data.confirm_password: raise HTTPException(status_code=400, detail="Passwords do not match") # Check if email exists if users_collection.find_one({"email": user_data.email}): raise HTTPException(status_code=400, detail="Email already registered") # Check if phone exists if users_collection.find_one({"phone": user_data.phone}): raise HTTPException(status_code=400, detail="Phone number already registered") # Generate OTP otp = ''.join([str(random.randint(0, 9)) for _ in range(6)]) logger.info(f"=-=-=-=-=-= OTP: {otp}") # Call SMS service to send OTP async with httpx.AsyncClient() as client: response = await client.post( "http://sms_pm.rms.yt/send-otp", json={ "phone": user_data.phone, "purpose": "signup", "otp": otp # Pass the generated OTP to SMS service } ) #if response.status_code != 200: # raise HTTPException(status_code=500, detail="Failed to send OTP") # Hash password and store initial user data hashed_password = bcrypt.hashpw(user_data.password.encode('utf-8'), bcrypt.gensalt()) # Store user data result = users_collection.insert_one({ "name": user_data.name, "email": user_data.email, "phone": user_data.phone, "password": hashed_password, "signup_step": 1, "created_at": datetime.utcnow(), "status": "pending" }) # Store verification data verification_collection.insert_one({ "phone": user_data.phone, "otp": otp, "purpose": "signup", "verified": False, "created_at": datetime.utcnow(), "expires_at": datetime.utcnow() + timedelta(minutes=10) # OTP expires in 10 minutes }) return {"message": "OTP sent successfully", "phone": user_data.phone} # Step 1b: Verify Phone OTP @app.post("/signup/verify-phone") async def verify_phone(verification: PhoneVerification): try: # Find verification record verify_record = verification_collection.find_one({ "phone": verification.phone, "otp": verification.otp, "verified": False }) if not verify_record: raise HTTPException(status_code=400, detail="Invalid OTP") # Update verification status verification_collection.update_one( {"_id": verify_record["_id"]}, {"$set": {"verified": True}} ) # Update user status users_collection.update_one( {"phone": verification.phone}, {"$set": {"phone_verified": True}} ) return {"message": "Phone verified successfully"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Business Type Management @app.post("/business-types") async def create_business_type(business_type: BusinessType): try: # Check if type already exists if business_types_collection.find_one({"name": business_type.name}): raise HTTPException(status_code=400, detail="Business type already exists") result = business_types_collection.insert_one({ "name": business_type.name, "created_at": datetime.utcnow() }) return { "id": str(result.inserted_id), "name": business_type.name } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/business-types", response_model=List[BusinessTypeResponse]) async def get_business_types(): try: types = [] cursor = business_types_collection.find() for type_doc in cursor: types.append({ "id": str(type_doc["_id"]), "name": type_doc["name"] }) return types except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.put("/business-types/{type_id}") async def update_business_type(type_id: str, business_type: BusinessType): try: result = await business_types_collection.update_one( {"_id": ObjectId(type_id)}, {"$set": {"name": business_type.name}} ) if result.modified_count == 0: raise HTTPException(status_code=404, detail="Business type not found") return {"message": "Business type updated successfully"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Step 2: Business Details @app.post("/signup/step2") async def signup_step2(business_data: BusinessDetails, user_email: str): try: # Verify user exists and has completed step 1 user = users_collection.find_one({ "email": user_email, "phone_verified": True }) if not user: raise HTTPException(status_code=400, detail="Complete step 1 first") # Create business record business_doc = { "name": business_data.business_name, "type": business_data.business_type, "address": { "street": business_data.street_address, "city": business_data.city, "postal_code": business_data.postal_code }, "user_id": str(user["_id"]), "created_at": datetime.utcnow(), "status": "active" } result = businesses_collection.insert_one(business_doc) # Update user's signup step users_collection.update_one( {"_id": user["_id"]}, { "$set": { "signup_step": 2, "business_id": str(result.inserted_id) } } ) return { "message": "Business details saved successfully", "business_id": str(result.inserted_id) } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Step 3: AI Assistant Selection @app.post("/signup/step3") @app.get("/available-assistants") async def get_available_assistants(): """Get list of available AI assistant templates from the agent service""" try: # Call agent service to get templates logger.info(f"=-=-=-=-=-= AGENT_SERVICE: {os.getenv('AGENT_SERVICE')}") async with httpx.AsyncClient(verify=False) as client: response = await client.get( f"{os.getenv('AGENT_SERVICE')}/templates" ) if response.status_code != 200: raise HTTPException( status_code=response.status_code, detail=f"Agent service error: {response.text}" ) templates = response.json() return { "assistants": templates["templates"], "count": templates["count"] } except httpx.RequestError as e: raise HTTPException(status_code=500, detail=f"Agent service request failed: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/signup/step3/create-agent") async def signup_step3(selection: AIAssistantSelection): try: # Verify user has completed previous steps user = users_collection.find_one({"email": selection.user_email, "signup_step": 2}) if not user: raise HTTPException(status_code=400, detail="Complete previous steps first") # Call agent service to create agent from template async with httpx.AsyncClient(verify=False, timeout=30.0) as client: response = await client.post( f"{os.getenv('AGENT_SERVICE')}/agents/create-from-template", json={ "template_id": selection.template_id, "business_id": selection.business_id } ) if response.status_code != 200: raise HTTPException( status_code=response.status_code, detail=f"Agent service error: {response.text}" ) agent_response = response.json() logger.info(f"=-=-=-=-=-= AGENT_RESPONSE: {agent_response}") # Update user with AI assistant details users_collection.update_one( {"_id": user["_id"]}, { "$set": { "signup_step": 3, "assistant_id": agent_response["agent_id"], "retell_agent_id": agent_response["retell_agent_id"] } } ) return { "message": "AI assistant created successfully", "assistant": { "id": agent_response["agent_id"], "retell_agent_id": agent_response["retell_agent_id"] } } except Exception as e: logger.error(f"Error in signup_step3: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) # Step 4: Services Setup @app.post("/signup/step4") async def signup_step4(request: ServiceRequest): try: # Store services in business collection businesses_collection.update_one( {"_id": ObjectId(request.business_id)}, { "$set": { "services": [service.dict() for service in request.services] } } ) # Update user signup step users_collection.update_one( {"business_id": request.business_id}, {"$set": {"signup_step": 4}} ) return {"message": "Services added successfully"} except Exception as e: logger.error(f"Error in signup_step4: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) # Final Step: Work Schedule @app.post("/signup/final") async def signup_final(request: FinalStepRequest): try: # Store schedule in business collection businesses_collection.update_one( {"_id": ObjectId(request.business_id)}, { "$set": { "schedule": request.schedule.dict(), "status": "active" } } ) # Complete user signup users_collection.update_one( {"business_id": request.business_id}, { "$set": { "signup_step": "completed", "status": "active" } } ) return {"message": "Signup completed successfully"} except Exception as e: logger.error(f"Error in signup_final: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) # Sign In @app.post("/signin") async def signin(credentials: SignIn): try: user = await users_collection.find_one({"email": credentials.email}) if not user: raise HTTPException(status_code=401, detail="Invalid credentials") if not bcrypt.checkpw(credentials.password.encode('utf-8'), user["password"]): raise HTTPException(status_code=401, detail="Invalid credentials") if user["status"] != "active": raise HTTPException(status_code=403, detail="Account not active") # Generate JWT token token_data = { "sub": str(user["_id"]), "email": user["email"], "business_id": user.get("business_id"), "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) } token = jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM) return { "access_token": token, "token_type": "bearer", "user": { "id": str(user["_id"]), "email": user["email"], "name": user["name"], "business_id": user.get("business_id") } } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Forgot Password @app.post("/forgot-password") async def forgot_password(email: str): try: user = await users_collection.find_one({"email": email}) if not user: raise HTTPException(status_code=404, detail="User not found") # Send OTP via SMS service async with httpx.AsyncClient() as client: response = await client.post( "http://sms-service:8003/send-otp", json={ "phone": user["phone"], "purpose": "reset_password" } ) if response.status_code != 200: raise HTTPException(status_code=500, detail="Failed to send OTP") return {"message": "Reset code sent to your phone"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # Reset Password @app.post("/reset-password") async def reset_password(phone: str, otp: str, new_password: str): try: # Verify OTP async with httpx.AsyncClient() as client: response = await client.post( "http://sms-service:8003/verify-otp", json={ "phone": phone, "otp": otp, "purpose": "reset_password" } ) if response.status_code != 200: raise HTTPException(status_code=400, detail="Invalid OTP") # Update password hashed_password = bcrypt.hashpw(new_password.encode('utf-8'), bcrypt.gensalt()) result = await users_collection.update_one( {"phone": phone}, {"$set": {"password": hashed_password}} ) if result.modified_count == 0: raise HTTPException(status_code=404, detail="User not found") return {"message": "Password reset successfully"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.on_event("shutdown") async def shutdown_event(): db.close()