Mercurial > public > sg101

annotate user_photos/forms.py @ 1189:83dd2db291f7

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