cmhub/apps/licensing/admin.py

from django import forms
from django.contrib import admin, messages
from django.contrib.auth import get_user_model
from django.http import HttpResponseRedirect
from django.shortcuts import get_object_or_404
from django.template.response import TemplateResponse
from django.urls import path, reverse

from .models import (
    ClientDevice,
    DeviceCredential,
    DeviceBindingAudit,
    DeviceSession,
    LicenseEvent,
    LicenseSeat,
    LegacyMigrationGrant,
    MigrationRequest,
    SoftwareEntitlement,
    SoftwareOrder,
    SoftwarePlan,
)
from .services import (
    LicensingError,
    grant_software_entitlement,
    create_legacy_migration_grant,
    renew_software_entitlement,
    revoke_software_entitlement,
)


def _masked_fingerprint(value: str) -> str:
    if not value:
        return ""
    return f"{value[:8]}...{value[-6:]}"


@admin.register(ClientDevice)
class ClientDeviceAdmin(admin.ModelAdmin):
    list_display = (
        "user",
        "product_code",
        "platform",
        "client_version",
        "status",
        "last_seen_at",
        "first_seen_at",
    )
    list_filter = ("product_code", "platform", "status", "last_seen_at")
    search_fields = ("user__username", "user__email")
    list_select_related = ("user",)
    ordering = ("-last_seen_at", "-id")
    readonly_fields = (
        "user",
        "product_code",
        "device_id_version",
        "device_fingerprint_masked",
        "public_key_fingerprint_masked",
        "platform",
        "client_version",
        "status",
        "first_seen_at",
        "last_seen_at",
        "created_at",
        "updated_at",
    )
    fields = readonly_fields

    def has_add_permission(self, request):
        return False

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

    @admin.display(description="设备摘要")
    def device_fingerprint_masked(self, obj):
        return _masked_fingerprint(obj.device_fingerprint)

    @admin.display(description="安装公钥摘要")
    def public_key_fingerprint_masked(self, obj):
        return _masked_fingerprint(obj.public_key_fingerprint)


@admin.register(DeviceSession)
class DeviceSessionAdmin(admin.ModelAdmin):
    list_display = ("device", "expires_at", "last_used_at", "revoked_at", "created_at")
    list_filter = ("revoked_at", "expires_at")
    search_fields = ("device__user__username", "device__user__email")
    list_select_related = ("device", "device__user")
    readonly_fields = (
        "device",
        "expires_at",
        "last_used_at",
        "revoked_at",
        "created_at",
    )
    fields = readonly_fields

    def has_add_permission(self, request):
        return False

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

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


@admin.register(DeviceBindingAudit)
class DeviceBindingAuditAdmin(admin.ModelAdmin):
    list_display = ("user", "device", "api_key", "action", "reason", "created_at")
    list_filter = ("action", "created_at")
    search_fields = ("user__username", "user__email", "api_key__key_prefix")
    list_select_related = ("user", "device", "api_key")
    readonly_fields = (
        "user",
        "device",
        "api_key",
        "action",
        "reason",
        "client_version",
        "created_at",
    )
    fields = readonly_fields

    def has_add_permission(self, request):
        return False

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

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


@admin.register(SoftwarePlan)
class SoftwarePlanAdmin(admin.ModelAdmin):
    list_display = (
        "name",
        "product_code",
        "duration_days",
        "price",
        "device_limit",
        "grace_days",
        "status",
        "updated_at",
    )
    list_filter = ("product_code", "status")
    search_fields = ("name",)
    list_editable = ("status",)
    readonly_fields = ("created_at", "updated_at")


class EntitlementGrantForm(forms.Form):
    user = forms.ModelChoiceField(queryset=get_user_model().objects.all(), label="用户")
    plan = forms.ModelChoiceField(
        queryset=SoftwarePlan.objects.filter(status=SoftwarePlan.Status.ACTIVE),
        label="套餐",
    )
    reason = forms.CharField(label="授予原因", widget=forms.Textarea(attrs={"rows": 4}))

    def clean_reason(self):
        reason = self.cleaned_data["reason"].strip()
        if not reason:
            raise forms.ValidationError("必须填写操作原因。")
        return reason


class EntitlementReasonForm(forms.Form):
    reason = forms.CharField(label="操作原因", widget=forms.Textarea(attrs={"rows": 4}))

    def clean_reason(self):
        reason = self.cleaned_data["reason"].strip()
        if not reason:
            raise forms.ValidationError("必须填写操作原因。")
        return reason


@admin.register(SoftwareEntitlement)
class SoftwareEntitlementAdmin(admin.ModelAdmin):
    change_form_template = "admin/licensing/softwareentitlement/change_form.html"
    list_display = (
        "user",
        "product_code",
        "plan_name",
        "plan_device_limit",
        "status",
        "starts_at",
        "expires_at",
        "grace_expires_at",
    )
    list_filter = ("product_code", "status", "expires_at")
    search_fields = ("user__username", "user__email", "plan_name")
    list_select_related = ("user", "source_plan")
    readonly_fields = (
        "user",
        "product_code",
        "source_plan",
        "plan_name",
        "plan_duration_days",
        "plan_price",
        "plan_device_limit",
        "plan_grace_days",
        "status",
        "starts_at",
        "expires_at",
        "grace_expires_at",
        "revoked_at",
        "created_at",
        "updated_at",
    )
    fields = readonly_fields

    def has_add_permission(self, request):
        return False

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

    def get_urls(self):
        urls = super().get_urls()
        custom_urls = [
            path(
                "grant/",
                self.admin_site.admin_view(self.grant_view),
                name="licensing_softwareentitlement_grant",
            ),
            path(
                "<path:object_id>/renew/",
                self.admin_site.admin_view(self.renew_view),
                name="licensing_softwareentitlement_renew",
            ),
            path(
                "<path:object_id>/revoke/",
                self.admin_site.admin_view(self.revoke_view),
                name="licensing_softwareentitlement_revoke",
            ),
        ]
        return custom_urls + urls

    def grant_view(self, request):
        form = EntitlementGrantForm(request.POST or None)
        if request.method == "POST" and form.is_valid():
            try:
                entitlement = grant_software_entitlement(
                    user=form.cleaned_data["user"],
                    plan=form.cleaned_data["plan"],
                    reason=form.cleaned_data["reason"],
                    actor=request.user,
                )
            except LicensingError as exc:
                form.add_error(None, exc.message)
            else:
                self.message_user(request, "软件权益已授予。", messages.SUCCESS)
                return HttpResponseRedirect(self._change_url(entitlement))
        return self._operation_response(
            request,
            title="手工授予软件权益",
            form=form,
        )

    def renew_view(self, request, object_id):
        entitlement = get_object_or_404(SoftwareEntitlement, pk=object_id)
        return self._entitlement_operation_view(
            request,
            entitlement=entitlement,
            action="renew",
            title="续期软件权益",
        )

    def revoke_view(self, request, object_id):
        entitlement = get_object_or_404(SoftwareEntitlement, pk=object_id)
        return self._entitlement_operation_view(
            request,
            entitlement=entitlement,
            action="revoke",
            title="撤销软件权益",
        )

    def _entitlement_operation_view(self, request, *, entitlement, action, title):
        form = EntitlementReasonForm(request.POST or None)
        if request.method == "POST" and form.is_valid():
            try:
                operation = (
                    renew_software_entitlement
                    if action == "renew"
                    else revoke_software_entitlement
                )
                operation(
                    entitlement=entitlement,
                    reason=form.cleaned_data["reason"],
                    actor=request.user,
                )
            except LicensingError as exc:
                form.add_error(None, exc.message)
            else:
                self.message_user(request, f"软件权益已{'续期' if action == 'renew' else '撤销'}。", messages.SUCCESS)
                return HttpResponseRedirect(self._change_url(entitlement))
        return self._operation_response(
            request,
            title=title,
            form=form,
            entitlement=entitlement,
        )

    def _operation_response(self, request, *, title, form, entitlement=None):
        context = {
            **self.admin_site.each_context(request),
            "title": title,
            "opts": self.model._meta,
            "form": form,
            "entitlement": entitlement,
        }
        return TemplateResponse(
            request,
            "admin/licensing/entitlement_operation.html",
            context,
        )

    def _change_url(self, entitlement):
        return reverse("admin:licensing_softwareentitlement_change", args=(entitlement.pk,))


class ReadOnlyLicenseAdmin(admin.ModelAdmin):
    def get_readonly_fields(self, request, obj=None):
        return tuple(field.name for field in self.model._meta.fields)

    def has_add_permission(self, request):
        return False

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

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


class HiddenLegacyLicenseAdmin(ReadOnlyLicenseAdmin):
    """Keep legacy records addressable without advertising them in admin."""

    def get_model_perms(self, request):
        return {}


@admin.register(LicenseSeat)
class LicenseSeatAdmin(HiddenLegacyLicenseAdmin):
    list_display = ("entitlement", "seat_number", "device", "bound_at", "released_at")
    list_filter = ("entitlement__product_code", "bound_at", "released_at")
    search_fields = (
        "entitlement__user__username",
        "entitlement__user__email",
        "device__user__username",
    )
    list_select_related = ("entitlement", "entitlement__user", "device")


@admin.register(LicenseEvent)
class LicenseEventAdmin(ReadOnlyLicenseAdmin):
    list_display = ("entitlement", "action", "seat", "device", "actor", "reason", "created_at")
    list_filter = ("action", "entitlement__product_code", "created_at")
    search_fields = (
        "entitlement__user__username",
        "entitlement__user__email",
        "reason",
    )
    list_select_related = ("entitlement", "seat", "device", "actor")


class LegacyMigrationGrantForm(EntitlementGrantForm):
    reason = forms.CharField(label="迁移原因", widget=forms.Textarea(attrs={"rows": 4}))


@admin.register(LegacyMigrationGrant)
class LegacyMigrationGrantAdmin(HiddenLegacyLicenseAdmin):
    list_display = ("user", "product_code", "entitlement", "status", "actor", "created_at")
    list_filter = ("product_code", "status", "created_at")
    search_fields = ("user__username", "user__email", "reason")
    list_select_related = ("user", "entitlement", "actor")

    def get_urls(self):
        urls = super().get_urls()
        return [
            path(
                "grant/",
                self.admin_site.admin_view(self.grant_view),
                name="licensing_legacymigrationgrant_grant",
            ),
        ] + urls

    def grant_view(self, request):
        form = LegacyMigrationGrantForm(request.POST or None)
        if request.method == "POST" and form.is_valid():
            try:
                grant = create_legacy_migration_grant(
                    user=form.cleaned_data["user"],
                    plan=form.cleaned_data["plan"],
                    reason=form.cleaned_data["reason"],
                    actor=request.user,
                )
            except LicensingError as exc:
                form.add_error(None, exc.message)
            else:
                self.message_user(request, "存量迁移资格已授予。", messages.SUCCESS)
                return HttpResponseRedirect(
                    reverse("admin:licensing_legacymigrationgrant_change", args=(grant.pk,))
                )
        context = {
            **self.admin_site.each_context(request),
            "title": "授予存量迁移资格",
            "opts": self.model._meta,
            "form": form,
        }
        return TemplateResponse(
            request,
            "admin/licensing/entitlement_operation.html",
            context,
        )


@admin.register(MigrationRequest)
class MigrationRequestAdmin(HiddenLegacyLicenseAdmin):
    list_display = ("request_id", "user", "device", "migration_grant", "status", "expires_at", "confirmed_at")
    list_filter = ("status", "device__product_code", "expires_at")
    search_fields = ("=request_id", "user__username", "user__email")
    list_select_related = ("user", "device", "migration_grant")


@admin.register(DeviceCredential)
class DeviceCredentialAdmin(HiddenLegacyLicenseAdmin):
    list_display = ("token_prefix", "user", "product_code", "device", "entitlement", "expires_at", "revoked_at")
    list_filter = ("product_code", "revoked_at", "expires_at")
    search_fields = ("token_prefix", "user__username", "user__email")
    list_select_related = ("user", "device", "entitlement", "seat")


@admin.register(SoftwareOrder)
class SoftwareOrderAdmin(ReadOnlyLicenseAdmin):
    list_display = (
        "created_at",
        "order_no",
        "user",
        "plan_name",
        "amount_money",
        "pay_method",
        "status",
        "payment_txn_no",
        "entitlement",
        "fulfilled_at",
    )
    list_filter = ("product_code", "pay_method", "status", "created_at")
    search_fields = ("=order_no", "=payment_txn_no", "user__username", "user__email")
    list_select_related = ("user", "source_plan", "entitlement", "fulfillment_event")