Introduces a weekly summary email for newly added products, including a new email template and Celery periodic task. Adds 'include_in_week_summary_email' to Product and 'newsletter' to CustomUser. Provides an admin endpoint to manually trigger the weekly email, updates Celery Beat schedule, and adds email templates for verification and password reset.
176 lines
5.7 KiB
Python
176 lines
5.7 KiB
Python
import uuid
|
|
from django.db import models
|
|
from django.contrib.auth.models import AbstractUser, UserManager, Group, Permission
|
|
from django.core.validators import RegexValidator
|
|
|
|
from django.utils.crypto import get_random_string
|
|
from django.conf import settings
|
|
from django.db import models
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
|
|
from vontor_cz.models import SoftDeleteModel
|
|
|
|
from django.contrib.auth.models import UserManager
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class CustomUserManager(UserManager):
|
|
# Inherit get_by_natural_key and all auth behaviors
|
|
use_in_migrations = True
|
|
|
|
class ActiveUserManager(CustomUserManager):
|
|
def get_queryset(self):
|
|
return super().get_queryset().filter(is_active=True)
|
|
|
|
class CustomUser(SoftDeleteModel, AbstractUser):
|
|
groups = models.ManyToManyField(
|
|
Group,
|
|
related_name="customuser_set",
|
|
blank=True,
|
|
help_text="The groups this user belongs to.",
|
|
related_query_name="customuser",
|
|
)
|
|
user_permissions = models.ManyToManyField(
|
|
Permission,
|
|
related_name="customuser_set",
|
|
blank=True,
|
|
help_text="Specific permissions for this user.",
|
|
related_query_name="customuser",
|
|
)
|
|
|
|
class Role(models.TextChoices):
|
|
ADMIN = "admin", "cz#Administrátor"
|
|
MANAGER = "mod", "cz#Moderator"
|
|
CUSTOMER = "regular", "cz#Regular"
|
|
|
|
role = models.CharField(max_length=20, choices=Role.choices, default=Role.CUSTOMER)
|
|
|
|
phone_number = models.CharField(
|
|
null=True,
|
|
blank=True,
|
|
unique=True,
|
|
max_length=16,
|
|
validators=[RegexValidator(r'^\+?\d{9,15}$', message="Zadejte platné telefonní číslo.")]
|
|
)
|
|
|
|
email_verified = models.BooleanField(default=False)
|
|
email = models.EmailField(unique=True, db_index=True)
|
|
|
|
# + fields for email verification flow
|
|
email_verification_token = models.CharField(max_length=128, null=True, blank=True, db_index=True)
|
|
email_verification_sent_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
newsletter = models.BooleanField(default=True)
|
|
|
|
#misc
|
|
gdpr = models.BooleanField(default=False)
|
|
is_active = models.BooleanField(default=False)
|
|
create_time = models.DateTimeField(auto_now_add=True)
|
|
|
|
|
|
#adresa
|
|
postal_code = models.CharField(max_length=20, blank=True)
|
|
city = models.CharField(null=True, blank=True, max_length=100)
|
|
street = models.CharField(null=True, blank=True, max_length=200)
|
|
street_number = models.PositiveIntegerField(null=True, blank=True)
|
|
country = models.CharField(null=True, blank=True, max_length=100)
|
|
|
|
# firemní fakturační údaje
|
|
company_name = models.CharField(max_length=255, blank=True)
|
|
ico = models.CharField(max_length=20, blank=True)
|
|
dic = models.CharField(max_length=20, blank=True)
|
|
|
|
postal_code = models.CharField(
|
|
blank=True,
|
|
null=True,
|
|
|
|
max_length=5,
|
|
validators=[
|
|
RegexValidator(
|
|
regex=r'^\d{5}$',
|
|
message="Postal code must contain exactly 5 digits.",
|
|
code='invalid_postal_code'
|
|
)
|
|
]
|
|
)
|
|
|
|
USERNAME_FIELD = "username"
|
|
REQUIRED_FIELDS = [
|
|
"email"
|
|
]
|
|
|
|
|
|
# Ensure default manager has get_by_natural_key
|
|
objects = CustomUserManager()
|
|
# Optional convenience manager for active users only
|
|
active = ActiveUserManager()
|
|
|
|
def delete(self, *args, **kwargs):
|
|
self.is_active = False
|
|
|
|
return super().delete(*args, **kwargs)
|
|
|
|
def save(self, *args, **kwargs):
|
|
is_new = self._state.adding # True if object hasn't been saved yet
|
|
|
|
# Pre-save flags for new users
|
|
if is_new:
|
|
if self.is_superuser or self.role == "admin":
|
|
# ensure admin flags are consistent
|
|
self.is_active = True
|
|
self.is_staff = True
|
|
self.is_superuser = True
|
|
self.role = "admin"
|
|
else:
|
|
self.is_staff = False
|
|
|
|
# First save to obtain a primary key
|
|
super().save(*args, **kwargs)
|
|
|
|
# Assign group after we have a PK
|
|
if is_new:
|
|
from django.contrib.auth.models import Group
|
|
group, _ = Group.objects.get_or_create(name=self.role)
|
|
# Use add/set now that PK exists
|
|
self.groups.set([group])
|
|
|
|
return super().save(*args, **kwargs)
|
|
|
|
def generate_email_verification_token(self, length: int = 48, save: bool = True) -> str:
|
|
token = get_random_string(length=length)
|
|
self.email_verification_token = token
|
|
self.email_verification_sent_at = timezone.now()
|
|
if save:
|
|
self.save(update_fields=["email_verification_token", "email_verification_sent_at"])
|
|
return token
|
|
|
|
def verify_email_token(self, token: str, max_age_hours: int = 48, save: bool = True) -> bool:
|
|
if not token or not self.email_verification_token:
|
|
return False
|
|
# optional expiry check
|
|
if self.email_verification_sent_at:
|
|
age = timezone.now() - self.email_verification_sent_at
|
|
if age > timedelta(hours=max_age_hours):
|
|
return False
|
|
if token != self.email_verification_token:
|
|
return False
|
|
|
|
if not self.email_verified:
|
|
self.email_verified = True
|
|
# clear token after success
|
|
self.email_verification_token = None
|
|
self.email_verification_sent_at = None
|
|
if save:
|
|
self.save(update_fields=["email_verified", "email_verification_token", "email_verification_sent_at"])
|
|
return True
|
|
|
|
def get_anonymous_user():
|
|
"""Return the singleton anonymous user."""
|
|
User = CustomUser
|
|
return User.objects.get(username="anonymous")
|
|
|
|
|