280 lines
11 KiB
Python
280 lines
11 KiB
Python
"""Account models"""
|
|
from datetime import datetime
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import AbstractUser, UserManager as BaseUserManager
|
|
from django.contrib.auth.tokens import default_token_generator as password_token_generator
|
|
from django.core.mail import send_mail
|
|
from django.db import models
|
|
from django.template.loader import render_to_string, get_template
|
|
from django.utils import timezone
|
|
from django.utils.encoding import force_bytes
|
|
from django.utils.html import mark_safe
|
|
from django.utils.http import urlsafe_base64_encode
|
|
from django.utils.translation import ugettext_lazy as _
|
|
from rest_framework.authtoken.models import Token
|
|
|
|
from authorization.models import Application
|
|
from establishment.models import Establishment
|
|
from location.models import Country
|
|
from main.models import SiteSettings
|
|
from utils.models import GMTokenGenerator
|
|
from utils.models import ImageMixin, ProjectBaseMixin, PlatformMixin
|
|
from utils.tokens import GMRefreshToken
|
|
|
|
|
|
class Role(ProjectBaseMixin):
|
|
"""Base Role model."""
|
|
STANDARD_USER = 1
|
|
COMMENTS_MODERATOR = 2
|
|
COUNTRY_ADMIN = 3
|
|
CONTENT_PAGE_MANAGER = 4
|
|
ESTABLISHMENT_MANAGER = 5
|
|
REVIEWER_MANGER = 6
|
|
RESTAURANT_REVIEWER = 7
|
|
|
|
ROLE_CHOICES = (
|
|
(STANDARD_USER, 'Standard user'),
|
|
(COMMENTS_MODERATOR, 'Comments moderator'),
|
|
(COUNTRY_ADMIN, 'Country admin'),
|
|
(CONTENT_PAGE_MANAGER, 'Content page manager'),
|
|
(ESTABLISHMENT_MANAGER, 'Establishment manager'),
|
|
(REVIEWER_MANGER, 'Reviewer manager'),
|
|
(RESTAURANT_REVIEWER, 'Restaurant reviewer')
|
|
)
|
|
role = models.PositiveIntegerField(verbose_name=_('Role'), choices=ROLE_CHOICES,
|
|
null=False, blank=False)
|
|
country = models.ForeignKey(Country, verbose_name=_('Country'),
|
|
null=True, blank=True, on_delete=models.SET_NULL)
|
|
# is_list = models.BooleanField(verbose_name=_('list'), default=True, null=False)
|
|
# is_create = models.BooleanField(verbose_name=_('create'), default=False, null=False)
|
|
# is_update = models.BooleanField(verbose_name=_('update'), default=False, null=False)
|
|
# is_delete = models.BooleanField(verbose_name=_('delete'), default=False, null=False)
|
|
|
|
|
|
class UserManager(BaseUserManager):
|
|
"""Extended manager for User model."""
|
|
|
|
use_in_migrations = False
|
|
|
|
def make(self, username: str, email: str, password: str, newsletter: bool) -> object:
|
|
"""Register new user"""
|
|
obj = self.model(
|
|
username=username,
|
|
email=email,
|
|
newsletter=newsletter)
|
|
obj.set_password(password)
|
|
obj.save()
|
|
return obj
|
|
|
|
|
|
class UserQuerySet(models.QuerySet):
|
|
"""Extended queryset for User model."""
|
|
|
|
def active(self, switcher=True):
|
|
"""Filter only active users."""
|
|
return self.filter(is_active=switcher)
|
|
|
|
def by_oauth2_access_token(self, token):
|
|
"""Find users by access token"""
|
|
return self.filter(oauth2_provider_accesstoken__token=token,
|
|
oauth2_provider_accesstoken__expires__gt=timezone.now())
|
|
|
|
def by_oauth2_refresh_token(self, token):
|
|
"""Find users by access token"""
|
|
return self.filter(oauth2_provider_refreshtoken__token=token,
|
|
oauth2_provider_refreshtoken__expires__gt=timezone.now())
|
|
|
|
|
|
class User(AbstractUser):
|
|
"""Base user model."""
|
|
image_url = models.URLField(verbose_name=_('Image URL path'),
|
|
blank=True, null=True, default=None,
|
|
max_length=500)
|
|
cropped_image_url = models.URLField(verbose_name=_('Cropped image URL path'),
|
|
blank=True, null=True, default=None)
|
|
email = models.EmailField(_('email address'), unique=True,
|
|
null=True, default=None)
|
|
unconfirmed_email = models.EmailField(_('unconfirmed email'), blank=True, null=True, default=None)
|
|
email_confirmed = models.BooleanField(_('email status'), default=False)
|
|
newsletter = models.NullBooleanField(default=True)
|
|
old_id = models.IntegerField(null=True, blank=True, default=None)
|
|
|
|
|
|
EMAIL_FIELD = 'email'
|
|
USERNAME_FIELD = 'username'
|
|
REQUIRED_FIELDS = ['email']
|
|
|
|
roles = models.ManyToManyField(Role, verbose_name=_('Roles'), through='UserRole')
|
|
objects = UserManager.from_queryset(UserQuerySet)()
|
|
|
|
class Meta:
|
|
"""Meta class."""
|
|
verbose_name = _('User')
|
|
verbose_name_plural = _('Users')
|
|
|
|
def __str__(self):
|
|
"""String method."""
|
|
return "%s:%s" % (self.email, self.get_short_name())
|
|
|
|
def get_user_info(self):
|
|
"""Get information about user"""
|
|
return {
|
|
"username": self.username,
|
|
"first_name": self.first_name if self.first_name else None,
|
|
"last_name": self.last_name if self.last_name else None,
|
|
"email": self.email if self.email else None,
|
|
"newsletter": self.newsletter,
|
|
"is_active": self.is_active
|
|
}
|
|
|
|
def change_status(self, switcher: bool = False):
|
|
"""Method to set user status to active or inactive"""
|
|
self.is_active = switcher
|
|
self.save()
|
|
|
|
def create_jwt_tokens(self, source: int = None):
|
|
"""Create JWT tokens for user"""
|
|
token = GMRefreshToken.for_user(self, source)
|
|
return {
|
|
'access_token': str(token.access_token),
|
|
'refresh_token': str(token),
|
|
}
|
|
|
|
def expire_access_tokens(self):
|
|
"""Expire all access tokens"""
|
|
self.access_tokens.update(expires_at=timezone.now())
|
|
|
|
def expire_refresh_tokens(self):
|
|
"""Expire all refresh tokens"""
|
|
self.refresh_tokens.update(expires_at=timezone.now())
|
|
|
|
def confirm_email(self):
|
|
"""Method to confirm user email address"""
|
|
if self.unconfirmed_email is not None:
|
|
self.email = self.unconfirmed_email
|
|
self.unconfirmed_email = None
|
|
self.email_confirmed = True
|
|
self.save()
|
|
|
|
def approve(self):
|
|
"""Set user is_active status to True"""
|
|
self.is_active = True
|
|
self.save()
|
|
|
|
def get_body_email_message(self, subject: str, message: str, emails=None):
|
|
"""Prepare the body of the email message"""
|
|
return {
|
|
'subject': subject,
|
|
'message': str(message[0]),
|
|
'html_message': message[1],
|
|
'from_email': settings.EMAIL_HOST_USER,
|
|
'recipient_list': emails if emails else [self.email, ],
|
|
}
|
|
|
|
def send_email(self, subject: str, message: str, emails=None):
|
|
"""Send an email to reset user password"""
|
|
send_mail(**self.get_body_email_message(subject=subject,
|
|
message=message,
|
|
emails=emails))
|
|
|
|
@property
|
|
def confirm_email_token(self):
|
|
"""Make a token for finish signup."""
|
|
return GMTokenGenerator(purpose=GMTokenGenerator.CONFIRM_EMAIL).make_token(self)
|
|
|
|
@property
|
|
def change_email_token(self):
|
|
"""Make a token for change email."""
|
|
return GMTokenGenerator(purpose=GMTokenGenerator.CHANGE_EMAIL).make_token(self)
|
|
|
|
@property
|
|
def reset_password_token(self):
|
|
"""Make a token for finish signup."""
|
|
return password_token_generator.make_token(self)
|
|
|
|
@property
|
|
def get_user_uidb64(self):
|
|
"""Get base64 value for user by primary key identifier"""
|
|
return urlsafe_base64_encode(force_bytes(self.pk))
|
|
|
|
def base_template(self, country_code='www', username='', subject=''):
|
|
"""Base email template"""
|
|
socials = SiteSettings.objects.by_country_code(country_code).first()
|
|
return {
|
|
'title': subject,
|
|
'domain_uri': settings.DOMAIN_URI,
|
|
'uidb64': self.get_user_uidb64,
|
|
'site_name': settings.SITE_NAME,
|
|
'year': datetime.now().year,
|
|
'twitter_page_url': socials.twitter_page_url if socials else '#',
|
|
'instagram_page_url': socials.instagram_page_url if socials else '#',
|
|
'facebook_page_url': socials.facebook_page_url if socials else '#',
|
|
'send_to': username,
|
|
}
|
|
|
|
@property
|
|
def image_tag(self):
|
|
return mark_safe(f'<img src="{self.image_url}" />')
|
|
|
|
@property
|
|
def cropped_image_tag(self):
|
|
return mark_safe(f'<img src="{self.cropped_image_url}" />')
|
|
|
|
def reset_password_template(self, country_code, username, subject):
|
|
"""Get reset password template"""
|
|
context = {'token': self.reset_password_token,
|
|
'country_code': country_code}
|
|
context.update(self.base_template(country_code, username, subject))
|
|
return render_to_string(
|
|
template_name=settings.RESETTING_TOKEN_TEMPLATE,
|
|
context=context), get_template(settings.RESETTING_TOKEN_TEMPLATE).render(context)
|
|
|
|
def notify_password_changed_template(self, country_code, username, subject):
|
|
"""Get notification email template"""
|
|
context = {'contry_code': country_code}
|
|
context.update(self.base_template(country_code, username, subject))
|
|
return render_to_string(
|
|
template_name=settings.NOTIFICATION_PASSWORD_TEMPLATE,
|
|
context=context,
|
|
), get_template(settings.NOTIFICATION_PASSWORD_TEMPLATE).render(context)
|
|
|
|
def confirm_email_template(self, country_code, username, subject):
|
|
"""Get confirm email template"""
|
|
context = {'token': self.confirm_email_token,
|
|
'country_code': country_code}
|
|
context.update(self.base_template(country_code, username, subject))
|
|
return render_to_string(
|
|
template_name=settings.CONFIRM_EMAIL_TEMPLATE,
|
|
context=context), get_template(settings.CONFIRM_EMAIL_TEMPLATE).render(context)
|
|
|
|
def change_email_template(self, country_code, username, subject):
|
|
"""Get change email template"""
|
|
context = {'token': self.change_email_token,
|
|
'country_code': country_code}
|
|
context.update(self.base_template(country_code, username, subject))
|
|
return render_to_string(
|
|
template_name=settings.CHANGE_EMAIL_TEMPLATE,
|
|
context=context), get_template(settings.CHANGE_EMAIL_TEMPLATE).render(context)
|
|
|
|
@property
|
|
def favorite_establishment_ids(self):
|
|
"""Return establishment IDs that in favorites for current user."""
|
|
return self.favorites.by_content_type(app_label='establishment',
|
|
model='establishment')\
|
|
.values_list('object_id', flat=True)
|
|
|
|
@property
|
|
def favorite_recipe_ids(self):
|
|
"""Return recipe IDs that in favorites for current user."""
|
|
return self.favorites.by_content_type(app_label='recipe',
|
|
model='recipe')\
|
|
.values_list('object_id', flat=True)
|
|
|
|
|
|
class UserRole(ProjectBaseMixin):
|
|
"""UserRole model."""
|
|
user = models.ForeignKey(User, verbose_name=_('User'), on_delete=models.CASCADE)
|
|
role = models.ForeignKey(Role, verbose_name=_('Role'), on_delete=models.SET_NULL, null=True)
|
|
establishment = models.ForeignKey(Establishment, verbose_name=_('Establishment'),
|
|
on_delete=models.SET_NULL, null=True, blank=True) |