Misago/misago/users/views/admin/users.py

from django.contrib import messages
from django.contrib.auth import get_user_model, update_session_auth_hash
from django.db import transaction
from django.http import JsonResponse
from django.shortcuts import redirect
from django.utils.translation import gettext_lazy as _

from ....acl.useracl import get_user_acl
from ....admin.auth import authorize_admin
from ....admin.views import generic
from ....categories.models import Category
from ....core.mail import mail_users
from ....core.pgutils import chunk_queryset
from ....threads.models import Thread
from ...avatars.dynamic import set_avatar as set_dynamic_avatar
from ...datadownloads import request_user_data_download, user_has_data_download_request
from ...forms.admin import (
    BanUsersForm,
    EditUserForm,
    EditUserFormFactory,
    NewUserForm,
    create_search_users_form,
)
from ...models import Ban
from ...profilefields import profilefields
from ...setupnewuser import setup_new_user
from ...signatures import set_user_signature

User = get_user_model()


class UserAdmin(generic.AdminBaseMixin):
    root_link = "misago:admin:users:accounts:index"
    templates_dir = "misago/admin/users"
    model = User

    def create_form_type(self, request, target):
        add_is_active_fields = False
        add_admin_fields = False

        if not target.is_deleting_account:
            if not target.is_staff:
                add_is_active_fields = True
            elif request.user.is_superuser:
                add_is_active_fields = request.user.pk != target.pk

        if request.user.is_superuser:
            add_admin_fields = request.user.pk != target.pk

        return EditUserFormFactory(
            self.form,
            target,
            add_is_active_fields=add_is_active_fields,
            add_admin_fields=add_admin_fields,
        )


class UsersList(UserAdmin, generic.ListView):
    items_per_page = 24
    ordering = [
        ("-id", _("From newest")),
        ("id", _("From oldest")),
        ("slug", _("A to z")),
        ("-slug", _("Z to a")),
        ("posts", _("Biggest posters")),
        ("-posts", _("Smallest posters")),
    ]
    selection_label = _("With users: 0")
    empty_selection_label = _("Select users")
    mass_actions = [
        {
            "action": "activate",
            "name": _("Activate accounts"),
            "icon": "fa fa-check-square-o",
        },
        {"action": "ban", "name": _("Ban users"), "icon": "fa fa-lock"},
        {
            "action": "request_data_download",
            "name": _("Request data download"),
            "icon": "fa fa-download",
        },
        {
            "action": "delete_accounts",
            "name": _("Delete accounts"),
            "icon": "fa fa-times-circle",
            "confirmation": _("Are you sure you want to delete selected users?"),
        },
        {
            "action": "delete_all",
            "name": _("Delete all"),
            "icon": "fa fa-eraser",
            "confirmation": _(
                "Are you sure you want to delete selected users? "
                "This will also delete all content associated with their accounts."
            ),
            "is_atomic": False,
        },
    ]

    def get_queryset(self):
        qs = super().get_queryset()
        return qs.select_related("rank")

    def get_search_form(self, request):
        return create_search_users_form()

    def action_activate(self, request, users):
        inactive_users = []
        for user in users:
            if user.requires_activation:
                inactive_users.append(user)

        if not inactive_users:
            message = _("You have to select inactive users.")
            raise generic.MassActionError(message)
        else:
            activated_users_pks = [u.pk for u in inactive_users]
            queryset = User.objects.filter(pk__in=activated_users_pks)
            queryset.update(requires_activation=User.ACTIVATION_NONE)

            subject = _("Your account on %(forum_name)s forums has been activated")
            mail_subject = subject % {"forum_name": request.settings.forum_name}

            mail_users(
                inactive_users,
                mail_subject,
                "misago/emails/activation/by_admin",
                context={"settings": request.settings},
            )

            messages.success(request, _("Selected users accounts have been activated."))

    def action_ban(
        self, request, users
    ):  # pylint: disable=too-many-locals, too-many-nested-blocks, too-many-branches
        users = users.order_by("slug")
        for user in users:
            if user.is_superuser:
                message = _("%(user)s is super admin and can't be banned.")
                mesage = message % {"user": user.username}
                raise generic.MassActionError(mesage)

        form = BanUsersForm(users=users)
        if "finalize" in request.POST:
            form = BanUsersForm(request.POST, users=users)
            if form.is_valid():
                cleaned_data = form.cleaned_data
                banned_values = []

                ban_kwargs = {
                    "user_message": cleaned_data.get("user_message"),
                    "staff_message": cleaned_data.get("staff_message"),
                    "expires_on": cleaned_data.get("expires_on"),
                }

                for user in users:
                    for ban in cleaned_data["ban_type"]:
                        banned_value = None

                        if ban == "usernames":
                            check_type = Ban.USERNAME
                            banned_value = user.username.lower()

                        if ban == "emails":
                            check_type = Ban.EMAIL
                            banned_value = user.email.lower()

                        if ban == "domains":
                            check_type = Ban.EMAIL
                            banned_value = user.email.lower()
                            at_pos = banned_value.find("@")
                            banned_value = "*%s" % banned_value[at_pos:]

                        if ban == "ip" and user.joined_from_ip:
                            check_type = Ban.IP
                            banned_value = user.joined_from_ip

                        if ban in ("ip_first", "ip_two") and user.joined_from_ip:
                            check_type = Ban.IP

                            if ":" in user.joined_from_ip:
                                ip_separator = ":"
                            if "." in user.joined_from_ip:
                                ip_separator = "."

                            bits = user.joined_from_ip.split(ip_separator)
                            if ban == "ip_first":
                                formats = (bits[0], ip_separator)
                            if ban == "ip_two":
                                formats = (bits[0], ip_separator, bits[1], ip_separator)
                            banned_value = "%s*" % ("".join(formats))

                        if banned_value and banned_value not in banned_values:
                            ban_kwargs.update(
                                {"check_type": check_type, "banned_value": banned_value}
                            )
                            Ban.objects.create(**ban_kwargs)
                            banned_values.append(banned_value)

                Ban.objects.invalidate_cache()
                messages.success(request, _("Selected users have been banned."))
                return None

        return self.render(
            request,
            template="misago/admin/users/ban.html",
            context={"users": users, "form": form},
        )

    def action_request_data_download(self, request, users):
        for user in users:
            if not user_has_data_download_request(user):
                request_user_data_download(user, requester=request.user)

        messages.success(
            request, _("Data download requests have been placed for selected users.")
        )

    def action_delete_accounts(self, request, users):
        for user in users:
            if user == request.user:
                raise generic.MassActionError(_("You can't delete yourself."))
            if user.is_staff or user.is_superuser:
                message = _("%(user)s is admin and can't be deleted.") % {
                    "user": user.username
                }
                raise generic.MassActionError(message)

        for user in users:
            user.delete()

        messages.success(request, _("Selected users have been deleted."))

    def action_delete_all(self, request, users):
        for user in users:
            if user == request.user:
                raise generic.MassActionError(_("You can't delete yourself."))
            if user.is_staff or user.is_superuser:
                message = _("%(user)s is admin and can't be deleted.") % {
                    "user": user.username
                }
                raise generic.MassActionError(message)

        return self.render(
            request, template="misago/admin/users/delete.html", context={"users": users}
        )


class NewUser(UserAdmin, generic.ModelFormView):
    form = NewUserForm
    template = "new.html"
    message_submit = _('New user "%(user)s" has been registered.')

    def initialize_form(self, form, request, target):
        if request.method == "POST":
            return form(request.POST, request.FILES, instance=target, request=request)
        return form(instance=target, request=request)

    def handle_form(self, form, request, target):
        new_user = User.objects.create_user(
            form.cleaned_data["username"],
            form.cleaned_data["email"],
            form.cleaned_data["new_password"],
            title=form.cleaned_data["title"],
            rank=form.cleaned_data.get("rank"),
            joined_from_ip=request.user_ip,
        )

        if form.cleaned_data.get("staff_level"):
            new_user.staff_level = form.cleaned_data["staff_level"]

        if form.cleaned_data.get("roles"):
            new_user.roles.add(*form.cleaned_data["roles"])

        new_user.update_acl_key()
        setup_new_user(request.settings, new_user)

        messages.success(request, self.message_submit % {"user": target.username})
        return redirect("misago:admin:users:accounts:edit", pk=new_user.pk)


class EditUser(UserAdmin, generic.ModelFormView):
    form = EditUserForm
    template = "edit.html"
    message_submit = _('User "%(user)s" has been edited.')

    def real_dispatch(self, request, target):
        target.old_username = target.username
        target.old_is_avatar_locked = target.is_avatar_locked
        return super().real_dispatch(request, target)

    def initialize_form(self, form, request, target):
        if request.method == "POST":
            return form(request.POST, request.FILES, instance=target, request=request)
        return form(instance=target, request=request)

    def handle_form(self, form, request, target):
        target.username = target.old_username
        if target.username != form.cleaned_data.get("username"):
            target.set_username(
                form.cleaned_data.get("username"), changed_by=request.user
            )

        if form.cleaned_data.get("new_password"):
            target.set_password(form.cleaned_data["new_password"])

            if target.pk == request.user.pk:
                authorize_admin(request, target)
                update_session_auth_hash(request, target)

        if form.cleaned_data.get("email"):
            target.set_email(form.cleaned_data["email"])
            if target.pk == request.user.pk:
                authorize_admin(request, target)

        if form.cleaned_data.get("is_avatar_locked"):
            if not target.old_is_avatar_locked:
                set_dynamic_avatar(target)

        if "is_staff" in form.fields and "is_superuser" in form.fields:
            target.is_staff = form.cleaned_data.get("is_staff")
            target.is_superuser = form.cleaned_data.get("is_superuser")

        if "is_active" in form.fields and "is_active_staff_message" in form.fields:
            target.is_active = form.cleaned_data.get("is_active")
            target.is_active_staff_message = form.cleaned_data.get(
                "is_active_staff_message"
            )

        target.rank = form.cleaned_data.get("rank")

        target.roles.clear()
        target.roles.add(*form.cleaned_data["roles"])

        target_acl = get_user_acl(target, request.cache_versions)
        set_user_signature(
            request, target, target_acl, form.cleaned_data.get("signature")
        )

        profilefields.update_user_profile_fields(request, target, form)

        target.update_acl_key()
        target.save()

        messages.success(request, self.message_submit % {"user": target.username})


class DeletionStep(UserAdmin, generic.ButtonView):
    is_atomic = False

    def check_permissions(self, request, target):
        if not request.is_ajax():
            return _("This action can't be accessed directly.")

        if target == request.user:
            return _("You can't delete yourself.")

        if target.is_staff or target.is_superuser:
            return _("%(user)s is admin and can't be deleted.") % {
                "user": target.username
            }

    def execute_step(self, user):
        raise NotImplementedError(
            "execute_step method should return dict with "
            "number of deleted_count and is_completed keys"
        )

    def button_action(self, request, target):
        return JsonResponse(self.execute_step(target))


class DeleteThreadsStep(DeletionStep):
    def execute_step(self, user):
        recount_categories = set()

        deleted_threads = 0
        is_completed = False

        for thread in user.thread_set.order_by("-id")[:50]:
            recount_categories.add(thread.category_id)
            with transaction.atomic():
                thread.delete()
                deleted_threads += 1

        if recount_categories:
            for category in Category.objects.filter(id__in=recount_categories):
                category.synchronize()
                category.save()
        else:
            is_completed = True

        return {"deleted_count": deleted_threads, "is_completed": is_completed}


class DeletePostsStep(DeletionStep):
    def execute_step(self, user):
        recount_categories = set()
        recount_threads = set()

        deleted_posts = 0
        is_completed = False

        for post in user.post_set.order_by("-id")[:50]:
            recount_categories.add(post.category_id)
            recount_threads.add(post.thread_id)
            with transaction.atomic():
                post.delete()
                deleted_posts += 1

        if recount_categories:
            changed_threads_qs = Thread.objects.filter(id__in=recount_threads)
            for thread in chunk_queryset(changed_threads_qs, 50):
                thread.synchronize()
                thread.save()

            for category in Category.objects.filter(id__in=recount_categories):
                category.synchronize()
                category.save()
        else:
            is_completed = True

        return {"deleted_count": deleted_posts, "is_completed": is_completed}


class DeleteAccountStep(DeletionStep):
    def execute_step(self, user):
        user.delete(delete_content=True)
        return {"is_completed": True}