# ============================================================ # app/services/otp_service.py – Resend 0.7+ compatible # ============================================================ import random import logging import re from datetime import datetime, timedelta, timezone from fastapi import HTTPException, status from app.database import get_db from app.config import settings import resend # ← new import logger = logging.getLogger(__name__) class OTPService: """OTP Service for sending and verifying OTPs""" def __init__(self): self.resend_client = None self._initialize_email_service() # ------------------------------------------------------------------ # Email service init (Resend 0.7+) # ------------------------------------------------------------------ def _initialize_email_service(self): try: if settings.RESEND_API_KEY: resend.api_key = settings.RESEND_API_KEY # global key logger.info("Resend email service initialized") except Exception as e: logger.error(f"Failed to initialize email service: {str(e)}") # ------------------------------------------------------------------ # Helper methods (unchanged) # ------------------------------------------------------------------ @staticmethod def _validate_identifier(identifier: str) -> str: email_pattern = r"^[^\s@]+@[^\s@]+\.[^\s@]+$" phone_pattern = r"^\+\d{1,15}$" if re.match(email_pattern, identifier): return "email" elif re.match(phone_pattern, identifier): return "phone" raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid email or phone format" ) @staticmethod def _generate_otp() -> str: return str(random.randint(1000, 9999)) # ------------------------------------------------------------------ # Generate OTP (unchanged) # ------------------------------------------------------------------ async def generate_otp(self, identifier: str, purpose: str) -> str: self._validate_identifier(identifier) db = await get_db() otps_collection = db["otps"] await otps_collection.delete_many({ "identifier": identifier, "purpose": purpose }) code = self._generate_otp() otp_doc = { "identifier": identifier, "code": code, "purpose": purpose, "isVerified": False, "attempts": 0, "createdAt": datetime.utcnow(), } try: await otps_collection.insert_one(otp_doc) logger.info(f"OTP generated for {purpose}: {identifier}") return code except Exception as e: logger.error(f"Error generating OTP: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to generate OTP" ) # ------------------------------------------------------------------ # Send OTP (unchanged) # ------------------------------------------------------------------ async def send_otp(self, identifier: str, purpose: str) -> None: identifier_type = self._validate_identifier(identifier) otp_code = await self.generate_otp(identifier, purpose) if identifier_type == "email": await self._send_otp_via_email(identifier, otp_code, purpose) else: await self._send_otp_via_sms(identifier, otp_code, purpose) # ------------------------------------------------------------------ # EMAIL – Resend 0.7+ syntax # ------------------------------------------------------------------ async def _send_otp_via_email(self, email: str, otp: str, purpose: str) -> None: try: template = self._generate_email_template(otp, purpose) subject = ( "Lojiz - Verify Your Account" if purpose == "signup" else "Lojiz - Reset Your Password" ) params = { "from": f"{settings.RESEND_FROM_NAME} <{settings.RESEND_FROM_EMAIL}>", "to": email, "subject": subject, "html": template, } resp = resend.Emails.send(params) # ← new API if resp.get("error"): raise Exception(resp["error"].get("message", "Failed to send email")) logger.info(f"Email sent for {purpose}: {email}") except Exception as e: logger.error(f"Error sending email: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to send OTP email" ) # ------------------------------------------------------------------ # SMS stub (unchanged) # ------------------------------------------------------------------ async def _send_otp_via_sms(self, phone: str, otp: str, purpose: str) -> None: logger.warning(f"SMS not implemented for {purpose}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="SMS service not configured" ) # ------------------------------------------------------------------ # Template (unchanged) # ------------------------------------------------------------------ @staticmethod def _generate_email_template(otp: str, purpose: str) -> str: current_year = datetime.now().year expiry_minutes = settings.OTP_EXPIRY_MINUTES subject_line = "Verify Your Account" if purpose == "signup" else "Reset Your Password" intro_text = ( "Your One-Time Password (OTP) for account verification is:" if purpose == "signup" else "We received a request to reset your password. Use this OTP code to proceed:" ) return f""" Your OTP Code

{subject_line}

Hello,

{intro_text}

{otp}
This code expires in {expiry_minutes} minutes

Please do not share this code with anyone. If you didn't request this code, please ignore this email.

© {current_year} Lojiz. All rights reserved.

""" # ------------------------------------------------------------------ # Verify / delete / expiry helpers (unchanged) # ------------------------------------------------------------------ async def verify_otp(self, identifier: str, code: str, purpose: str) -> bool: db = await get_db() otps_collection = db["otps"] otp_doc = await otps_collection.find_one({ "identifier": identifier, "purpose": purpose, "isVerified": False }) if not otp_doc: raise HTTPException(status_code=400, detail="OTP not found or already verified") expiry_time = otp_doc["createdAt"] + timedelta(minutes=settings.OTP_EXPIRY_MINUTES) if datetime.utcnow() > expiry_time: await otps_collection.delete_one({"_id": otp_doc["_id"]}) raise HTTPException(status_code=400, detail="OTP has expired") if otp_doc["attempts"] >= settings.OTP_MAX_ATTEMPTS: await otps_collection.delete_one({"_id": otp_doc["_id"]}) raise HTTPException(status_code=400, detail="Maximum attempts exceeded") if otp_doc["code"] != code: otp_doc["attempts"] += 1 await otps_collection.update_one({"_id": otp_doc["_id"]}, {"$set": {"attempts": otp_doc["attempts"]}}) attempts_left = settings.OTP_MAX_ATTEMPTS - otp_doc["attempts"] raise HTTPException(status_code=400, detail=f"Invalid OTP. {attempts_left} attempts remaining") await otps_collection.update_one({"_id": otp_doc["_id"]}, {"$set": {"isVerified": True}}) logger.info(f"OTP verified for {purpose}: {identifier}") return True async def delete_otp(self, identifier: str, purpose: str) -> None: db = await get_db() otps_collection = db["otps"] await otps_collection.delete_many({"identifier": identifier, "purpose": purpose}) logger.info(f"OTP deleted for {purpose}: {identifier}") async def is_otp_expired(self, identifier: str, purpose: str) -> bool: db = await get_db() otps_collection = db["otps"] otp_doc = await otps_collection.find_one({"identifier": identifier, "purpose": purpose, "isVerified": False}) if not otp_doc: return True expiry_time = otp_doc["createdAt"] + timedelta(minutes=settings.OTP_EXPIRY_MINUTES) return datetime.utcnow() > expiry_time # ------------------------------------------------------------------ # Singleton # ------------------------------------------------------------------ otp_service = OTPService()