import logging import random import string from datetime import timedelta # import service_clients from django.conf import settings from django.contrib.auth.base_user import BaseUserManager from django.contrib.auth.hashers import make_password, check_password 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 _ import uuid from django_minio_backend import MinioBackend, iso_date_prefix from utils.clients.payam_sms import send_sms from .provinces_and_cities import state # from .tasks import send_notification from apps.users.constans import MAX_OTP_TRY, DEVELOPMENT_PHONE_NUMBERS, GenderChoices, EducationDegreeChoices, \ MAX_OTP_VALID_DURATION logger = logging.getLogger(__name__) 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 and not username: 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() 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) 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): uuid = models.UUIDField(primary_key=True, editable=False, default=uuid.uuid4, unique=True, db_index=True) name = models.CharField(_('name'), max_length=30, blank=True) first_name = models.CharField(_('first name'), max_length=30, blank=True) last_name = models.CharField(_('last name'), max_length=150, blank=True) email = models.EmailField(_('email address'), blank=True) email_verified = models.BooleanField(null=True) # country = models.CharField(_('country'), max_length=50, blank=True, null=True) # province = models.CharField(_('province'), max_length=50, blank=True, null=True) # city = models.CharField(_('city'), max_length=50, blank=True, null=True) # postal_code = models.CharField(_('postal_code'), max_length=20, blank=True, null=True) # address = models.TextField(_('address'), null=True, blank=True) province = models.IntegerField(verbose_name=_("Province"), blank=True, null=True) city = models.IntegerField(verbose_name=_("City"), blank=True, null=True) postal_code = models.CharField(_('postal_code'), max_length=20, blank=True, null=True) address = models.TextField(_('address'), null=True, blank=True) is_staff = models.BooleanField( _('staff status'), default=False, help_text=_('Designates whether the user can log into this admin site.'), ) is_active = models.BooleanField( _('active'), default=True, help_text=_( 'Designates whether this user should be treated as active. ' 'Unselect this instead of deleting accounts.' ), ) password = models.CharField(_('password'), max_length=128) 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=128, 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) ott = models.CharField(_('one time token'), max_length=36, blank=True, null=True) ott_expire = models.DateTimeField(_('otp expire'), blank=True, null=True) last_checkout_request = models.DateTimeField(_('otp expire'), max_length=30, blank=True, null=True) balance = models.BigIntegerField(_('balance'), default=0) iban = models.CharField(null=True, max_length=30, blank=True) iban_verified = models.BooleanField(null=True) avatar = models.ImageField(null=True, blank=True, storage=MinioBackend(), upload_to=iso_date_prefix) gender = models.CharField(_('gender'), choices=GenderChoices.choices, blank=True, null=True) education_degree = models.CharField(_('education degree'), choices=EducationDegreeChoices.choices, max_length=10, null=True, blank=True) education_field = models.CharField(_('education field'), max_length=100, null=True, blank=True) birthdate = models.DateField(_('birthdate'), null=True, blank=True) last_update = models.DateTimeField(_('last update'), max_length=30, blank=True, null=True, auto_now=True) last_login = models.DateTimeField(_('last login'), blank=True, null=True) date_joined = models.DateTimeField(_('date joined'), default=timezone.now) objects = UserManager() class Meta: verbose_name = _("user") verbose_name_plural = _("users") ordering = ['-date_joined'] def get_display_name(self): return str(self.name or self.get_full_name() or self.username or self.email or self.phone_number or self.pk) @property def city_name(self): return state.get_city_by_id(self.city) @property def province_name(self): return state.get_province_by_id(self.province) def set_otp(self): if not settings.SMS_SEND: self._otp = '77501' elif self.phone_number in settings.TEST_PHONENUMBERS: self._otp = '88555' else: self._otp = ''.join(random.choice('0123456789') for _ in range(5)) self.otp = make_password(self._otp) self.otp_expire = timezone.now() + timedelta(seconds=MAX_OTP_VALID_DURATION) self.otp_try = 0 self.save() def set_ott(self): self.ott = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(32)) # self.ott = ''.join(random.choice(string.ascii_letters+string.digits+string.punctuation) for _ in range(32)) self.ott_expire = timezone.now() + timedelta(seconds=MAX_OTP_VALID_DURATION) def otp_is_valid(self): return bool(self.otp and self.otp_expire and timezone.now() <= self.otp_expire) def ott_is_valid(self): return bool(self.ott and self.ott_expire and timezone.now() <= self.ott_expire) def check_otp(self, otp): if self.otp_try <= MAX_OTP_TRY: try: result = otp and check_password(otp, self.otp) and self.otp_is_valid() except Exception as e: result = False if result: self.otp = None if not self.date_joined: self.date_joined = timezone.now() else: self.otp_try += 1 self.save() else: result = False return result def check_ott(self, ott: str): ott = ott.strip() if ott and self.ott == ott and self.ott_is_valid(): self.ott = None self.date_joined = timezone.now() result = True else: self.ott = None result = False self.save() return result def check_auth(self, field, value): result = False if field == 'otp': result = self.check_otp(value) elif field == 'password': result = self.check_password(value) elif field == 'ott': result = self.check_ott(value) if result: self.last_login = timezone.now() self.save() return result def send_otp(self): print(self._otp) self.notify(body=str(self._otp), title='OTP') def notify(self, body, title=None, notification_type='sms'): message = ("وینسو" "\n" f"رمزیکبارمصرف: {body}" # "\n" # # f"code is: {body}\n" # f"{settings.SMS_OTP_SIGNITURE}" ) if settings.SMS_SEND: send_sms(self.phone_number.strip('+'), message) else: logger.info(f'otp for: {self.phone_number} is {message}') return # TODO: enable celery send_notification( phone_number=self.phone_number, title=title, body=body, user_uuid=self.uuid, notification_type=1, ) def __str__(self): return self.name or self.get_full_name() or self.username or self.phone_number or self.email or _('no name')