state-corp-backend/src/apps/user/admin.py

"""Admin configuration for user app."""

from contextlib import suppress
from functools import wraps

from apps.core.exceptions import BaseAPIException
from apps.core.models import BackgroundJob
from apps.user.management import lock_managed_users, validate_access_change
from apps.user.models import Profile, User
from apps.user.services import UserService
from django.contrib import admin, messages
from django.contrib.admin.sites import NotRegistered
from django.contrib.admin.utils import unquote
from django.contrib.auth.admin import UserAdmin as BaseUserAdmin
from django.contrib.auth.models import Group
from django.db import transaction
from django.http import HttpResponseRedirect
from django.urls import reverse
from django.utils.html import format_html
from django.utils.translation import gettext_lazy as _

try:
    from django_celery_beat.models import (
        ClockedSchedule,
        CrontabSchedule,
        IntervalSchedule,
        PeriodicTask,
        SolarSchedule,
    )
except ImportError:  # pragma: no cover - optional dependency import guard
    ClockedSchedule = None
    CrontabSchedule = None
    IntervalSchedule = None
    PeriodicTask = None
    SolarSchedule = None

try:
    from rest_framework_simplejwt.token_blacklist.models import (
        BlacklistedToken,
        OutstandingToken,
    )
except ImportError:  # pragma: no cover - optional dependency import guard
    BlacklistedToken = None
    OutstandingToken = None


def _unregister(model) -> None:
    if model is None:
        return

    with suppress(NotRegistered):
        admin.site.unregister(model)


_unregister(User)


def _management_errors(view):
    """Present guarded account-change failures through Django admin messages."""

    @wraps(view)
    def wrapped(self, request, *args, **kwargs):
        try:
            return view(self, request, *args, **kwargs)
        except BaseAPIException as exc:
            self.message_user(request, exc.message, level=messages.ERROR)
            return HttpResponseRedirect(reverse("admin:user_user_changelist"))

    return wrapped


class ProfileInline(admin.StackedInline):
    """Inline для профиля пользователя."""

    model = Profile
    extra = 0
    max_num = 1
    can_delete = False
    verbose_name_plural = "Профиль"
    fk_name = "user"
    fields = ["first_name", "mid_name", "last_name", "bio", "avatar", "date_of_birth"]


@admin.register(User)
class UserAdmin(BaseUserAdmin):
    """Admin для пользователей без групп."""

    inlines = [ProfileInline]
    list_display = [
        "username",
        "email",
        "phone",
        "is_verified_badge",
        "is_active_badge",
        "is_staff",
        "created_at",
    ]
    list_filter = ["is_staff", "is_superuser", "is_active", "is_verified", "created_at"]
    search_fields = ["username", "email", "phone"]
    ordering = ["-created_at"]
    list_per_page = 50
    date_hierarchy = "created_at"

    fieldsets = (
        (None, {"fields": ("username", "password")}),
        (_("Personal info"), {"fields": ("email", "phone")}),
        (
            _("Permissions"),
            {
                "fields": (
                    "is_active",
                    "is_staff",
                    "is_superuser",
                    "is_verified",
                    "user_permissions",
                ),
                "classes": ("collapse",),
            },
        ),
        (
            _("Important dates"),
            {"fields": ("last_login", "date_joined", "created_at", "updated_at")},
        ),
    )

    add_fieldsets = (
        (
            None,
            {
                "classes": ("wide",),
                "fields": (
                    "username",
                    "email",
                    "password1",
                    "password2",
                    "is_staff",
                    "is_active",
                ),
            },
        ),
    )

    readonly_fields = ["created_at", "updated_at", "last_login", "date_joined"]
    actions = ["verify_users", "unverify_users", "activate_users", "deactivate_users"]

    @_management_errors
    def changeform_view(self, request, object_id=None, form_url="", extra_context=None):
        return super().changeform_view(request, object_id, form_url, extra_context)

    @_management_errors
    def delete_view(self, request, object_id, extra_context=None):
        return super().delete_view(request, object_id, extra_context)

    @_management_errors
    def user_change_password(self, request, id, form_url=""):
        user = self.get_object(request, unquote(id))
        if request.method == "POST" and user is not None:
            with lock_managed_users(actor_id=request.user.pk, user_ids=[user.pk]):
                return super().user_change_password(request, id, form_url)
        return super().user_change_password(request, id, form_url)

    @_management_errors
    def changelist_view(self, request, extra_context=None):
        # delete_selected writes LogEntry rows before delete_queryset. Roll those
        # back too if the complete selection fails an account-management guard.
        with transaction.atomic():
            return super().changelist_view(request, extra_context)

    def save_model(self, request, obj, form, change):
        with lock_managed_users(
            actor_id=request.user.pk, user_ids=[obj.pk] if change else []
        ) as users:
            if change:
                validate_access_change(
                    users[obj.pk],
                    actor_id=request.user.pk,
                    is_active=obj.is_active,
                    is_staff=obj.is_staff,
                )
            super().save_model(request, obj, form, change)

    def delete_model(self, request, obj):
        UserService.delete_user(obj.pk, actor_id=request.user.pk)

    def delete_queryset(self, request, queryset):
        UserService.delete_users(
            queryset.values_list("pk", flat=True), actor_id=request.user.pk
        )

    def is_verified_badge(self, obj):
        if obj.is_verified:
            return format_html(
                '<span style="color: white; background: #28a745; padding: 3px 10px; border-radius: 3px;">✓</span>'
            )
        return format_html(
            '<span style="color: white; background: #dc3545; padding: 3px 10px; border-radius: 3px;">✗</span>'
        )

    is_verified_badge.short_description = "Верифицирован"
    is_verified_badge.admin_order_field = "is_verified"

    def is_active_badge(self, obj):
        if obj.is_active:
            return format_html(
                '<span style="color: white; background: #28a745; padding: 3px 10px; border-radius: 3px;">Активен</span>'
            )
        return format_html(
            '<span style="color: white; background: #dc3545; padding: 3px 10px; border-radius: 3px;">Неактивен</span>'
        )

    is_active_badge.short_description = "Статус"
    is_active_badge.admin_order_field = "is_active"

    @admin.action(description="Верифицировать выбранных пользователей")
    def verify_users(self, request, queryset):
        updated = queryset.update(is_verified=True)
        self.message_user(request, f"Верифицировано {updated} пользователей")

    @admin.action(description="Снять верификацию")
    def unverify_users(self, request, queryset):
        updated = queryset.update(is_verified=False)
        self.message_user(request, f"Снята верификация у {updated} пользователей")

    @admin.action(description="Активировать пользователей")
    def activate_users(self, request, queryset):
        updated = UserService.set_users_active(
            queryset.values_list("pk", flat=True),
            actor_id=request.user.pk,
            is_active=True,
        )
        self.message_user(request, f"Активировано {updated} пользователей")

    @admin.action(description="Деактивировать пользователей")
    def deactivate_users(self, request, queryset):
        updated = UserService.set_users_active(
            queryset.values_list("pk", flat=True),
            actor_id=request.user.pk,
            is_active=False,
        )
        self.message_user(request, f"Деактивировано {updated} пользователей")


for model in (
    Group,
    BackgroundJob,
    OutstandingToken,
    BlacklistedToken,
    PeriodicTask,
    CrontabSchedule,
    IntervalSchedule,
    SolarSchedule,
    ClockedSchedule,
):
    _unregister(model)