Misago/misago/users/signals.py

from collections import OrderedDict
from datetime import timedelta

from django.contrib.auth import get_user_model
from django.db.models import Q
from django.dispatch import Signal, receiver
from django.utils import timezone
from django.utils.translation import gettext as _

from misago.conf import settings
from misago.core.pgutils import chunk_queryset

from .models import AuditTrail
from .profilefields import profilefields


UserModel = get_user_model()

anonymize_user_data = Signal()
archive_user_data = Signal()
delete_user_content = Signal()
remove_old_ips = Signal()
username_changed = Signal()


@receiver(archive_user_data)
def archive_user_details(sender, archive=None, **kwargs):
    archive.add_dict('details', OrderedDict([
        (_('Username'), sender.username),
        (_('E-mail'), sender.email),
        (_('Joined on'), sender.joined_on),
        (_('Joined from ip'), sender.joined_from_ip or 'unavailable'),
    ]))


@receiver(archive_user_data)
def archive_user_profile_fields(sender, archive=None, **kwargs):
    clean_profile_fields = OrderedDict()
    for profile_fields_group in profilefields.get_fields_groups():
        for profile_field in profile_fields_group['fields']:
            if sender.profile_fields.get(profile_field.fieldname):
                field_value = sender.profile_fields[profile_field.fieldname]
                clean_profile_fields[str(profile_field.label)] = field_value
                
    if clean_profile_fields:
        archive.add_dict('profile_fields', clean_profile_fields)


@receiver(archive_user_data)
def archive_user_avatar(sender, archive=None, **kwargs):
    archive.add_model_file(sender.avatar_tmp, directory='avatar', prefix='tmp')
    archive.add_model_file(sender.avatar_src, directory='avatar', prefix='src')
    for avatar in sender.avatar_set.iterator():
        archive.add_model_file(avatar.image, directory='avatar', prefix=avatar.size)


@receiver(archive_user_data)
def archive_user_audit_trail(sender, archive=None, **kwargs):
    queryset = sender.audittrail_set.order_by('id')
    for audit_trail in chunk_queryset(queryset):
        item_name = audit_trail.created_on.strftime('%H%M%S-audit-trail')
        archive.add_text(item_name, audit_trail.ip_address, date=audit_trail.created_on)


@receiver(archive_user_data)
def archive_user_name_history(sender, archive=None, **kwargs):
    for name_change in sender.namechanges.order_by('id').iterator():
        item_name = name_change.changed_on.strftime('%H%M%S-name-change')
        archive.add_dict(
            item_name,
            OrderedDict([
                (_("New username"), name_change.new_username),
                (_("Old username"), name_change.old_username),
            ]),
            date=name_change.changed_on,
        )


@receiver(username_changed)
def handle_name_change(sender, **kwargs):
    sender.user_renames.update(changed_by_username=sender.username)


@receiver(remove_old_ips)
def remove_old_registrations_ips(sender, **kwargs):
    datetime_cutoff = timezone.now() - timedelta(days=settings.MISAGO_IP_STORE_TIME)
    ip_is_too_new = Q(joined_on__gt=datetime_cutoff)
    ip_is_already_removed = Q(joined_from_ip__isnull=True)
    
    queryset = UserModel.objects.exclude(ip_is_too_new | ip_is_already_removed)
    queryset.update(joined_from_ip=None)


@receiver(remove_old_ips)
def remove_old_audit_trails(sender, **kwargs):
    removal_cutoff = timezone.now() - timedelta(days=settings.MISAGO_IP_STORE_TIME)
    AuditTrail.objects.filter(created_on__lte=removal_cutoff).delete()