Spaces:
Sleeping
Sleeping
File size: 6,065 Bytes
2cd3d4e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | import re
import logging
from app.repositories.cache_repository import get_otp, set_otp
from app.repositories.db_repository import get_customer_by_email, get_customer_by_mobile, save_customer
from app.models.customer import CustomerRegister
from app.utils.auth_utils import validate_email, validate_mobile
import random
logger = logging.getLogger(__name__)
async def validate_email_mobile(email: str, mobile: str) -> dict:
"""
Validate email and mobile format and check if they already exist in the database.
Args:
email (str): The email address to validate.
mobile (str): The mobile number to validate.
Returns:
dict: Validation result with consolidated error messages or success message.
"""
errors = []
# Validate email format
if not validate_email(email):
errors.append("Invalid email format")
# Validate mobile format
if not validate_mobile(mobile):
errors.append("Invalid mobile number format")
# Check if email already exists in the database
existing_email = await get_customer_by_email(email)
if existing_email:
errors.append("Email is already registered")
# Check if mobile already exists in the database
existing_mobile = await get_customer_by_mobile(mobile)
if existing_mobile:
errors.append("Mobile number is already registered")
# Return errors if any
if errors:
return {"status": "error", "errors": errors}
else:
# Send verification emails and SMS if no errors
await send_email_verification(email)
await send_sms_verification(mobile)
# If all validations pass
return {"status": "success", "message": "Email and mobile are valid"}
async def save_customer_to_db(user: CustomerRegister) -> dict:
"""
Save customer details to MongoDB after checking for duplicates.
Args:
user (CustomerRegister): The customer details to save.
Returns:
dict: Success message or error message if email or mobile already exists.
"""
# Check if email already exists in the database
existing_email = await get_customer_by_email(user.email)
if existing_email:
return {"status": "error", "message": "Email is already registered"}
# Check if mobile already exists in the database
existing_mobile = await get_customer_by_mobile(user.mobile)
if existing_mobile:
return {"status": "error", "message": "Mobile number is already registered"}
# Save customer details to the database
await save_customer(user.dict())
return {"status": "success", "message": "Customer registered successfully"}
async def send_email_verification(email: str):
"""
Send an email verification link or code.
Args:
email (str): The email address to send the verification to.
"""
# Implement your email sending logic here
# Generate a 6-digit OTP
otp = ''.join(random.choices("0123456789", k=6))
# Store OTP in cache with a 15-minute expiry
await set_otp(email, {"otp": otp, "expiry_duration": 15 * 60}) # 15 minutes in seconds
# Send the OTP to the email address
# (You would typically use an email service here)
# For demonstration, we'll just log the OTP
logger.info("Sending email verification to: %s with OTP: %s", email, otp)
logger.info("Sending email verification to: %s", email)
async def send_sms_verification(mobile: str):
"""
Send an SMS verification code.
Args:
mobile (str): The mobile number to send the verification to.
"""
# Implement your SMS sending logic here
# Generate a 6-digit OTP
otp = ''.join(random.choices("0123456789", k=6))
# Store OTP in cache with a 15-minute expiry
await set_otp(mobile, {"otp": otp, "expiry_duration": 15 * 60}) # 15 minutes in seconds
# Send the OTP to the mobile number
# (You would typically use an SMS service here)
# For demonstration, we'll just log the OTP
logger.info("Sending SMS verification to: %s with OTP: %s", mobile, otp)
logger.info("Sending SMS verification to: %s", mobile)
async def verify_otp_and_save_customer(user: CustomerRegister) -> dict:
"""
Verify SMS OTP and email OTP, and save the complete customer object in the database.
Args:
user (CustomerRegister): The complete customer object including email, mobile, and OTPs.
Returns:
dict: Result indicating success or failure.
"""
errors = []
# Retrieve OTPs from cache
email_otp_data = await get_otp(user.email)
mobile_otp_data = await get_otp(user.mobile)
# Verify email OTP
if not email_otp_data or email_otp_data["otp"] != user.email_ver_code:
errors.append("Invalid or expired email OTP")
# Verify mobile OTP
if not mobile_otp_data or mobile_otp_data["otp"] != user.mobile_ver_code:
errors.append("Invalid or expired mobile OTP")
# Return errors if any
if errors:
return {"status": "error", "errors": errors}
# Save the complete customer object to the database
customer_data = user.dict()
customer_data["status"] = "completed" # Update status to completed
await save_customer(customer_data)
return {"status": "success", "message": "Customer successfully registered and verified"}
async def get_otp_from_cache(key: str) -> dict:
"""
Retrieve OTP data for the given key (email or mobile) from the cache.
Args:
key (str): The email or mobile number to retrieve the OTP for.
Returns:
dict: The OTP data if found, or None if not found.
"""
try:
# Call the get_otp function from the cache repository
otp_data = await get_otp(key)
return otp_data
except Exception as e:
# Log the error and re-raise it
logger.error(f"Failed to retrieve OTP for key {key}: {e}")
raise |