Mercurial > public > sg101

annotate user_photos/forms.py @ 934:9d6c2ed2f348

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add unit test for antispam signal handlers.
author Brian Neal <bgneal@gmail.com>
date Thu, 16 Apr 2015 20:49:19 -0500
parents b6e98717690b
children 51a2051588f5
rev   line source
bgneal@695 1 """Forms for the user_photos application."""
bgneal@700 2 import datetime
bgneal@749 3 import hashlib
bgneal@700 4
bgneal@695 5 from django import forms
bgneal@697 6 from django.conf import settings
bgneal@695 7
bgneal@700 8 from core.s3 import S3Bucket
bgneal@700 9 from core.image_uploader import upload
bgneal@701 10 from core.services import get_redis_connection
bgneal@696 11 from user_photos.models import Photo
bgneal@696 12
bgneal@695 13
bgneal@701 14 def rate_limit(key, limit, seconds):
bgneal@701 15 """Use Redis to do a rate limit check. Returns True if the limit is violated
bgneal@701 16 and False otherwise.
bgneal@701 17
bgneal@701 18 key - the key to check in Redis
bgneal@701 19 limit - the rate limit maximum value
bgneal@701 20 seconds - the rate limit period in seconds
bgneal@701 21
bgneal@701 22 """
bgneal@701 23 conn = get_redis_connection()
bgneal@701 24 val = conn.incr(key)
bgneal@701 25 if val == 1:
bgneal@701 26 conn.expire(key, seconds)
bgneal@701 27 return val > limit
bgneal@701 28
bgneal@701 29
bgneal@695 30 class UploadForm(forms.Form):
bgneal@695 31 image_file = forms.ImageField()
bgneal@696 32
bgneal@696 33 def __init__(self, *args, **kwargs):
bgneal@696 34 self.user = kwargs.pop('user')
bgneal@696 35 super(UploadForm, self).__init__(*args, **kwargs)
bgneal@696 36
bgneal@701 37 def clean(self):
bgneal@701 38 cleaned_data = super(UploadForm, self).clean()
bgneal@701 39
bgneal@701 40 # rate limit uploads
bgneal@701 41 key = 'user_photos:counts:' + self.user.username
bgneal@701 42 limit = settings.USER_PHOTOS_RATE_LIMIT
bgneal@701 43 if rate_limit(key, *limit):
bgneal@701 44 raise forms.ValidationError("You've exceeded your upload quota. "
bgneal@701 45 "Please try again later.")
bgneal@701 46
bgneal@701 47 return cleaned_data
bgneal@701 48
bgneal@696 49 def save(self):
bgneal@696 50 """Processes the image and creates a new Photo object, which is saved to
bgneal@696 51 the database. The new Photo instance is returned.
bgneal@696 52
bgneal@749 53 Note that we do de-duplication. A signature is computed for the photo.
bgneal@749 54 If the user has already uploaded a file with the same signature, that
bgneal@749 55 photo object is returned instead.
bgneal@749 56
bgneal@696 57 This function should only be called if is_valid() returns True.
bgneal@696 58
bgneal@696 59 """
bgneal@749 60 # Check for duplicate uploads from this user
bgneal@749 61 signature = self._signature()
bgneal@749 62 try:
bgneal@749 63 return Photo.objects.get(user=self.user, signature=signature)
bgneal@749 64 except Photo.DoesNotExist:
bgneal@749 65 pass
bgneal@749 66
bgneal@749 67 # This must not be a duplicate, proceed with upload to S3
bgneal@700 68 bucket = S3Bucket(access_key=settings.USER_PHOTOS_ACCESS_KEY,
bgneal@700 69 secret_key=settings.USER_PHOTOS_SECRET_KEY,
bgneal@700 70 base_url=settings.USER_PHOTOS_BASE_URL,
bgneal@700 71 bucket_name=settings.USER_PHOTOS_BUCKET)
bgneal@700 72
bgneal@700 73 now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
bgneal@700 74 metadata = {'user': self.user.username, 'date': now}
bgneal@700 75
bgneal@700 76 url, thumb_url = upload(fp=self.cleaned_data['image_file'],
bgneal@700 77 bucket=bucket,
bgneal@700 78 metadata=metadata,
bgneal@700 79 new_size=settings.USER_PHOTOS_MAX_SIZE,
bgneal@700 80 thumb_size=settings.USER_PHOTOS_THUMB_SIZE)
bgneal@700 81
bgneal@749 82 photo = Photo(user=self.user, url=url, thumb_url=thumb_url,
bgneal@749 83 signature=signature)
bgneal@696 84 photo.save()
bgneal@696 85 return photo
bgneal@749 86
bgneal@749 87 def _signature(self):
bgneal@749 88 """Calculates and returns a signature for the image file as a hex digest
bgneal@749 89 string.
bgneal@749 90
bgneal@749 91 This function should only be called if is_valid() is True.
bgneal@749 92
bgneal@749 93 """
bgneal@749 94 fp = self.cleaned_data['image_file']
bgneal@749 95 md5 = hashlib.md5()
bgneal@749 96 for chunk in fp.chunks():
bgneal@749 97 md5.update(chunk)
bgneal@749 98 return md5.hexdigest()