Egor fureunoir Gorbunov
ad320235d6
- Refactored monetary fields across models to use `DecimalField` for improved precision. - Implemented two-factor authentication (2FA) for admin logins with OTP codes. - Added ability to generate admin OTP via management commands. - Updated Docker Compose override for dev-specific port bindings. - Included template for 2FA OTP verification to enhance security. Additional changes: - Upgraded and downgraded various dependencies (e.g., django-celery-beat and yarl). - Replaced float-based calculations with decimal for consistent rounding behavior. - Improved admin user management commands for activation and OTP generation.
284 lines
9 KiB
Python
284 lines
9 KiB
Python
from uuid import uuid4
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AbstractUser
|
|
from django.contrib.auth.models import Group as BaseGroup
|
|
from django.core.cache import cache
|
|
from django.core.exceptions import ValidationError
|
|
from django.db.models import (
|
|
CASCADE,
|
|
SET_NULL,
|
|
BooleanField,
|
|
CharField,
|
|
DateTimeField,
|
|
EmailField,
|
|
ForeignKey,
|
|
ImageField,
|
|
Index,
|
|
JSONField,
|
|
TextField,
|
|
UUIDField,
|
|
)
|
|
from django.templatetags.static import static
|
|
from django.utils import timezone
|
|
from django.utils.functional import cached_property
|
|
from django.utils.translation import gettext_lazy as _
|
|
from encrypted_fields.fields import EncryptedCharField
|
|
from rest_framework_simplejwt.token_blacklist.models import (
|
|
BlacklistedToken as BaseBlacklistedToken,
|
|
)
|
|
from rest_framework_simplejwt.token_blacklist.models import (
|
|
OutstandingToken as BaseOutstandingToken,
|
|
)
|
|
|
|
from engine.core.abstract import NiceModel
|
|
from engine.payments.models import Balance
|
|
from engine.vibes_auth.choices import SenderType, ThreadStatus
|
|
from engine.vibes_auth.managers import UserManager
|
|
from engine.vibes_auth.validators import validate_phone_number
|
|
from schon.fields import EncryptedJSONTextField
|
|
|
|
|
|
class User(AbstractUser, NiceModel):
|
|
__doc__ = _(
|
|
"Represents a User entity with customized fields and methods for extended functionality. "
|
|
"This class extends the AbstractUser model and integrates additional features like "
|
|
"custom email login, validation methods, subscription status, verification, and "
|
|
"attributes storage. It also provides utilities for managing recently viewed items and "
|
|
"token-based activation for verifying accounts. The User model is designed to handle "
|
|
"specific use cases for enhanced user management."
|
|
)
|
|
|
|
def get_uuid_as_path(self, *args):
|
|
return "users/" + str(self.uuid) + "/" + args[0]
|
|
|
|
email = EmailField(_("email"), unique=True, help_text=_("user email address"))
|
|
phone_number = EncryptedCharField(
|
|
_("phone_number"),
|
|
max_length=20,
|
|
blank=True,
|
|
null=True,
|
|
help_text=_("user phone number"),
|
|
validators=[
|
|
validate_phone_number,
|
|
],
|
|
)
|
|
username: None = None
|
|
first_name = EncryptedCharField(
|
|
_("first_name"), max_length=150, blank=True, null=True
|
|
)
|
|
last_name = EncryptedCharField(
|
|
_("last_name"), max_length=150, blank=True, null=True
|
|
)
|
|
avatar = ImageField(
|
|
null=True,
|
|
verbose_name=_("avatar"),
|
|
upload_to=get_uuid_as_path,
|
|
blank=True,
|
|
help_text=_("user profile image"),
|
|
)
|
|
|
|
is_verified = BooleanField(
|
|
default=False,
|
|
verbose_name=_("is verified"),
|
|
help_text=_("user verification status"),
|
|
)
|
|
is_active = BooleanField(
|
|
_("is_active"),
|
|
default=False,
|
|
help_text=_("unselect this instead of deleting accounts"),
|
|
)
|
|
is_subscribed = BooleanField(
|
|
verbose_name=_("is_subscribed"),
|
|
help_text=_("user's newsletter subscription status"),
|
|
default=False,
|
|
)
|
|
|
|
activation_token = CharField(
|
|
max_length=128, blank=True, default="", verbose_name=_("activation token")
|
|
)
|
|
activation_token_created = DateTimeField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("activation token created"),
|
|
)
|
|
unsubscribe_token = UUIDField(
|
|
default=uuid4,
|
|
verbose_name=_("unsubscribe token"),
|
|
help_text=_("token for secure one-click unsubscribe from campaigns"),
|
|
)
|
|
language = CharField(
|
|
choices=settings.LANGUAGES,
|
|
default=settings.LANGUAGE_CODE,
|
|
null=False,
|
|
blank=False,
|
|
max_length=7,
|
|
)
|
|
attributes = EncryptedJSONTextField(
|
|
verbose_name=_("attributes"), default=dict, blank=True, null=True
|
|
)
|
|
|
|
payments_balance: "Balance"
|
|
|
|
USERNAME_FIELD = "email"
|
|
REQUIRED_FIELDS = []
|
|
# noinspection PyClassVar
|
|
objects = UserManager()
|
|
|
|
@cached_property
|
|
def avatar_url(self) -> str:
|
|
try:
|
|
return self.avatar.url
|
|
except ValueError:
|
|
return static("person.png")
|
|
|
|
def add_to_recently_viewed(self, product_uuid):
|
|
recently_viewed = self.recently_viewed
|
|
if product_uuid not in recently_viewed:
|
|
if not len(recently_viewed) >= 48:
|
|
recently_viewed.append(product_uuid)
|
|
cache.set(f"user_{self.uuid}_rv", recently_viewed)
|
|
else:
|
|
recently_viewed.pop(0)
|
|
recently_viewed.append(product_uuid)
|
|
cache.set(f"user_{self.uuid}_rv", recently_viewed)
|
|
|
|
@property
|
|
def recently_viewed(self):
|
|
return cache.get(f"user_{self.uuid}_rv", [])
|
|
|
|
def save(self, *args, **kwargs):
|
|
if self._state.adding and self.activation_token_created is None:
|
|
self.activation_token_created = timezone.now()
|
|
if self._state.adding and not self.activation_token:
|
|
self.refresh_activation_token()
|
|
super().save(*args, **kwargs)
|
|
|
|
def refresh_activation_token(self) -> str:
|
|
"""Generate a fresh activation token, store its hash, return raw token."""
|
|
import hashlib
|
|
|
|
raw_token = str(uuid4())
|
|
self.activation_token = hashlib.sha256(raw_token.encode()).hexdigest()
|
|
self.activation_token_created = timezone.now()
|
|
return raw_token
|
|
|
|
def check_token(self, token) -> bool:
|
|
import hashlib
|
|
from datetime import timedelta
|
|
|
|
hashed = hashlib.sha256(str(token).encode()).hexdigest()
|
|
if hashed != self.activation_token:
|
|
return False
|
|
if self.activation_token_created:
|
|
if timezone.now() > self.activation_token_created + timedelta(hours=24):
|
|
return False
|
|
return True
|
|
|
|
def __str__(self):
|
|
return self.email
|
|
|
|
class Meta:
|
|
swappable = "AUTH_USER_MODEL"
|
|
verbose_name = _("user")
|
|
verbose_name_plural = _("users")
|
|
|
|
|
|
class AdminOTPCode(NiceModel):
|
|
user = ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=CASCADE,
|
|
related_name="otp_codes",
|
|
)
|
|
code = CharField(max_length=6)
|
|
is_used = BooleanField(default=False)
|
|
|
|
def is_valid(self) -> bool:
|
|
from datetime import timedelta
|
|
|
|
return not self.is_used and timezone.now() < self.created + timedelta(minutes=5)
|
|
|
|
class Meta:
|
|
verbose_name = _("admin OTP code")
|
|
verbose_name_plural = _("admin OTP codes")
|
|
|
|
|
|
class ChatThread(NiceModel):
|
|
user = ForeignKey(
|
|
User, null=True, blank=True, on_delete=SET_NULL, related_name="chat_threads"
|
|
)
|
|
email = EmailField(blank=True, default="", help_text=_("For anonymous threads"))
|
|
assigned_to = ForeignKey(
|
|
User,
|
|
null=True,
|
|
blank=True,
|
|
on_delete=SET_NULL,
|
|
related_name="assigned_chat_threads",
|
|
)
|
|
status = CharField(
|
|
max_length=16, choices=ThreadStatus.choices, default=ThreadStatus.OPEN
|
|
)
|
|
last_message_at = DateTimeField(default=timezone.now)
|
|
attributes = JSONField(default=dict, blank=True)
|
|
|
|
class Meta:
|
|
indexes = [
|
|
Index(fields=["status", "modified"], name="chatthread_status_mod_idx"),
|
|
Index(
|
|
fields=["assigned_to", "status"], name="chatthread_assigned_status_idx"
|
|
),
|
|
Index(fields=["user"], name="chatthread_user_idx"),
|
|
Index(fields=["email"], name="chatthread_email_idx"),
|
|
]
|
|
ordering = ("-modified",)
|
|
verbose_name = _("Chat thread")
|
|
verbose_name_plural = _("Chat threads")
|
|
|
|
def clean(self) -> None:
|
|
super().clean()
|
|
if not self.user and not self.email:
|
|
raise ValidationError(
|
|
{"email": _("provide user or email for anonymous thread.")}
|
|
)
|
|
if self.assigned_to and not self.assigned_to.is_staff:
|
|
raise ValidationError({"assigned_to": _("assignee must be a staff user.")})
|
|
|
|
|
|
class ChatMessage(NiceModel):
|
|
thread = ForeignKey(ChatThread, on_delete=CASCADE, related_name="messages")
|
|
sender_type = CharField(max_length=16, choices=SenderType.choices)
|
|
sender_user = ForeignKey(
|
|
User, null=True, blank=True, on_delete=SET_NULL, related_name="chat_messages"
|
|
)
|
|
text = TextField()
|
|
sent_at = DateTimeField(default=timezone.now)
|
|
attributes = JSONField(default=dict, blank=True)
|
|
|
|
class Meta:
|
|
indexes = [
|
|
Index(fields=["thread", "sent_at"], name="chatmessage_thread_sent_idx"),
|
|
]
|
|
ordering = ("sent_at", "uuid")
|
|
verbose_name = _("Chat message")
|
|
verbose_name_plural = _("Chat messages")
|
|
|
|
|
|
class Group(BaseGroup):
|
|
class Meta:
|
|
proxy = True
|
|
verbose_name = _("group")
|
|
verbose_name_plural = _("groups")
|
|
|
|
|
|
class OutstandingToken(BaseOutstandingToken):
|
|
class Meta:
|
|
proxy = True
|
|
verbose_name = _("outstanding token")
|
|
verbose_name_plural = _("outstanding tokens")
|
|
|
|
|
|
class BlacklistedToken(BaseBlacklistedToken):
|
|
class Meta:
|
|
proxy = True
|
|
verbose_name = _("blacklisted token")
|
|
verbose_name_plural = _("blacklisted tokens")
|