269 lines
10 KiB
Python
269 lines
10 KiB
Python
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 apps.core.models import SMSPolicy
|
|
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
|
|
self.otp_expire = None
|
|
|
|
if not self.date_joined:
|
|
self.date_joined = timezone.now()
|
|
else:
|
|
self.otp_try += 1
|
|
self.save()
|
|
else:
|
|
result = False
|
|
self.otp = None
|
|
self.otp_expire = None
|
|
self.save()
|
|
|
|
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}"
|
|
)
|
|
|
|
sms_client = SMSPolicy.get_client()
|
|
sms_client.send_sms(self.phone_number.strip('+'), message)
|
|
|
|
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')
|