Misago/misago/threads/views/admin/attachments.py

from django.contrib import messages
from django.db import transaction
from django.utils.translation import gettext_lazy as _

from misago.admin.views import generic
from misago.threads.forms import SearchAttachmentsForm
from misago.threads.models import Attachment, Post


class AttachmentAdmin(generic.AdminBaseMixin):
    root_link = 'misago:admin:system:attachments:index'
    model = Attachment
    templates_dir = 'misago/admin/attachments'
    message_404 = _("Requested attachment could not be found.")

    def get_queryset(self):
        qs = super().get_queryset()
        return qs.select_related('filetype', 'uploader', 'post', 'post__thread', 'post__category')


class AttachmentsList(AttachmentAdmin, generic.ListView):
    items_per_page = 20
    ordering = [
        ('-id', _("From newest")),
        ('id', _("From oldest")),
        ('filename', _("A to z")),
        ('-filename', _("Z to a")),
        ('size', _("Smallest files")),
        ('-size', _("Largest files")),
    ]
    selection_label = _('With attachments: 0')
    empty_selection_label = _('Select attachments')
    mass_actions = [
        {
            'action': 'delete',
            'name': _("Delete attachments"),
            'icon': 'fa fa-times-circle',
            'confirmation': _("Are you sure you want to delete selected attachments?"),
            'is_atomic': False,
        },
    ]

    def get_search_form(self, request):
        return SearchAttachmentsForm

    def action_delete(self, request, attachments):
        deleted_attachments = []
        desynced_posts = []

        for attachment in attachments:
            if attachment.post:
                deleted_attachments.append(attachment.pk)
                desynced_posts.append(attachment.post_id)

        if desynced_posts:
            with transaction.atomic():
                for post in Post.objects.filter(id__in=desynced_posts):
                    self.delete_from_cache(post, deleted_attachments)

        for attachment in attachments:
            attachment.delete()

        message = _("Selected attachments have been deleted.")
        messages.success(request, message)

    def delete_from_cache(self, post, attachments):
        if not post.attachments_cache:
            return  # admin action may be taken due to desynced state

        clean_cache = []
        for a in post.attachments_cache:
            if a['id'] not in attachments:
                clean_cache.append(a)

        post.attachments_cache = clean_cache or None
        post.save(update_fields=['attachments_cache'])


class DeleteAttachment(AttachmentAdmin, generic.ButtonView):
    def button_action(self, request, target):
        if target.post:
            self.delete_from_cache(target)
        target.delete()
        message = _('Attachment "%(filename)s" has been deleted.')
        messages.success(request, message % {'filename': target.filename})

    def delete_from_cache(self, attachment):
        if not attachment.post.attachments_cache:
            return  # admin action may be taken due to desynced state

        clean_cache = []
        for a in attachment.post.attachments_cache:
            if a['id'] != attachment.id:
                clean_cache.append(a)

        attachment.post.attachments_cache = clean_cache or None
        attachment.post.save(update_fields=['attachments_cache'])