142 lines
4.6 KiB
Python
142 lines
4.6 KiB
Python
from pathlib import Path
|
|
from typing import Optional
|
|
import logging
|
|
import time
|
|
|
|
from django.contrib.admin.utils import NestedObjects
|
|
from django.contrib.auth.models import AbstractUser, Group
|
|
from django.db import models, DEFAULT_DB_ALIAS, transaction
|
|
from django.templatetags.static import static
|
|
|
|
from common.model_mixins import TrackChangesMixin
|
|
from files.utils import get_sha256_from_value
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def user_image_upload_to(instance, filename):
|
|
assert instance.pk
|
|
prefix = 'avatars/'
|
|
_hash = get_sha256_from_value(instance.pk)
|
|
extension = Path(filename).suffix
|
|
path = Path(prefix, _hash[:2], _hash).with_suffix(extension)
|
|
return path
|
|
|
|
|
|
def shortuid() -> str:
|
|
"""Generate a 14-characters long string ID based on time."""
|
|
return hex(int(time.monotonic() * 10**10))[2:]
|
|
|
|
|
|
class User(TrackChangesMixin, AbstractUser):
|
|
track_changes_to_fields = {
|
|
'is_superuser',
|
|
'is_staff',
|
|
'date_deletion_requested',
|
|
'confirmed_email_at',
|
|
'full_name',
|
|
'email',
|
|
}
|
|
|
|
class Meta:
|
|
db_table = 'auth_user'
|
|
|
|
email = models.EmailField(blank=False, null=False, unique=True)
|
|
full_name = models.CharField(max_length=255, blank=True, default='')
|
|
image = models.ImageField(upload_to=user_image_upload_to, blank=True, null=True)
|
|
badges = models.JSONField(null=True, blank=True)
|
|
|
|
date_deletion_requested = models.DateTimeField(null=True, blank=True)
|
|
confirmed_email_at = models.DateTimeField(null=True, blank=True)
|
|
|
|
def __str__(self) -> str:
|
|
return f'{self.full_name or self.username}'
|
|
|
|
@property
|
|
def image_url(self) -> Optional[str]:
|
|
"""Return a URL of the Profile image."""
|
|
if not self.image:
|
|
return static('common/images/blank-profile-pic.png')
|
|
|
|
return self.image.url
|
|
|
|
def _get_nested_objects_collector(self) -> NestedObjects:
|
|
collector = NestedObjects(using=DEFAULT_DB_ALIAS)
|
|
collector.collect([self])
|
|
return collector
|
|
|
|
@property
|
|
def can_be_deleted(self) -> bool:
|
|
"""Fetch objects referencing this profile and determine if it can be deleted."""
|
|
if self.is_staff or self.is_superuser:
|
|
return False
|
|
collector = self._get_nested_objects_collector()
|
|
if collector.protected:
|
|
return False
|
|
return True
|
|
|
|
def request_deletion(self, date_deletion_requested):
|
|
"""Store date of the deletion request and deactivate the user."""
|
|
if not self.can_be_deleted:
|
|
logger.error('Deletion requested for a protected account pk=%s, ignoring', self.pk)
|
|
return
|
|
logger.warning(
|
|
'Deletion of pk=%s requested on %s, deactivating this account',
|
|
self.pk,
|
|
date_deletion_requested,
|
|
)
|
|
self.is_active = False
|
|
self.date_deletion_requested = date_deletion_requested
|
|
self.save(update_fields=['is_active', 'date_deletion_requested'])
|
|
|
|
@transaction.atomic
|
|
def anonymize_or_delete(self):
|
|
"""Delete user completely if they don't have public records, otherwise anonymize.
|
|
|
|
Does nothing if deletion hasn't been explicitly requested earlier.
|
|
"""
|
|
if not self.can_be_deleted:
|
|
logger.error(
|
|
'User.anonymize called, but pk=%s cannot be deleted',
|
|
self.pk,
|
|
)
|
|
return
|
|
|
|
if not self.date_deletion_requested:
|
|
logger.error(
|
|
"User.anonymize_or_delete called, but deletion of pk=%s hasn't been requested",
|
|
self.pk,
|
|
)
|
|
return
|
|
|
|
if self.image:
|
|
try:
|
|
self.image.delete(save=False)
|
|
except Exception:
|
|
logger.exception(
|
|
'Unable to delete image %s for pk=%s',
|
|
self.image.name,
|
|
self.pk,
|
|
)
|
|
|
|
# Simply delete the user if there are no publicly listed records linked to it
|
|
if self.extensions.listed.count() == 0 and self.files.listed.count() == 0:
|
|
logger.warning('Deleted user pk=%s', self.pk)
|
|
self.delete()
|
|
else:
|
|
username = f'del{shortuid()}'
|
|
self.__class__.objects.filter(pk=self.pk).update(
|
|
email=f'{username}@example.com',
|
|
full_name='',
|
|
username=username,
|
|
badges=None,
|
|
is_active=False,
|
|
image=None,
|
|
)
|
|
logger.warning('Anonymized user pk=%s', self.pk)
|
|
|
|
@property
|
|
def is_moderator(self):
|
|
# Used to review and approve extensions
|
|
return self.groups.filter(name='moderators').exists()
|