|
| 1 | +"""Admin configuration for dataroom models.""" |
| 2 | + |
| 3 | +import io |
| 4 | +import os |
| 5 | +import re |
| 6 | +import zipfile |
| 7 | +from datetime import datetime |
| 8 | + |
| 9 | +from django.conf import settings |
| 10 | +from django.contrib import admin |
| 11 | +from django.http import FileResponse, HttpRequest, HttpResponse |
| 12 | +from django.utils.html import format_html |
| 13 | +from django.utils.safestring import mark_safe |
| 14 | + |
| 15 | +from .models import BulkDownload, Customer, DataEndpoint, FileDownload, UploadedFile |
| 16 | + |
| 17 | + |
| 18 | +class DataEndpointInline(admin.TabularInline): |
| 19 | + """Inline admin for data endpoints.""" |
| 20 | + |
| 21 | + model = DataEndpoint |
| 22 | + extra = 0 |
| 23 | + fields = ("name", "status", "created_at", "copy_url_button") |
| 24 | + readonly_fields = ("created_at", "copy_url_button") |
| 25 | + can_delete = False |
| 26 | + |
| 27 | + def copy_url_button(self, obj: DataEndpoint) -> str: |
| 28 | + """Display a button to copy the upload URL.""" |
| 29 | + if obj.pk: |
| 30 | + url = f"{settings.BASE_URL}/upload/{obj.id}/" |
| 31 | + return format_html( |
| 32 | + '<button type="button" onclick="navigator.clipboard.writeText(\'{}\'); ' |
| 33 | + 'alert(\'URL copied to clipboard!\');" ' |
| 34 | + 'style="padding: 4px 8px; cursor: pointer;">Copy Upload URL</button>', |
| 35 | + url, |
| 36 | + ) |
| 37 | + return "-" |
| 38 | + |
| 39 | + copy_url_button.short_description = "Upload URL" # type: ignore[attr-defined] |
| 40 | + |
| 41 | + |
| 42 | +@admin.register(Customer) |
| 43 | +class CustomerAdmin(admin.ModelAdmin): |
| 44 | + """Admin for Customer model.""" |
| 45 | + |
| 46 | + list_display = ("name", "created_by", "created_at", "endpoint_count") |
| 47 | + list_filter = ("created_at", "created_by") |
| 48 | + search_fields = ("name", "notes") |
| 49 | + readonly_fields = ("created_at",) |
| 50 | + inlines = [DataEndpointInline] |
| 51 | + |
| 52 | + fieldsets = ( |
| 53 | + (None, {"fields": ("name", "notes")}), |
| 54 | + ("Metadata", {"fields": ("created_by", "created_at")}), |
| 55 | + ) |
| 56 | + |
| 57 | + def endpoint_count(self, obj: Customer) -> int: |
| 58 | + """Show number of endpoints for this customer.""" |
| 59 | + return obj.endpoints.count() |
| 60 | + |
| 61 | + endpoint_count.short_description = "Endpoints" # type: ignore[attr-defined] |
| 62 | + |
| 63 | + def save_model(self, request: HttpRequest, obj: Customer, form, change: bool) -> None: # type: ignore[no-untyped-def] |
| 64 | + """Set created_by to current user if creating new customer.""" |
| 65 | + if not change: # Only set on creation |
| 66 | + obj.created_by = request.user |
| 67 | + super().save_model(request, obj, form, change) |
| 68 | + |
| 69 | + |
| 70 | +class UploadedFileInline(admin.TabularInline): |
| 71 | + """Inline admin for uploaded files.""" |
| 72 | + |
| 73 | + model = UploadedFile |
| 74 | + extra = 0 |
| 75 | + fields = ("filename", "file_size_display", "uploaded_at", "is_deleted_display") |
| 76 | + readonly_fields = ("filename", "file_size_display", "uploaded_at", "is_deleted_display") |
| 77 | + can_delete = False |
| 78 | + |
| 79 | + def file_size_display(self, obj: UploadedFile) -> str: |
| 80 | + """Display file size in human-readable format.""" |
| 81 | + size = obj.file_size_bytes |
| 82 | + for unit in ["B", "KB", "MB", "GB"]: |
| 83 | + if size < 1024.0: |
| 84 | + return f"{size:.1f} {unit}" |
| 85 | + size /= 1024.0 |
| 86 | + return f"{size:.1f} TB" |
| 87 | + |
| 88 | + file_size_display.short_description = "Size" # type: ignore[attr-defined] |
| 89 | + |
| 90 | + def is_deleted_display(self, obj: UploadedFile) -> str: |
| 91 | + """Display deletion status.""" |
| 92 | + if obj.is_deleted: |
| 93 | + return format_html('<span style="color: red;">Deleted</span>') |
| 94 | + return format_html('<span style="color: green;">Active</span>') |
| 95 | + |
| 96 | + is_deleted_display.short_description = "Status" # type: ignore[attr-defined] |
| 97 | + |
| 98 | + |
| 99 | +@admin.register(DataEndpoint) |
| 100 | +class DataEndpointAdmin(admin.ModelAdmin): |
| 101 | + """Admin for DataEndpoint model.""" |
| 102 | + |
| 103 | + list_display = ("name", "customer", "status", "created_by", "created_at", "file_count", "upload_url_link") |
| 104 | + list_filter = ("status", "created_at", "created_by") |
| 105 | + search_fields = ("name", "customer__name", "description") |
| 106 | + readonly_fields = ("id", "created_at", "upload_url_display") |
| 107 | + inlines = [UploadedFileInline] |
| 108 | + actions = ["download_endpoint_as_zip"] |
| 109 | + |
| 110 | + fieldsets = ( |
| 111 | + (None, {"fields": ("customer", "name", "description", "status")}), |
| 112 | + ("Upload Information", {"fields": ("id", "upload_url_display")}), |
| 113 | + ("Metadata", {"fields": ("created_by", "created_at")}), |
| 114 | + ) |
| 115 | + |
| 116 | + def file_count(self, obj: DataEndpoint) -> int: |
| 117 | + """Show number of files for this endpoint.""" |
| 118 | + return obj.files.filter(deleted_at__isnull=True).count() |
| 119 | + |
| 120 | + file_count.short_description = "Active Files" # type: ignore[attr-defined] |
| 121 | + |
| 122 | + def upload_url_display(self, obj: DataEndpoint) -> str: |
| 123 | + """Display the full upload URL with copy button.""" |
| 124 | + if obj.pk: |
| 125 | + url = f"{settings.BASE_URL}/upload/{obj.id}/" |
| 126 | + return format_html( |
| 127 | + '<div><a href="{}" target="_blank">{}</a> ' |
| 128 | + '<button type="button" onclick="navigator.clipboard.writeText(\'{}\'); ' |
| 129 | + 'alert(\'URL copied to clipboard!\');" ' |
| 130 | + 'style="padding: 4px 8px; cursor: pointer; margin-left: 10px;">Copy URL</button></div>', |
| 131 | + url, |
| 132 | + url, |
| 133 | + url, |
| 134 | + ) |
| 135 | + return "-" |
| 136 | + |
| 137 | + upload_url_display.short_description = "Upload URL" # type: ignore[attr-defined] |
| 138 | + |
| 139 | + def upload_url_link(self, obj: DataEndpoint) -> str: |
| 140 | + """Show clickable link in list view.""" |
| 141 | + if obj.pk: |
| 142 | + url = f"/upload/{obj.id}/" |
| 143 | + return format_html('<a href="{}" target="_blank">View Upload Page</a>', url) |
| 144 | + return "-" |
| 145 | + |
| 146 | + upload_url_link.short_description = "Upload Page" # type: ignore[attr-defined] |
| 147 | + |
| 148 | + def save_model(self, request: HttpRequest, obj: DataEndpoint, form, change: bool) -> None: # type: ignore[no-untyped-def] |
| 149 | + """Set created_by to current user if creating new endpoint.""" |
| 150 | + if not change: # Only set on creation |
| 151 | + obj.created_by = request.user |
| 152 | + super().save_model(request, obj, form, change) |
| 153 | + |
| 154 | + @admin.action(description="Download all files as ZIP") |
| 155 | + def download_endpoint_as_zip(self, request: HttpRequest, queryset) -> HttpResponse: # type: ignore[no-untyped-def] |
| 156 | + """Download all files from selected endpoint as a zip file.""" |
| 157 | + if queryset.count() != 1: |
| 158 | + self.message_user(request, "Please select exactly one endpoint to download.", level="error") |
| 159 | + return HttpResponse() |
| 160 | + |
| 161 | + endpoint = queryset.first() |
| 162 | + |
| 163 | + # Get all non-deleted files for this endpoint |
| 164 | + files = endpoint.files.filter(deleted_at__isnull=True).order_by("filename") |
| 165 | + |
| 166 | + # Check if there are any files to download |
| 167 | + if not files.exists(): |
| 168 | + self.message_user(request, "No files available to download for this endpoint.", level="warning") |
| 169 | + return HttpResponse() |
| 170 | + |
| 171 | + # Get client IP |
| 172 | + x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR") |
| 173 | + if x_forwarded_for: |
| 174 | + ip_address = x_forwarded_for.split(",")[0] |
| 175 | + else: |
| 176 | + ip_address = request.META.get("REMOTE_ADDR") |
| 177 | + |
| 178 | + # Calculate total size |
| 179 | + total_bytes = sum(f.file_size_bytes for f in files) |
| 180 | + |
| 181 | + # Create zip filename |
| 182 | + timestamp = datetime.now().strftime("%Y-%m-%d-%H%M%S") |
| 183 | + customer_name_clean = re.sub(r"[^\w\-]", "_", endpoint.customer.name) |
| 184 | + endpoint_name_clean = re.sub(r"[^\w\-]", "_", endpoint.name) |
| 185 | + zip_filename = f"{customer_name_clean}-{endpoint_name_clean}-{timestamp}.zip" |
| 186 | + |
| 187 | + # Create in-memory buffer for zip file |
| 188 | + buffer = io.BytesIO() |
| 189 | + |
| 190 | + # Create zip file |
| 191 | + with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zip_file: |
| 192 | + for uploaded_file in files: |
| 193 | + # Construct full file path |
| 194 | + file_path = os.path.join(settings.MEDIA_ROOT, uploaded_file.file_path) |
| 195 | + |
| 196 | + # Check if file exists on disk |
| 197 | + if os.path.exists(file_path): |
| 198 | + # Add file to zip with original filename |
| 199 | + zip_file.write(file_path, uploaded_file.filename) |
| 200 | + |
| 201 | + # Create individual FileDownload audit record |
| 202 | + FileDownload.objects.create( |
| 203 | + file=uploaded_file, |
| 204 | + downloaded_by=request.user, |
| 205 | + ip_address=ip_address, |
| 206 | + ) |
| 207 | + |
| 208 | + # Create BulkDownload audit record |
| 209 | + BulkDownload.objects.create( |
| 210 | + endpoint=endpoint, |
| 211 | + downloaded_by=request.user, |
| 212 | + ip_address=ip_address, |
| 213 | + file_count=files.count(), |
| 214 | + total_bytes=total_bytes, |
| 215 | + ) |
| 216 | + |
| 217 | + # Show success message to user |
| 218 | + self.message_user( |
| 219 | + request, |
| 220 | + f"Downloaded {files.count()} files from {endpoint.name}", |
| 221 | + level="success", |
| 222 | + ) |
| 223 | + |
| 224 | + # Get zip content |
| 225 | + zip_content = buffer.getvalue() |
| 226 | + |
| 227 | + # Create response |
| 228 | + response = HttpResponse(zip_content, content_type="application/zip") |
| 229 | + response["Content-Disposition"] = f'attachment; filename="{zip_filename}"' |
| 230 | + response["Content-Length"] = len(zip_content) |
| 231 | + |
| 232 | + return response |
| 233 | + |
| 234 | + |
| 235 | +@admin.register(UploadedFile) |
| 236 | +class UploadedFileAdmin(admin.ModelAdmin): |
| 237 | + """Admin for UploadedFile model.""" |
| 238 | + |
| 239 | + list_display = ( |
| 240 | + "filename", |
| 241 | + "endpoint", |
| 242 | + "file_size_display", |
| 243 | + "uploaded_at", |
| 244 | + "is_deleted_display", |
| 245 | + "download_count", |
| 246 | + ) |
| 247 | + list_filter = ("uploaded_at", "deleted_at", "endpoint__customer", "endpoint") |
| 248 | + search_fields = ("filename", "endpoint__name", "endpoint__customer__name") |
| 249 | + readonly_fields = ( |
| 250 | + "filename", |
| 251 | + "file_path", |
| 252 | + "file_size_bytes", |
| 253 | + "content_type", |
| 254 | + "uploaded_at", |
| 255 | + "uploaded_by_ip", |
| 256 | + "deleted_at", |
| 257 | + "deleted_by_ip", |
| 258 | + ) |
| 259 | + actions = ["download_file"] |
| 260 | + |
| 261 | + fieldsets = ( |
| 262 | + (None, {"fields": ("endpoint", "filename", "file_size_bytes", "content_type")}), |
| 263 | + ("Upload Information", {"fields": ("uploaded_at", "uploaded_by_ip", "file_path")}), |
| 264 | + ("Deletion Information", {"fields": ("deleted_at", "deleted_by_ip")}), |
| 265 | + ) |
| 266 | + |
| 267 | + def file_size_display(self, obj: UploadedFile) -> str: |
| 268 | + """Display file size in human-readable format.""" |
| 269 | + size = obj.file_size_bytes |
| 270 | + for unit in ["B", "KB", "MB", "GB"]: |
| 271 | + if size < 1024.0: |
| 272 | + return f"{size:.1f} {unit}" |
| 273 | + size /= 1024.0 |
| 274 | + return f"{size:.1f} TB" |
| 275 | + |
| 276 | + file_size_display.short_description = "Size" # type: ignore[attr-defined] |
| 277 | + file_size_display.admin_order_field = "file_size_bytes" # type: ignore[attr-defined] |
| 278 | + |
| 279 | + def is_deleted_display(self, obj: UploadedFile) -> str: |
| 280 | + """Display deletion status with color.""" |
| 281 | + if obj.is_deleted: |
| 282 | + return format_html('<span style="color: red; font-weight: bold;">Deleted</span>') |
| 283 | + return format_html('<span style="color: green; font-weight: bold;">Active</span>') |
| 284 | + |
| 285 | + is_deleted_display.short_description = "Status" # type: ignore[attr-defined] |
| 286 | + |
| 287 | + def download_count(self, obj: UploadedFile) -> int: |
| 288 | + """Show number of times file has been downloaded.""" |
| 289 | + return obj.downloads.count() |
| 290 | + |
| 291 | + download_count.short_description = "Downloads" # type: ignore[attr-defined] |
| 292 | + |
| 293 | + @admin.action(description="Download selected files") |
| 294 | + def download_file(self, request: HttpRequest, queryset) -> HttpResponse: # type: ignore[no-untyped-def] |
| 295 | + """Download the selected file and log the download.""" |
| 296 | + if queryset.count() != 1: |
| 297 | + self.message_user(request, "Please select exactly one file to download.", level="error") |
| 298 | + return HttpResponse() |
| 299 | + |
| 300 | + uploaded_file = queryset.first() |
| 301 | + |
| 302 | + # Build full file path |
| 303 | + file_path = os.path.join(settings.MEDIA_ROOT, uploaded_file.file_path) |
| 304 | + |
| 305 | + if not os.path.exists(file_path): |
| 306 | + self.message_user(request, f"File not found: {uploaded_file.filename}", level="error") |
| 307 | + return HttpResponse() |
| 308 | + |
| 309 | + # Get client IP |
| 310 | + x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR") |
| 311 | + if x_forwarded_for: |
| 312 | + ip_address = x_forwarded_for.split(",")[0] |
| 313 | + else: |
| 314 | + ip_address = request.META.get("REMOTE_ADDR") |
| 315 | + |
| 316 | + # Log the download |
| 317 | + FileDownload.objects.create( |
| 318 | + file=uploaded_file, |
| 319 | + downloaded_by=request.user, |
| 320 | + ip_address=ip_address, |
| 321 | + ) |
| 322 | + |
| 323 | + # Serve the file |
| 324 | + response = FileResponse(open(file_path, "rb"), as_attachment=True, filename=uploaded_file.filename) |
| 325 | + return response |
| 326 | + |
| 327 | + |
| 328 | +@admin.register(FileDownload) |
| 329 | +class FileDownloadAdmin(admin.ModelAdmin): |
| 330 | + """Admin for FileDownload model (read-only audit log).""" |
| 331 | + |
| 332 | + list_display = ("file", "downloaded_by", "downloaded_at", "ip_address") |
| 333 | + list_filter = ("downloaded_at", "downloaded_by") |
| 334 | + search_fields = ("file__filename", "downloaded_by__email", "ip_address") |
| 335 | + readonly_fields = ("file", "downloaded_by", "downloaded_at", "ip_address") |
| 336 | + |
| 337 | + def has_add_permission(self, request: HttpRequest) -> bool: |
| 338 | + """Prevent manual creation of download logs.""" |
| 339 | + return False |
| 340 | + |
| 341 | + def has_delete_permission(self, request: HttpRequest, obj=None) -> bool: # type: ignore[no-untyped-def] |
| 342 | + """Prevent deletion of audit logs.""" |
| 343 | + return False |
| 344 | + |
| 345 | + def has_change_permission(self, request: HttpRequest, obj=None) -> bool: # type: ignore[no-untyped-def] |
| 346 | + """Make this read-only.""" |
| 347 | + return False |
| 348 | + |
| 349 | + |
| 350 | +@admin.register(BulkDownload) |
| 351 | +class BulkDownloadAdmin(admin.ModelAdmin): |
| 352 | + """Admin for BulkDownload model (read-only audit log).""" |
| 353 | + |
| 354 | + list_display = ("endpoint", "downloaded_by", "downloaded_at", "file_count", "total_size_display", "ip_address") |
| 355 | + list_filter = ("downloaded_at", "downloaded_by", "endpoint__customer") |
| 356 | + search_fields = ("endpoint__name", "endpoint__customer__name", "downloaded_by__email", "ip_address") |
| 357 | + readonly_fields = ("endpoint", "downloaded_by", "downloaded_at", "file_count", "total_bytes", "ip_address") |
| 358 | + |
| 359 | + def total_size_display(self, obj: BulkDownload) -> str: |
| 360 | + """Display total size in human-readable format.""" |
| 361 | + size = obj.total_bytes |
| 362 | + for unit in ["B", "KB", "MB", "GB"]: |
| 363 | + if size < 1024.0: |
| 364 | + return f"{size:.1f} {unit}" |
| 365 | + size /= 1024.0 |
| 366 | + return f"{size:.1f} TB" |
| 367 | + |
| 368 | + total_size_display.short_description = "Total Size" # type: ignore[attr-defined] |
| 369 | + total_size_display.admin_order_field = "total_bytes" # type: ignore[attr-defined] |
| 370 | + |
| 371 | + def has_add_permission(self, request: HttpRequest) -> bool: |
| 372 | + """Prevent manual creation of bulk download logs.""" |
| 373 | + return False |
| 374 | + |
| 375 | + def has_delete_permission(self, request: HttpRequest, obj=None) -> bool: # type: ignore[no-untyped-def] |
| 376 | + """Prevent deletion of audit logs.""" |
| 377 | + return False |
| 378 | + |
| 379 | + def has_change_permission(self, request: HttpRequest, obj=None) -> bool: # type: ignore[no-untyped-def] |
| 380 | + """Make this read-only.""" |
| 381 | + return False |
0 commit comments