schon/engine/vibes_auth/admin.py

from typing import Any

from django.contrib import admin, messages
from django.contrib.admin import register
from django.contrib.auth.admin import (
    GroupAdmin as BaseGroupAdmin,
)
from django.contrib.auth.admin import (
    UserAdmin as BaseUserAdmin,
)
from django.contrib.auth.models import Group as BaseGroup
from django.contrib.auth.models import Permission
from django.core.exceptions import PermissionDenied
from django.db.models import Prefetch, QuerySet
from django.http import HttpRequest
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from rest_framework_simplejwt.token_blacklist.admin import (
    BlacklistedTokenAdmin as BaseBlacklistedTokenAdmin,
)
from rest_framework_simplejwt.token_blacklist.admin import (
    OutstandingTokenAdmin as BaseOutstandingTokenAdmin,
)
from rest_framework_simplejwt.token_blacklist.models import (
    BlacklistedToken as BaseBlacklistedToken,
)
from rest_framework_simplejwt.token_blacklist.models import (
    OutstandingToken as BaseOutstandingToken,
)
from unfold.admin import ModelAdmin, TabularInline
from unfold.forms import AdminPasswordChangeForm, UserCreationForm
from unfold_markdown import MarkdownWidget

from engine.core.admin import ActivationActionsMixin
from engine.core.models import Order
from engine.payments.models import Balance
from engine.vibes_auth.emailing.choices import CampaignStatus
from engine.vibes_auth.emailing.models import (
    CampaignRecipient,
    EmailCampaign,
    EmailTemplate,
)
from engine.vibes_auth.emailing.tasks import (
    prepare_campaign_recipients,
    send_campaign_email_batch,
)
from engine.vibes_auth.forms import UserForm
from engine.vibes_auth.models import (
    AdminOTPCode,
    BlacklistedToken,
    ChatMessage,
    ChatThread,
    Group,
    OutstandingToken,
    ThreadStatus,
    User,
)


class BalanceInline(TabularInline):
    model = Balance
    can_delete = False
    extra = 0
    verbose_name = _("balance")
    verbose_name_plural = _("balance")
    is_navtab = True
    icon = "fa-solid fa-wallet"


class OrderInline(TabularInline):
    model = Order
    extra = 0
    verbose_name = _("order")
    verbose_name_plural = _("orders")
    is_navtab = True
    icon = "fa-solid fa-cart-shopping"


class UserAdmin(ActivationActionsMixin, BaseUserAdmin, ModelAdmin):
    inlines = (BalanceInline, OrderInline)
    fieldsets = (
        (None, {"fields": ("email", "password")}),
        (
            _("personal info"),
            {"fields": ("first_name", "last_name", "phone_number", "avatar")},
        ),
        (
            _("permissions"),
            {
                "fields": (
                    "is_active",
                    "is_verified",
                    "is_subscribed",
                    "is_staff",
                    "is_superuser",
                    "groups",
                    "user_permissions",
                )
            },
        ),
        (_("important dates"), {"fields": ("last_login", "date_joined")}),
        (_("additional info"), {"fields": ("language", "attributes")}),
    )
    add_fieldsets = (
        (
            None,
            {
                "classes": ("wide",),
                "fields": ("email", "password1", "password2"),
            },
        ),
    )
    list_display = ("email", "phone_number", "is_verified", "is_active", "is_staff")
    # phone_number is encrypted — DB-level search is not possible for it
    search_fields = ("email",)
    list_filter = (
        "is_verified",
        "is_active",
        "is_staff",
        "is_superuser",
        "is_subscribed",
    )
    ordering = ("email",)
    readonly_fields = ("password",)
    form = UserForm
    add_form = UserCreationForm
    change_password_form = AdminPasswordChangeForm

    def get_queryset(self, request: HttpRequest) -> QuerySet[User]:
        qs = super().get_queryset(request)
        return qs.prefetch_related(
            "groups", "payments_balance", "orders"
        ).prefetch_related(
            Prefetch(
                "user_permissions",
                queryset=Permission.objects.select_related("content_type"),
            )
        )

    def save_model(  # ty: ignore[invalid-method-override]
        self, request: HttpRequest, obj: Any, form: UserForm, change: Any
    ) -> None:
        if form.cleaned_data.get("attributes") is None:
            obj.attributes = None
        if (
            form.cleaned_data.get("is_superuser", False)
            and not request.user.is_superuser  # ty: ignore[possibly-missing-attribute]
        ):
            raise PermissionDenied(_("You cannot jump over your head!"))
        super().save_model(request, obj, form, change)  # ty: ignore[invalid-argument-type]


# noinspection PyUnusedLocal
@register(ChatThread)
class ChatThreadAdmin(ModelAdmin):
    list_display = (
        "uuid",
        "user",
        "email",
        "assigned_to",
        "status",
        "last_message_at",
        "is_active",
        "created",
        "modified",
    )
    list_filter = (
        "status",
        "is_active",
        ("assigned_to", admin.EmptyFieldListFilter),
    )
    search_fields = ("uuid", "email", "user__email", "user__username")
    autocomplete_fields = ("user", "assigned_to")
    actions = (
        "close_threads",
        "open_threads",
        "delete_selected",
    )
    readonly_fields = ("created", "modified")

    @admin.action(description=_("Close selected threads"))
    def close_threads(self, request, queryset):
        queryset.update(status=ThreadStatus.CLOSED)

    @admin.action(description=_("Open selected threads"))
    def open_threads(self, request, queryset):
        queryset.update(status=ThreadStatus.OPEN)


@register(ChatMessage)
class ChatMessageAdmin(admin.ModelAdmin):
    list_display = ("uuid", "thread", "sender_type", "sender_user", "sent_at")
    list_filter = ("sender_type",)
    search_fields = ("uuid", "thread__uuid", "sender_user__email")
    autocomplete_fields = ("thread", "sender_user")
    readonly_fields = ("created", "modified")


@register(EmailTemplate)
class EmailTemplateAdmin(ActivationActionsMixin, ModelAdmin):
    list_display = ("name", "slug", "subject", "is_active", "modified")
    list_filter = ("is_active",)
    search_fields = ("name", "slug", "subject")
    prepopulated_fields = {"slug": ("name",)}
    readonly_fields = ("uuid", "created", "modified")

    fieldsets = (
        (None, {"fields": ("name", "slug", "subject")}),
        (_("Content"), {"fields": ("content", "plain_content")}),
        (
            _("Documentation"),
            {
                "fields": ("available_variables",),
                "classes": ("collapse",),
            },
        ),
        (_("Metadata"), {"fields": ("uuid", "is_active", "created", "modified")}),
    )

    def formfield_for_dbfield(self, db_field, request, **kwargs):
        if db_field.name == "content":
            kwargs["widget"] = MarkdownWidget
        return super().formfield_for_dbfield(db_field, request, **kwargs)


class CampaignRecipientInline(TabularInline):
    model = CampaignRecipient
    extra = 0
    readonly_fields = (
        "user",
        "status",
        "sent_at",
        "opened_at",
        "clicked_at",
        "error_message",
    )
    can_delete = False
    max_num = 0  # Prevent adding new recipients via inline

    def has_add_permission(self, request, obj=None):
        return False


@register(EmailCampaign)
class EmailCampaignAdmin(ActivationActionsMixin, ModelAdmin):
    list_display = (
        "name",
        "template",
        "status",
        "scheduled_at",
        "sent_at",
        "stats_display",
        "is_active",
    )
    list_filter = ("status", "is_active", "template")
    search_fields = ("name",)
    readonly_fields = (
        "uuid",
        "created",
        "modified",
        "sent_at",
        "total_recipients",
        "sent_count",
        "failed_count",
        "opened_count",
        "clicked_count",
    )
    autocomplete_fields = ("template",)
    inlines = (CampaignRecipientInline,)
    actions = [
        "delete_selected",
        "activate_selected",
        "deactivate_selected",
        "prepare_recipients",
        "send_campaign_now",
        "cancel_campaign",
    ]

    fieldsets = (
        (None, {"fields": ("name", "template", "status")}),
        (_("Scheduling"), {"fields": ("scheduled_at", "sent_at")}),
        (
            _("Statistics"),
            {
                "fields": (
                    "total_recipients",
                    "sent_count",
                    "failed_count",
                    "opened_count",
                    "clicked_count",
                )
            },
        ),
        (_("Metadata"), {"fields": ("uuid", "is_active", "created", "modified")}),
    )

    @admin.display(description=_("stats"))
    def stats_display(self, obj: EmailCampaign) -> str:
        return (
            f"{obj.sent_count}/{obj.total_recipients} sent, {obj.opened_count} opened"
        )

    @admin.action(description=_("Prepare recipients (subscribed users)"))
    def prepare_recipients(
        self, request: HttpRequest, queryset: QuerySet[EmailCampaign]
    ) -> None:
        for campaign in queryset:
            if campaign.status != CampaignStatus.DRAFT:
                self.message_user(
                    request,
                    _("Campaign '{}' is not in draft status.").format(campaign.name),
                    level=messages.WARNING,
                )
                continue
            prepare_campaign_recipients.delay(str(campaign.uuid))
            self.message_user(
                request,
                _("Preparing recipients for campaign '{}'.").format(campaign.name),
                level=messages.SUCCESS,
            )

    @admin.action(description=_("Send campaign now"))
    def send_campaign_now(
        self, request: HttpRequest, queryset: QuerySet[EmailCampaign]
    ) -> None:
        for campaign in queryset:
            if campaign.status not in (CampaignStatus.DRAFT, CampaignStatus.SCHEDULED):
                self.message_user(
                    request,
                    _("Campaign '{}' cannot be sent (status: {}).").format(
                        campaign.name, campaign.status
                    ),
                    level=messages.WARNING,
                )
                continue
            if campaign.total_recipients == 0:
                self.message_user(
                    request,
                    _(
                        "Campaign '{}' has no recipients. Prepare recipients first."
                    ).format(campaign.name),
                    level=messages.WARNING,
                )
                continue

            campaign.status = CampaignStatus.SENDING
            campaign.save(update_fields=["status", "modified"])
            send_campaign_email_batch.delay(str(campaign.uuid))
            self.message_user(
                request,
                _("Started sending campaign '{}'.").format(campaign.name),
                level=messages.SUCCESS,
            )

    @admin.action(description=_("Cancel campaign"))
    def cancel_campaign(
        self, request: HttpRequest, queryset: QuerySet[EmailCampaign]
    ) -> None:
        updated = queryset.filter(
            status__in=(CampaignStatus.DRAFT, CampaignStatus.SCHEDULED)
        ).update(status=CampaignStatus.CANCELLED, modified=timezone.now())
        self.message_user(
            request,
            _("{} campaign(s) cancelled.").format(updated),
            level=messages.SUCCESS,
        )


@register(CampaignRecipient)
class CampaignRecipientAdmin(ModelAdmin):
    list_display = (
        "campaign",
        "user",
        "status",
        "sent_at",
        "opened_at",
        "clicked_at",
    )
    list_filter = ("status", "campaign")
    search_fields = ("user__email", "campaign__name")
    readonly_fields = (
        "uuid",
        "campaign",
        "user",
        "tracking_id",
        "created",
        "modified",
    )
    autocomplete_fields = ("campaign", "user")

    def has_add_permission(self, request):
        return False  # Recipients are added via campaign preparation


class GroupAdmin(BaseGroupAdmin, ModelAdmin):
    pass


class BlacklistedTokenAdmin(BaseBlacklistedTokenAdmin, ModelAdmin):
    pass


class OutstandingTokenAdmin(BaseOutstandingTokenAdmin, ModelAdmin):
    pass


@register(AdminOTPCode)
class AdminOTPCodeAdmin(ModelAdmin):
    list_display = ("user", "code", "is_used", "created")
    list_filter = ("is_used",)
    search_fields = ("user__email",)
    readonly_fields = ("user", "code", "is_used", "created", "modified")

    def has_add_permission(self, request):
        return False


admin.site.register(User, UserAdmin)

admin.site.unregister(BaseGroup)
admin.site.register(Group, GroupAdmin)

admin.site.unregister(BaseBlacklistedToken)
admin.site.register(BlacklistedToken, BlacklistedTokenAdmin)

admin.site.unregister(BaseOutstandingToken)
admin.site.register(OutstandingToken, OutstandingTokenAdmin)