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="Leave blank to keep the existing encrypted key.", ) 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("API key is required when creating an AI model.") 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")}), ( "Credentials", { "fields": ("api_key", "api_key_status"), "description": "The stored key is encrypted and never displayed.", }, ), ( "Capabilities and request defaults", { "fields": ( "capabilities", "timeout_seconds", "connect_timeout_seconds", "extra_body", ) }, ), ("Timestamps", {"fields": ("created_at", "updated_at")}), ) @admin.display(description="capabilities") 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 "not set" 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="changed fields") 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))