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.Version)
def _delete_version_file(
    sender: object, instance: extensions.models.Version, **kwargs: object
) -> None:
    # **N.B.**: this isn't part of an overloaded `Version.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
    version_file = instance.file
    logger.info('Deleting file pk=%s of Version pk=%s', version_file.pk, instance.pk)
    version_file.delete()


@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

    if hasattr(instance, 'version'):
        extension = instance.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 not in ['post_add', 'post_remove']:
        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:
            if action == 'post_remove':
                unfollow(user, extension, send_action=False, flag=Flag.AUTHOR)
            elif action == 'post_add':
                follow(user, 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'})