110 lines
4.1 KiB
Python
110 lines
4.1 KiB
Python
import uuid
|
|
from datetime import timedelta
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin, BaseUserManager
|
|
from django.db import models
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
|
|
|
|
class UserRole(models.TextChoices):
|
|
ADMIN = "admin", "Admin"
|
|
MANAGER = "manager", "Salon Manager"
|
|
STAFF = "staff", "Staff"
|
|
CUSTOMER = "customer", "Customer"
|
|
|
|
|
|
class UserManager(BaseUserManager):
|
|
def create_user(self, email=None, password=None, **extra_fields):
|
|
phone_number = extra_fields.get("phone_number")
|
|
if not phone_number:
|
|
raise ValueError("Phone number is required")
|
|
if email:
|
|
email = self.normalize_email(email)
|
|
user = self.model(email=email, **extra_fields)
|
|
if password:
|
|
user.set_password(password)
|
|
else:
|
|
user.set_unusable_password()
|
|
user.save(using=self._db)
|
|
return user
|
|
|
|
def create_superuser(self, phone_number, password=None, **extra_fields):
|
|
extra_fields.setdefault("is_staff", True)
|
|
extra_fields.setdefault("is_superuser", True)
|
|
extra_fields.setdefault("role", UserRole.ADMIN)
|
|
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(phone_number=phone_number, password=password, **extra_fields)
|
|
|
|
|
|
class User(AbstractBaseUser, PermissionsMixin):
|
|
email = models.EmailField(unique=True, null=True, blank=True)
|
|
phone_number = models.CharField(max_length=20, unique=True)
|
|
role = models.CharField(max_length=20, choices=UserRole.choices, default=UserRole.CUSTOMER)
|
|
first_name = models.CharField(max_length=150, blank=True)
|
|
last_name = models.CharField(max_length=150, blank=True)
|
|
is_phone_verified = models.BooleanField(default=False)
|
|
preferred_language = models.CharField(
|
|
max_length=10,
|
|
choices=settings.LANGUAGES,
|
|
default=settings.LANGUAGE_CODE,
|
|
)
|
|
is_active = models.BooleanField(default=True)
|
|
is_staff = models.BooleanField(default=False)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
objects = UserManager()
|
|
|
|
USERNAME_FIELD = "phone_number"
|
|
REQUIRED_FIELDS = [] # email is optional; phone_number is the identifier
|
|
|
|
class Meta:
|
|
constraints = [
|
|
models.CheckConstraint(
|
|
name="accounts_user_phone_e164_format",
|
|
condition=Q(phone_number__regex=r"^\+[1-9][0-9]{7,14}$"),
|
|
),
|
|
]
|
|
|
|
def __str__(self):
|
|
return self.email or self.phone_number or str(self.id)
|
|
|
|
|
|
class OtpChannel(models.TextChoices):
|
|
SMS = "sms", "SMS"
|
|
WHATSAPP = "whatsapp", "WhatsApp"
|
|
|
|
|
|
class OtpPurpose(models.TextChoices):
|
|
AUTH = "auth", "Authentication"
|
|
VERIFY = "verify", "Phone Verification"
|
|
|
|
|
|
class PhoneOTP(models.Model):
|
|
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
phone_number = models.CharField(max_length=20)
|
|
channel = models.CharField(max_length=20, choices=OtpChannel.choices)
|
|
purpose = models.CharField(max_length=20, choices=OtpPurpose.choices, default=OtpPurpose.AUTH)
|
|
provider = models.CharField(max_length=50)
|
|
code_hash = models.CharField(max_length=128)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
expires_at = models.DateTimeField()
|
|
verified_at = models.DateTimeField(null=True, blank=True)
|
|
attempt_count = models.PositiveSmallIntegerField(default=0)
|
|
max_attempts = models.PositiveSmallIntegerField(default=5)
|
|
# Request metadata for abuse controls and support investigations.
|
|
request_ip = models.GenericIPAddressField(null=True, blank=True, db_index=True)
|
|
device_signal = models.CharField(max_length=64, blank=True, default="", db_index=True)
|
|
|
|
def is_expired(self):
|
|
return timezone.now() >= self.expires_at
|
|
|
|
@classmethod
|
|
def expiry_at(cls):
|
|
return timezone.now() + timedelta(minutes=settings.OTP_EXPIRY_MINUTES)
|