import random from datetime import timedelta from django.contrib.auth.base_user import BaseUserManager from django.contrib.auth.models import AbstractUser from django.contrib.auth.validators import UnicodeUsernameValidator from django.db import models from django.utils import timezone from django.utils.translation import gettext_lazy as _ from rest_framework.exceptions import APIException from apps.users.constans import MAX_OTP_TRY class UserManager(BaseUserManager): use_in_migrations = True def _create_user(self, phone_number=None, username=None, email=None, password=None, **extra_fields): if not phone_number and not email: raise ValueError('The given phone_number or email must be set') email = self.normalize_email(email) username = self.model.normalize_username(username) # TODO: validate phone number user = self.model(phone_number=phone_number, username=username, email=email, **extra_fields) user.set_password(password) user.set_otp(with_save=False) user.set_code() user.date_joined = timezone.now() user.save(using=self._db) return user def create_user(self, phone_number=None, username=None, email=None, password=None, **extra_fields): extra_fields.setdefault('is_staff', False) extra_fields.setdefault('is_superuser', False) # TODO create OTP return self._create_user(phone_number=phone_number, username=username, email=email, password=password, **extra_fields) def create_superuser(self, username, email, password, **extra_fields): if not username: raise ValueError('The given username must be set') extra_fields.setdefault('is_staff', True) extra_fields.setdefault('is_superuser', True) if extra_fields.get('is_staff') is not True: raise ValueError('Superuser must have is_staff=True.') if extra_fields.get('is_superuser') is not True: raise ValueError('Superuser must have is_superuser=True.') return self._create_user(username=username, email=email, password=password, **extra_fields) class User(AbstractUser): # id # first_name # last_name # email # is_staff # is_active # password name = models.CharField(_('name'), max_length=30, blank=True) username_validator = UnicodeUsernameValidator() username = models.CharField( _('username'), max_length=150, unique=True, help_text=_('Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.'), validators=[username_validator], error_messages={ 'unique': _("A user with that username already exists."), }, blank=True, null=True ) phone_number = models.CharField(_('phone number'), max_length=30, blank=True, null=True, unique=True) otp = models.CharField(_('otp'), max_length=6, blank=True, null=True) otp_expire = models.DateTimeField(_('otp expire'), blank=True, null=True) otp_try = models.IntegerField(_('otp try'), blank=True, null=True, default=0) last_checkout_request = models.DateTimeField(_('otp expire'), max_length=30, blank=True, null=True) balance = models.IntegerField(_('balance'), default=0) iban = models.CharField(null=True, max_length=30, blank=True) iban_verified = models.BooleanField(null=True) avatar = models.ImageField(upload_to='avatars', null=True, blank=True) code = models.IntegerField(_('code'), unique=True, null=True, db_index=True) # unique random number last_update = models.DateTimeField(_('last update'), max_length=30, blank=True, null=True, auto_now=True) # last_login # date_joined objects = UserManager() def set_otp(self, with_save=True): self.otp = '12345' # if self.phone_number in DEVELOPMENT_PHONE_NUMBERS: # self.otp = '12345' # else: # self.otp = ''.join(random.choice('0123456789') for _ in range(5)) self.otp_expire = timezone.now() + timedelta(minutes=5) self.otp_try = 0 if with_save: self.save() def set_code(self): for code in random.sample(range(10000, 100000), 90000): try: code = str(code) self.code = code return self.save() except: pass raise APIException('system error') def otp_is_valid(self): return bool(self.otp and timezone.now() <= self.otp_expire) def check_otp(self, otp): if self.otp_try <= MAX_OTP_TRY: result = otp and self.otp == otp and self.otp_is_valid() if result: self.otp = None self.date_joined = timezone.now() else: self.otp_try += 1 else: self.otp = None result = False self.save() return result def check_auth(self, gateway, value): result = False if gateway == 'otp': result = self.check_otp(value) elif gateway == 'password': result = self.check_password(value) if result: self.last_login = timezone.now() self.save() return result def send_otp(self): # TODO: some action to sent OTP return True def __str__(self): return f"{self.code} - {self.username}"