from datetime import datetime, timedelta from django.contrib.auth import get_user_model from rest_framework import serializers from apps.user_auth.auth_service import SmsOtpService from utils.services.sms import DefaultSmsProvider from apps.user_auth.models import SmsOtp as SmsOtpModel from apps.user_auth.auth_service import SMSAuth from apps.user_auth.token_service import get_tokens_for_user from rest_framework import exceptions from django.db import transaction from apps.user_auth.models import JwtToken as JwtTokenModel from django.contrib.auth import authenticate from django.conf import settings from typing import Annotated import random CUserModel = get_user_model() def login_by_user_pass(username: str, password: str): user_obj = authenticate(username=username, password=password) if user_obj: try: token = get_tokens_for_user(user_obj) JwtTokenModel.objects.create( refresh=token['refresh'], access=token['access'], cuser=user_obj, ) except Exception as exp: raise exceptions.ValidationError({"error": str(exp)}) return token else: raise exceptions.ValidationError({"error": "invalid user or password"}) @transaction.atomic def send_otp_code(user_mobile: int) -> bool: try: CUserModel.objects.get(mobile=int(user_mobile)) except CUserModel.DoesNotExist: raise exceptions.ValidationError({'error': 'User not found'}) default_sms_provider = DefaultSmsProvider() otp_service = SmsOtpService(sms_provider=default_sms_provider) if settings.OTP_BYPASS: random_otp = 123456 else: random_otp = random.randint(100000, 999999) otp_is_send = otp_service.send_code(mobile=str(user_mobile), otp_code=random_otp) if not otp_is_send: raise exceptions.ValidationError({'error': 'Cannot Send Otp Code'}) # remove old otps SmsOtpModel.objects.filter(mobile=str(user_mobile)).delete() # save otp to database expire_time = datetime.now() + timedelta(minutes=4) otp_obj = SmsOtpModel.objects.create( mobile=str(user_mobile), otp=random_otp, expire=expire_time, ) return True def sms_auth(mobile, otp_code: int) -> dict | None: sms_auth = SMSAuth( mobile=str(mobile), otp_code=otp_code, otp_model=SmsOtpModel, ) otp_obj = sms_auth.authenticate() if otp_obj: try: user_obj = CUserModel.objects.get(mobile=otp_obj.mobile) token = get_tokens_for_user(user_obj) JwtTokenModel.objects.create( refresh=token['refresh'], access=token['access'], cuser=user_obj, ) except Exception as exp: raise exceptions.ValidationError({"error": str(exp)}) return token else: raise exceptions.ValidationError({"error": "invalid otp code or mobile"})