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.models.signals import m2m_changed, pre_save, post_save, pre_delete, post_delete
from django.dispatch import receiver

from constants.activity import Flag
from reviewers.models import ApprovalActivity
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()


def _delete_file(f, sender, instance, rel):
    source = f.source.name
    args = {'f_id': f.pk, 'h': f.hash, 'pk': instance.pk, 'sender': sender, 's': source, 'r': rel}
    logger.info('Deleting %(r)s file pk=%(f_id)s s=%(s)s hash=%(h)s of %(sender)s pk=%(pk)s', args)
    f.delete()


@receiver(post_delete, sender=extensions.models.Preview)
@receiver(post_delete, sender=extensions.models.Version)
def _delete_preview_or_version_file(sender: object, instance: object, **kwargs: object) -> None:
    f = instance.file
    _delete_file(f, sender, instance, rel=sender)


@receiver(post_delete, sender=extensions.models.Extension)
def _delete_featured_image_and_icon(sender: object, instance: object, **kwargs: object) -> None:
    for rel in ('featured_image', 'icon'):
        f = getattr(instance, rel)
        if not f:
            continue
        _delete_file(f, sender, instance, rel)


@receiver(pre_save, sender=extensions.models.Extension)
@receiver(pre_save, sender=extensions.models.Version)
def _record_changes(
    sender: object,
    instance: Union[extensions.models.Extension, extensions.models.Version],
    update_fields: object,
    **kwargs: object,
) -> None:
    was_changed, old_state = instance.pre_save_record(update_fields=update_fields)

    if hasattr(instance, 'name'):
        instance.sanitize('name', was_changed, old_state, **kwargs)
    if hasattr(instance, 'description'):
        instance.sanitize('description', was_changed, old_state, **kwargs)

    instance.record_status_change(was_changed, old_state, **kwargs)


@receiver(post_save, sender=extensions.models.Extension)
def _update_search_index(sender, instance, **kw):
    pass  # TODO: update search index


def extension_should_be_listed(extension):
    return (
        extension.latest_version is not None
        and extension.latest_version.is_listed
        and extension.status == extension.STATUSES.APPROVED
    )


@receiver(post_save, sender=extensions.models.Extension)
@receiver(post_save, sender=extensions.models.Version)
@receiver(post_save, sender=files.models.File)
def _set_is_listed(
    sender: object,
    instance: Union[extensions.models.Extension, extensions.models.Version, files.models.File],
    *args: object,
    **kwargs: object,
) -> None:
    if isinstance(instance, extensions.models.Extension):
        extension = instance
    elif isinstance(instance, extensions.models.Version):
        extension = instance.extension
    else:
        # Some file types (e.g., image or video) have no version associated to them.
        # But also files which were created but have not yet being related to the versions.
        # Since signals is called very early on, we can't assume file.extension will be available.
        if not hasattr(instance, 'version'):
            return
        extension = instance.extension

    old_is_listed = extension.is_listed
    new_is_listed = extension_should_be_listed(extension)

    if old_is_listed == new_is_listed:
        return

    if extension.status == extensions.models.Extension.STATUSES.APPROVED and not new_is_listed:
        extension.status = extensions.models.Extension.STATUSES.INCOMPLETE

    logger.info('Extension pk=%s becomes listed', extension.pk)
    extension.is_listed = new_is_listed
    extension.save()


@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)
@receiver(post_save, sender=extensions.models.Version)
def _auto_approve_subsequent_uploads(
    sender: object,
    instance: Union[extensions.models.Preview, extensions.models.Version],
    created: bool,
    raw: bool,
    **kwargs: object,
):
    if raw:
        return
    if not created:
        return
    if not instance.file_id:
        return

    # N.B.: currently, subsequent version and 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'})


@receiver(post_save, sender=extensions.models.Version)
def _create_approval_activity_for_new_version_if_listed(
    sender: object,
    instance: extensions.models.Version,
    created: bool,
    raw: bool,
    **kwargs: object,
):
    if raw:
        return
    if not created:
        return
    extension = instance.extension
    if not extension.is_listed or not instance.file:
        return
    ApprovalActivity(
        type=ApprovalActivity.ActivityType.UPLOADED_NEW_VERSION,
        user=instance.file.user,
        extension=instance.extension,
        message=f'uploaded new version: {instance.version}',
    ).save()