|
| 1 | +import os |
| 2 | +import boto3 |
| 3 | +from django.views.decorators.http import require_http_methods |
| 4 | +from django.shortcuts import render |
| 5 | +from django.http import JsonResponse |
| 6 | +from django.core.exceptions import ValidationError |
| 7 | +from django.conf import settings |
| 8 | +from django.conf import settings |
| 9 | +from django import forms |
| 10 | +from datetime import datetime |
| 11 | +from botocore.exceptions import ClientError, NoCredentialsError |
| 12 | + |
| 13 | +MAX_UPLOAD_SIZE = 5 * 1024 * 1024 |
| 14 | +ALLOWED_IMAGE_EXTENSIONS = (".jpg", ".jpeg", ".png", ".gif", ".webp") |
| 15 | + |
| 16 | + |
| 17 | +@require_http_methods(["GET", "POST"]) |
| 18 | +def image_upload_view(request): |
| 19 | + """Function-based view for handling image uploads""" |
| 20 | + |
| 21 | + if request.method == "GET": |
| 22 | + form = ImageUploadForm() |
| 23 | + return render(request, "image_upload.html", {"form": form}) |
| 24 | + |
| 25 | + elif request.method == "POST": |
| 26 | + # Check if it's an AJAX request |
| 27 | + if request.headers.get("X-Requested-With") == "XMLHttpRequest": |
| 28 | + try: |
| 29 | + if "image" not in request.FILES: |
| 30 | + return JsonResponse( |
| 31 | + {"success": False, "error": "No image file provided"}, |
| 32 | + status=400, |
| 33 | + ) |
| 34 | + |
| 35 | + image = request.FILES["image"] |
| 36 | + form = ImageUploadForm(request.POST, request.FILES) |
| 37 | + |
| 38 | + if form.is_valid(): |
| 39 | + uploader = S3ImageUploader() |
| 40 | + image.seek(0) # Reset file pointer |
| 41 | + result = uploader.upload_image(image, image.name) |
| 42 | + |
| 43 | + if result["success"]: |
| 44 | + return JsonResponse(result) |
| 45 | + else: |
| 46 | + return JsonResponse(result, status=500) |
| 47 | + else: |
| 48 | + return JsonResponse( |
| 49 | + {"success": False, "error": list(form.errors.values())[0][0]}, |
| 50 | + status=400, |
| 51 | + ) |
| 52 | + |
| 53 | + except Exception as e: |
| 54 | + return JsonResponse( |
| 55 | + {"success": False, "error": "Upload failed due to server error"}, |
| 56 | + status=500, |
| 57 | + ) |
| 58 | + |
| 59 | + # Handle regular form submission |
| 60 | + else: |
| 61 | + form = ImageUploadForm(request.POST, request.FILES) |
| 62 | + |
| 63 | + if form.is_valid(): |
| 64 | + image = form.cleaned_data["image"] |
| 65 | + uploader = S3ImageUploader() |
| 66 | + image.seek(0) |
| 67 | + result = uploader.upload_image(image, image.name) |
| 68 | + |
| 69 | + if result["success"]: |
| 70 | + return render( |
| 71 | + request, |
| 72 | + "image_upload_success.html", |
| 73 | + {"url": result["url"], "key": result["key"]}, |
| 74 | + ) |
| 75 | + else: |
| 76 | + form.add_error("image", result["error"]) |
| 77 | + |
| 78 | + return render(request, "image_upload.html", {"form": form}) |
| 79 | + |
| 80 | + |
| 81 | +class ImageUploadForm(forms.Form): |
| 82 | + image = forms.ImageField( |
| 83 | + widget=forms.ClearableFileInput(attrs={"accept": "image/*"}), |
| 84 | + help_text="Upload an image file (JPG, PNG, GIF, WebP). Max size: 5MB", |
| 85 | + ) |
| 86 | + |
| 87 | + def clean_image(self): |
| 88 | + image = self.cleaned_data.get("image") |
| 89 | + |
| 90 | + if not image: |
| 91 | + raise ValidationError("No image file provided.") |
| 92 | + |
| 93 | + # Check file size |
| 94 | + if image.size > MAX_UPLOAD_SIZE: |
| 95 | + raise ValidationError( |
| 96 | + f"File size must be less than {MAX_UPLOAD_SIZE / (1024*1024):.1f}MB" |
| 97 | + ) |
| 98 | + |
| 99 | + # Check file extension |
| 100 | + ext = os.path.splitext(image.name)[1].lower() |
| 101 | + if ext not in ALLOWED_IMAGE_EXTENSIONS: |
| 102 | + raise ValidationError( |
| 103 | + f"Unsupported file format. Allowed formats: {', '.join(ALLOWED_IMAGE_EXTENSIONS)}" |
| 104 | + ) |
| 105 | + |
| 106 | + return image |
| 107 | + |
| 108 | + |
| 109 | +class S3ImageUploader: |
| 110 | + def __init__(self): |
| 111 | + self.s3_client = boto3.client( |
| 112 | + "s3", |
| 113 | + aws_access_key_id=settings.AWS_ACCESS_KEY_ID, |
| 114 | + aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, |
| 115 | + region_name=settings.AWS_S3_REGION_NAME, |
| 116 | + ) |
| 117 | + self.bucket_name = settings.AWS_STORAGE_BUCKET_NAME |
| 118 | + |
| 119 | + def generate_s3_key(self, filename): |
| 120 | + """Generate S3 key with format: static/YYYY/filename.ext""" |
| 121 | + current_year = datetime.now().year |
| 122 | + base_name, ext = os.path.splitext(filename) |
| 123 | + # Sanitize filename (remove special characters, spaces) |
| 124 | + safe_base_name = "".join( |
| 125 | + c for c in base_name if c.isalnum() or c in ("-", "_") |
| 126 | + ).strip() |
| 127 | + if not safe_base_name: |
| 128 | + safe_base_name = "image" |
| 129 | + |
| 130 | + return f"static/{current_year}/{safe_base_name}{ext.lower()}" |
| 131 | + |
| 132 | + def check_key_exists(self, key): |
| 133 | + """Check if a key already exists in S3""" |
| 134 | + try: |
| 135 | + self.s3_client.head_object(Bucket=self.bucket_name, Key=key) |
| 136 | + return True |
| 137 | + except ClientError as e: |
| 138 | + if e.response["Error"]["Code"] == "404": |
| 139 | + return False |
| 140 | + else: |
| 141 | + raise e |
| 142 | + |
| 143 | + def get_unique_key(self, base_key): |
| 144 | + """Generate a unique key by appending numbers if necessary""" |
| 145 | + if not self.check_key_exists(base_key): |
| 146 | + return base_key |
| 147 | + |
| 148 | + # Extract parts of the key |
| 149 | + path_parts = base_key.split("/") |
| 150 | + filename_with_ext = path_parts[-1] |
| 151 | + path_prefix = "/".join(path_parts[:-1]) |
| 152 | + |
| 153 | + base_name, ext = os.path.splitext(filename_with_ext) |
| 154 | + |
| 155 | + counter = 2 |
| 156 | + while counter <= 1000: # Prevent infinite loop |
| 157 | + new_filename = f"{base_name}-{counter}{ext}" |
| 158 | + new_key = f"{path_prefix}/{new_filename}" |
| 159 | + |
| 160 | + if not self.check_key_exists(new_key): |
| 161 | + return new_key |
| 162 | + |
| 163 | + counter += 1 |
| 164 | + |
| 165 | + # If we've tried 1000 variations, use timestamp |
| 166 | + timestamp = int(datetime.now().timestamp()) |
| 167 | + fallback_filename = f"{base_name}-{timestamp}{ext}" |
| 168 | + return f"{path_prefix}/{fallback_filename}" |
| 169 | + |
| 170 | + def upload_image(self, file_obj, original_filename): |
| 171 | + """Upload image to S3 and return the final key and URL""" |
| 172 | + try: |
| 173 | + # Generate base S3 key |
| 174 | + base_key = self.generate_s3_key(original_filename) |
| 175 | + |
| 176 | + # Get unique key (handle duplicates) |
| 177 | + final_key = self.get_unique_key(base_key) |
| 178 | + |
| 179 | + # Upload file |
| 180 | + self.s3_client.upload_fileobj( |
| 181 | + file_obj, |
| 182 | + self.bucket_name, |
| 183 | + final_key, |
| 184 | + ExtraArgs={ |
| 185 | + "ContentType": self._get_content_type(original_filename), |
| 186 | + "ACL": "public-read", # Make publicly readable |
| 187 | + }, |
| 188 | + ) |
| 189 | + |
| 190 | + # Generate public URL |
| 191 | + url = f"https://static.simonwillison.net/{final_key}" |
| 192 | + |
| 193 | + return { |
| 194 | + "success": True, |
| 195 | + "key": final_key, |
| 196 | + "url": url, |
| 197 | + "bucket": self.bucket_name, |
| 198 | + } |
| 199 | + |
| 200 | + except NoCredentialsError: |
| 201 | + return {"success": False, "error": "AWS credentials not configured"} |
| 202 | + except ClientError as e: |
| 203 | + return {"success": False, "error": f"S3 upload failed: {str(e)}"} |
| 204 | + except Exception as e: |
| 205 | + return {"success": False, "error": f"Upload failed: {str(e)}"} |
| 206 | + |
| 207 | + def _get_content_type(self, filename): |
| 208 | + """Get content type based on file extension""" |
| 209 | + ext = os.path.splitext(filename)[1].lower() |
| 210 | + content_types = { |
| 211 | + ".jpg": "image/jpeg", |
| 212 | + ".jpeg": "image/jpeg", |
| 213 | + ".png": "image/png", |
| 214 | + ".gif": "image/gif", |
| 215 | + ".webp": "image/webp", |
| 216 | + } |
| 217 | + return content_types.get(ext, "application/octet-stream") |
0 commit comments