"""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( '' ) return format_html( '' ) 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( 'Активен' ) return format_html( 'Неактивен' ) 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)