Mercurial > public > sg101

annotate user_photos/forms.py @ 1200:b9514abc2a67

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Initial commit of ssg101.
author Brian Neal <bgneal@gmail.com>
date Sat, 24 Jun 2023 16:06:51 -0500
parents 7fc6c42b2f5b
children
rev   line source
bgneal@695 1 """Forms for the user_photos application."""
bgneal@749 2 import hashlib
bgneal@964 3 import os.path
bgneal@971 4 import urlparse
bgneal@700 5
bgneal@695 6 from django import forms
bgneal@697 7 from django.conf import settings
bgneal@695 8
bgneal@971 9 from core.download import download_file
bgneal@976 10 from core.functions import remove_file, TemporaryFile
bgneal@1195 11 from core.images.upload import process_upload
bgneal@701 12 from core.services import get_redis_connection
bgneal@696 13 from user_photos.models import Photo
bgneal@696 14
bgneal@695 15
bgneal@701 16 def rate_limit(key, limit, seconds):
bgneal@701 17 """Use Redis to do a rate limit check. Returns True if the limit is violated
bgneal@701 18 and False otherwise.
bgneal@701 19
bgneal@701 20 key - the key to check in Redis
bgneal@701 21 limit - the rate limit maximum value
bgneal@701 22 seconds - the rate limit period in seconds
bgneal@701 23
bgneal@701 24 """
bgneal@701 25 conn = get_redis_connection()
bgneal@701 26 val = conn.incr(key)
bgneal@701 27 if val == 1:
bgneal@701 28 conn.expire(key, seconds)
bgneal@701 29 return val > limit
bgneal@701 30
bgneal@701 31
bgneal@971 32 def rate_limit_user(user):
bgneal@971 33 """Shared function to rate limit user uploads."""
bgneal@971 34 key = 'user_photos:counts:' + user.username
bgneal@971 35 limit = settings.USER_PHOTOS_RATE_LIMIT
bgneal@971 36 if rate_limit(key, *limit):
bgneal@971 37 raise forms.ValidationError("You've exceeded your upload quota. "
bgneal@971 38 "Please try again later.")
bgneal@971 39
bgneal@971 40
bgneal@695 41 class UploadForm(forms.Form):
bgneal@695 42 image_file = forms.ImageField()
bgneal@696 43
bgneal@696 44 def __init__(self, *args, **kwargs):
bgneal@696 45 self.user = kwargs.pop('user')
bgneal@696 46 super(UploadForm, self).__init__(*args, **kwargs)
bgneal@696 47
bgneal@701 48 def clean(self):
bgneal@701 49 cleaned_data = super(UploadForm, self).clean()
bgneal@971 50 rate_limit_user(self.user)
bgneal@701 51 return cleaned_data
bgneal@701 52
bgneal@696 53 def save(self):
bgneal@696 54 """Processes the image and creates a new Photo object, which is saved to
bgneal@696 55 the database. The new Photo instance is returned.
bgneal@696 56
bgneal@749 57 Note that we do de-duplication. A signature is computed for the photo.
bgneal@749 58 If the user has already uploaded a file with the same signature, that
bgneal@749 59 photo object is returned instead.
bgneal@749 60
bgneal@696 61 This function should only be called if is_valid() returns True.
bgneal@696 62
bgneal@696 63 """
bgneal@749 64 # Check for duplicate uploads from this user
bgneal@749 65 signature = self._signature()
bgneal@749 66 try:
bgneal@749 67 return Photo.objects.get(user=self.user, signature=signature)
bgneal@749 68 except Photo.DoesNotExist:
bgneal@749 69 pass
bgneal@749 70
bgneal@964 71 # Trying to use PIL (or Pillow) on a Django UploadedFile is often
bgneal@964 72 # problematic because the file is often an in-memory file if it is under
bgneal@964 73 # a certain size. This complicates matters and many of the operations we try
bgneal@964 74 # to perform on it fail if this is the case. To get around these issues,
bgneal@964 75 # we make a copy of the file on the file system and operate on the copy.
bgneal@700 76
bgneal@964 77 fp = self.cleaned_data['image_file']
bgneal@964 78 ext = os.path.splitext(fp.name)[1]
bgneal@964 79
bgneal@964 80 # Write the UploadedFile to a temporary file on disk
bgneal@964 81 with TemporaryFile(suffix=ext) as t:
bgneal@964 82 for chunk in fp.chunks():
bgneal@964 83 t.file.write(chunk)
bgneal@964 84 t.file.close()
bgneal@964 85
bgneal@1195 86 url, thumb_url = process_upload(self.user, t.filename)
bgneal@700 87
bgneal@749 88 photo = Photo(user=self.user, url=url, thumb_url=thumb_url,
bgneal@749 89 signature=signature)
bgneal@696 90 photo.save()
bgneal@696 91 return photo
bgneal@749 92
bgneal@749 93 def _signature(self):
bgneal@749 94 """Calculates and returns a signature for the image file as a hex digest
bgneal@749 95 string.
bgneal@749 96
bgneal@749 97 This function should only be called if is_valid() is True.
bgneal@749 98
bgneal@749 99 """
bgneal@749 100 fp = self.cleaned_data['image_file']
bgneal@749 101 md5 = hashlib.md5()
bgneal@749 102 for chunk in fp.chunks():
bgneal@749 103 md5.update(chunk)
bgneal@749 104 return md5.hexdigest()
bgneal@971 105
bgneal@971 106
bgneal@971 107 class HotLinkImageForm(forms.Form):
bgneal@971 108 """Form for hot-linking images. If the supplied URL's host is on
bgneal@971 109 a white-list, return it as-is. Otherwise attempt to download the image at
bgneal@971 110 the given URL and upload it into our S3 storage. The URL to the new location
bgneal@971 111 is returned.
bgneal@971 112 """
bgneal@971 113 url = forms.URLField()
bgneal@971 114
bgneal@971 115 def __init__(self, *args, **kwargs):
bgneal@971 116 self.user = kwargs.pop('user')
bgneal@971 117 super(HotLinkImageForm, self).__init__(*args, **kwargs)
bgneal@971 118
bgneal@971 119 def clean_url(self):
bgneal@971 120 the_url = self.cleaned_data['url']
bgneal@971 121 self.url_parts = urlparse.urlsplit(the_url)
bgneal@971 122 if self.url_parts.scheme not in ['http', 'https']:
bgneal@971 123 raise forms.ValidationError("Invalid URL scheme")
bgneal@971 124 return the_url
bgneal@971 125
bgneal@971 126 def clean(self):
bgneal@971 127 cleaned_data = super(HotLinkImageForm, self).clean()
bgneal@971 128 rate_limit_user(self.user)
bgneal@971 129 return cleaned_data
bgneal@971 130
bgneal@971 131 def save(self):
bgneal@971 132 url = self.url_parts.geturl()
bgneal@971 133 if (self.url_parts.scheme == 'https' and
bgneal@971 134 self.url_parts.hostname in settings.USER_IMAGES_SOURCES):
bgneal@971 135 return url
bgneal@971 136
bgneal@971 137 # Try to download the file
bgneal@976 138 with remove_file(download_file(url)) as path:
bgneal@1195 139 url, _ = process_upload(self.user, path)
bgneal@971 140
bgneal@971 141 return url