cmhub/apps/ai/admin.py

from django import forms
from django.contrib import admin

from .audit import create_config_audit_log, snapshot_config
from .models import AiConfigAuditLog, AiModel, ModelAlias
from .security import AiKeyEncryptionError, encrypt_api_key


class AiModelAdminForm(forms.ModelForm):
    api_key = forms.CharField(
        label="API Key",
        required=False,
        widget=forms.PasswordInput(render_value=False),
        help_text="留空表示保留已有加密密钥。",
    )

    class Meta:
        model = AiModel
        fields = (
            "name",
            "url",
            "model",
            "api_type",
            "api_key",
            "capabilities",
            "timeout_seconds",
            "connect_timeout_seconds",
            "extra_body",
            "is_active",
    )

    def clean(self):
        cleaned_data = super().clean()
        api_key = cleaned_data.get("api_key")
        if not self.instance.pk and not api_key:
            raise forms.ValidationError("创建 AI 模型时必须填写 API Key。")
        if api_key:
            try:
                self._api_key_encrypted = encrypt_api_key(api_key)
            except AiKeyEncryptionError as exc:
                raise forms.ValidationError({"api_key": str(exc)}) from exc
        else:
            self._api_key_encrypted = ""
        return cleaned_data

    def save(self, commit=True):
        instance = super().save(commit=False)
        if self._api_key_encrypted:
            instance.api_key_encrypted = self._api_key_encrypted
        if commit:
            instance.save()
            self.save_m2m()
        return instance


@admin.register(AiModel)
class AiModelAdmin(admin.ModelAdmin):
    form = AiModelAdminForm
    list_display = (
        "name",
        "model",
        "api_type",
        "capabilities_display",
        "api_key_status",
        "is_active",
        "updated_at",
    )
    list_filter = ("api_type", "is_active")
    search_fields = ("name", "model", "url")
    readonly_fields = ("api_key_status", "created_at", "updated_at")
    fieldsets = (
        (None, {"fields": ("name", "url", "model", "api_type", "is_active")}),
        (
            "凭证",
            {
                "fields": ("api_key", "api_key_status"),
                "description": "已保存的密钥会加密存储，后台不会回显明文。",
            },
        ),
        (
            "能力与请求默认值",
            {
                "fields": (
                    "capabilities",
                    "timeout_seconds",
                    "connect_timeout_seconds",
                    "extra_body",
                )
            },
        ),
        ("时间信息", {"fields": ("created_at", "updated_at")}),
    )

    @admin.display(description="能力")
    def capabilities_display(self, obj):
        return ", ".join(sorted(obj.capabilities_set()))

    @admin.display(description="API Key")
    def api_key_status(self, obj):
        return obj.api_key_masked or "未设置"

    def save_model(self, request, obj, form, change):
        before = _snapshot_existing(obj) if change else None
        super().save_model(request, obj, form, change)
        create_config_audit_log(
            actor=request.user,
            action=(
                AiConfigAuditLog.Action.UPDATE
                if change
                else AiConfigAuditLog.Action.CREATE
            ),
            instance=obj,
            before=before,
            after=snapshot_config(obj),
        )

    def delete_model(self, request, obj):
        before = snapshot_config(obj)
        target_id = obj.pk
        target_repr = str(obj)
        super().delete_model(request, obj)
        create_config_audit_log(
            actor=request.user,
            action=AiConfigAuditLog.Action.DELETE,
            instance=obj,
            before=before,
            after=None,
            target_id=target_id,
            target_repr=target_repr,
        )

    def delete_queryset(self, request, queryset):
        snapshots = [
            (obj, obj.pk, str(obj), snapshot_config(obj))
            for obj in queryset
        ]
        super().delete_queryset(request, queryset)
        for obj, target_id, target_repr, before in snapshots:
            create_config_audit_log(
                actor=request.user,
                action=AiConfigAuditLog.Action.DELETE,
                instance=obj,
                before=before,
                after=None,
                target_id=target_id,
                target_repr=target_repr,
            )


@admin.register(ModelAlias)
class ModelAliasAdmin(admin.ModelAdmin):
    list_display = (
        "operation_type",
        "alias",
        "ai_model",
        "is_default",
        "is_active",
        "updated_at",
    )
    list_filter = ("operation_type", "is_default", "is_active")
    search_fields = ("alias", "ai_model__name", "ai_model__model")
    autocomplete_fields = ("ai_model",)
    readonly_fields = ("created_at", "updated_at")

    def save_model(self, request, obj, form, change):
        before = _snapshot_existing(obj) if change else None
        super().save_model(request, obj, form, change)
        create_config_audit_log(
            actor=request.user,
            action=(
                AiConfigAuditLog.Action.UPDATE
                if change
                else AiConfigAuditLog.Action.CREATE
            ),
            instance=obj,
            before=before,
            after=snapshot_config(obj),
        )

    def delete_model(self, request, obj):
        before = snapshot_config(obj)
        target_id = obj.pk
        target_repr = str(obj)
        super().delete_model(request, obj)
        create_config_audit_log(
            actor=request.user,
            action=AiConfigAuditLog.Action.DELETE,
            instance=obj,
            before=before,
            after=None,
            target_id=target_id,
            target_repr=target_repr,
        )

    def delete_queryset(self, request, queryset):
        snapshots = [
            (obj, obj.pk, str(obj), snapshot_config(obj))
            for obj in queryset
        ]
        super().delete_queryset(request, queryset)
        for obj, target_id, target_repr, before in snapshots:
            create_config_audit_log(
                actor=request.user,
                action=AiConfigAuditLog.Action.DELETE,
                instance=obj,
                before=before,
                after=None,
                target_id=target_id,
                target_repr=target_repr,
            )


@admin.register(AiConfigAuditLog)
class AiConfigAuditLogAdmin(admin.ModelAdmin):
    list_display = (
        "created_at",
        "actor",
        "action",
        "target_type",
        "target_id",
        "target_repr",
        "changed_fields_display",
    )
    list_filter = ("action", "target_type", "created_at")
    search_fields = (
        "target_repr",
        "actor__username",
        "actor__email",
    )
    ordering = ("-created_at", "-id")

    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

    @admin.display(description="变更字段")
    def changed_fields_display(self, obj):
        return ", ".join(obj.changed_fields)


def _snapshot_existing(obj):
    return snapshot_config(obj.__class__.objects.get(pk=obj.pk))