92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
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"})
|