extensions-website/extensions/signals.py

from typing import Union
import logging

from actstream.actions import follow, unfollow
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from django.db import transaction
from django.db.models.signals import m2m_changed, pre_save, post_save, pre_delete, post_delete
from django.dispatch import receiver

from constants.activity import Flag
import extensions.models
import files.models

logger = logging.getLogger(__name__)
User = get_user_model()


@receiver(pre_delete, sender=extensions.models.Extension)
@receiver(pre_delete, sender=extensions.models.Preview)
@receiver(pre_delete, sender=extensions.models.Version)
def _log_deletion(
    sender: object,
    instance: Union[
        extensions.models.Extension, extensions.models.Version, extensions.models.Preview
    ],
    **kwargs: object,
) -> None:
    instance.record_deletion()


@receiver(post_delete, sender=extensions.models.VersionFile)
def _delete_versionfiles_file(
    sender: object, instance: extensions.models.VersionFile, **kwargs: object
) -> None:
    # **N.B.**: this isn't part of an overloaded `VersionFile.delete()` method because
    # that method isn't called when `Extension.delete()` cascades to deleting the versions:
    #
    # delete() method for an object is not necessarily called ... as a result of a cascading delete
    # https://docs.djangoproject.com/en/4.2/topics/db/models/#overriding-predefined-model-methods

    file = instance.file
    logger.info('Deleting File pk=%s of VersionFile pk=%s', file.pk, instance.pk)
    file.delete()

    # this code is already quite convoluted :double-facepalm:
    # TODO? maybe find some way to have a predictable order of deletion
    try:
        instance.version.update_platforms()
        if instance.version.files.count() == 0:
            # this was the last file, clean up the version
            logger.info(
                'Deleting Version pk=%s because its last file was deleted',
                instance.version.pk,
            )
            instance.version.delete()
    except extensions.models.Version.DoesNotExist:
        pass


@receiver(pre_save, sender=extensions.models.Extension)
def _record_changes(
    sender: object,
    instance: extensions.models.Extension,
    update_fields: object,
    **kwargs: object,
) -> None:
    was_changed, old_state = instance.pre_save_record(update_fields=update_fields)
    instance.sanitize('name', was_changed, old_state, **kwargs)
    instance.sanitize('description', was_changed, old_state, **kwargs)
    instance.record_status_change(was_changed, old_state, **kwargs)


# TODO? move this out into version.approve that would take care of updating file.status and
# recomputing extension's is_listed and latest_version fields
@receiver(post_save, sender=files.models.File)
def _update_version(
    sender: object,
    instance: files.models.File,
    raw: bool,
    *args: object,
    **kwargs: object,
) -> None:
    if raw:
        return

    version = instance.version.first()
    if version:
        # TODO double-check if not just version.extension
        extension = version.extension
        with transaction.atomic():
            # it's important to update is_listed before computing latest_version
            # because latest_version for listed and unlisted extensions are defined differently
            extension.update_is_listed()
            extension.update_latest_version()


@receiver(post_save, sender=extensions.models.Extension)
def _set_is_listed(
    sender: object,
    instance: extensions.models.Extension,
    raw: bool,
    *args: object,
    **kwargs: object,
) -> None:
    if raw:
        return

    instance.update_is_listed()


@receiver(post_save, sender=extensions.models.Extension)
def _setup_followers(
    sender: object,
    instance: extensions.models.Extension,
    created: bool,
    **kwargs: object,
) -> None:
    if not created:
        return

    for user in instance.authors.all():
        follow(user, instance, send_action=False, flag=Flag.AUTHOR)
    for user in Group.objects.get(name='moderators').user_set.all():
        follow(user, instance, send_action=False, flag=Flag.MODERATOR)


@receiver(m2m_changed, sender=extensions.models.Extension.authors.through)
def _update_authors_follow(instance, action, model, reverse, pk_set, **kwargs):
    if action != 'post_add':
        return

    if model == extensions.models.Extension and not reverse:
        targets = extensions.models.Extension.objects.filter(pk__in=pk_set)
        users = [instance]
    else:
        targets = [instance]
        users = User.objects.filter(pk__in=pk_set)

    for user in users:
        for extension in targets:
            follow(user, extension, send_action=False, flag=Flag.AUTHOR)


# used by admin, the above m2m_changed is not sufficient
@receiver(post_save, sender=extensions.models.Maintainer)
def _add_author_follow(
    sender: object,
    instance: extensions.models.Maintainer,
    created: bool,
    raw: bool,
    **kwargs: object,
):
    if raw:
        return
    if not created:
        return
    follow(instance.user, instance.extension, send_action=False, flag=Flag.AUTHOR)


@receiver(post_delete, sender=extensions.models.Maintainer)
def _add_author_unfollow(
    sender: object,
    instance: extensions.models.Maintainer,
    **kwargs: object,
):
    unfollow(instance.user, instance.extension, send_action=False, flag=Flag.AUTHOR)


@receiver(post_save, sender=extensions.models.Preview)
def _auto_approve_subsequent_uploads(
    sender: object,
    instance: extensions.models.Preview,
    created: bool,
    raw: bool,
    **kwargs: object,
):
    if raw:
        return
    if not created:
        return
    if not instance.file_id:
        return

    # N.B.: currently, subsequent preview uploads get approved automatically,
    # if extension is currently listed (meaning, it was approved by a human already).
    extension = instance.extension
    file = instance.file
    if extension.is_listed:
        file.status = files.models.File.STATUSES.APPROVED
        args = {'f_id': file.pk, 'pk': instance.pk, 'sender': sender, 's': file.source.name}
        logger.info('Auto-approving file pk=%(f_id)s of %(sender)s pk=%(pk)s source=%(s)s', args)
        file.save(update_fields={'status', 'date_modified'})