<s> from __future__ import absolute_import <EOL> from django . conf import settings <EOL> from django . template . defaultfilters import slugify <EOL> from django . utils . encoding import force_text <EOL> from django . utils . safestring import mark_safe <EOL> import unicodedata <EOL> from zerver . lib . avatar import user_avatar_hash <EOL> from zerver . decorator import JsonableError <EOL> from boto . s3 . key import Key <EOL> from boto . s3 . connection import S3Connection <EOL> from mimetypes import guess_type , guess_extension <EOL> from zerver . models import get_user_profile_by_id <EOL> import base64 <EOL> import os <EOL> import re <EOL> from PIL import Image , ImageOps <EOL> from six . moves import cStringIO as StringIO <EOL> import random <EOL> def sanitize_name ( value ) : <EOL> """<STR_LIT>""" <EOL> value = force_text ( value ) <EOL> value = unicodedata . normalize ( '<STR_LIT>' , value ) <EOL> value = re . sub ( '<STR_LIT>' , '<STR_LIT>' , value , flags = re . U ) . strip ( ) <EOL> return mark_safe ( re . sub ( '<STR_LIT>' , '<STR_LIT:->' , value , flags = re . U ) ) <EOL> def random_name ( bytes = <NUM_LIT> ) : <EOL> return base64 . urlsafe_b64encode ( os . urandom ( bytes ) ) <EOL> class BadImageError ( JsonableError ) : <EOL> pass <EOL> def resize_avatar ( image_data ) : <EOL> AVATAR_SIZE = <NUM_LIT:100> <EOL> try : <EOL> im = Image . open ( StringIO ( image_data ) ) <EOL> im = ImageOps . fit ( im , ( AVATAR_SIZE , AVATAR_SIZE ) , Image . ANTIALIAS ) <EOL> except IOError : <EOL> raise BadImageError ( "<STR_LIT>" ) <EOL> out = StringIO ( ) <EOL> im . save ( out , format = '<STR_LIT>' ) <EOL> return out . getvalue ( ) <EOL> def get_bucket ( conn , bucket_name ) : <EOL> bucket = conn . get_bucket ( bucket_name , validate = False ) <EOL> return bucket <EOL> def upload_image_to_s3 ( <EOL> bucket_name , <EOL> file_name , <EOL> content_type , <EOL> user_profile , <EOL> contents , <EOL> ) : <EOL> conn = S3Connection ( settings . S3_KEY , settings . S3_SECRET_KEY ) <EOL> bucket = get_bucket ( conn , bucket_name ) <EOL> key = Key ( bucket ) <EOL> key . key = file_name <EOL> key . set_metadata ( "<STR_LIT>" , str ( user_profile . id ) ) <EOL> key . set_metadata ( "<STR_LIT>" , str ( user_profile . realm . id ) ) <EOL> if content_type : <EOL> headers = { '<STR_LIT:Content-Type>' : content_type } <EOL> else : <EOL> headers = None <EOL> key . set_contents_from_string ( contents , headers = headers ) <EOL> def get_file_info ( request , user_file ) : <EOL> uploaded_file_name = user_file . name <EOL> content_type = request . GET . get ( '<STR_LIT>' ) <EOL> if content_type is None : <EOL> content_type = guess_type ( uploaded_file_name ) [ <NUM_LIT:0> ] <EOL> else : <EOL> uploaded_file_name = uploaded_file_name + guess_extension ( content_type ) <EOL> return uploaded_file_name , content_type <EOL> def upload_message_image_s3 ( uploaded_file_name , content_type , file_data , user_profile , target_realm = None ) : <EOL> bucket_name = settings . S3_AUTH_UPLOADS_BUCKET <EOL> s3_file_name = "<STR_LIT:/>" . join ( [ <EOL> str ( target_realm . id if target_realm is not None else user_profile . realm . id ) , <EOL> random_name ( <NUM_LIT> ) , <EOL> sanitize_name ( uploaded_file_name ) <EOL> ] ) <EOL> url = "<STR_LIT>" % ( s3_file_name ) <EOL> upload_image_to_s3 ( <EOL> bucket_name , <EOL> s3_file_name , <EOL> content_type , <EOL> user_profile , <EOL> file_data <EOL> ) <EOL> return url <EOL> def get_signed_upload_url ( path ) : <EOL> conn = S3Connection ( settings . S3_KEY , settings . S3_SECRET_KEY ) <EOL> return conn . generate_url ( <NUM_LIT:15> , '<STR_LIT:GET>' , bucket = settings . S3_AUTH_UPLOADS_BUCKET , key = path ) <EOL> def get_realm_for_filename ( path ) : <EOL> conn = S3Connection ( settings . S3_KEY , settings . S3_SECRET_KEY ) <EOL> key = get_bucket ( conn , settings . S3_AUTH_UPLOADS_BUCKET ) . get_key ( path ) <EOL> if key is None : <EOL> return None <EOL> return get_user_profile_by_id ( key . metadata [ "<STR_LIT>" ] ) . realm . id <EOL> def upload_avatar_image_s3 ( user_file , user_profile , email ) : <EOL> content_type = guess_type ( user_file . name ) [ <NUM_LIT:0> ] <EOL> bucket_name = settings . S3_AVATAR_BUCKET <EOL> s3_file_name = user_avatar_hash ( email ) <EOL> image_data = user_file . read ( ) <EOL> upload_image_to_s3 ( <EOL> bucket_name , <EOL> s3_file_name + "<STR_LIT>" , <EOL> content_type , <EOL> user_profile , <EOL> image_data , <EOL> ) <EOL> resized_data = resize_avatar ( image_data ) <EOL> upload_image_to_s3 ( <EOL> bucket_name , <EOL> s3_file_name , <EOL> '<STR_LIT>' , <EOL> user_profile , <EOL> resized_data , <EOL> ) <EOL> def mkdirs ( path ) : <EOL> dirname = os . path . dirname ( path ) <EOL> if not os . path . isdir ( dirname ) : <EOL> os . makedirs ( dirname ) <EOL> def write_local_file ( type , path , file_data ) : <EOL> file_path = os . path . join ( settings . LOCAL_UPLOADS_DIR , type , path ) <EOL> mkdirs ( file_path ) <EOL> with open ( file_path , '<STR_LIT:wb>' ) as f : <EOL> f . write ( file_data ) <EOL> def upload_message_image_local ( uploaded_file_name , content_type , file_data , user_profile , target_realm = None ) : <EOL> path = "<STR_LIT:/>" . join ( [ <EOL> str ( user_profile . realm . id ) , <EOL> format ( random . randint ( <NUM_LIT:0> , <NUM_LIT:255> ) , '<STR_LIT:x>' ) , <EOL> random_name ( <NUM_LIT> ) , <EOL> sanitize_name ( uploaded_file_name ) <EOL> ] ) <EOL> write_local_file ( '<STR_LIT>' , path , file_data ) <EOL> return '<STR_LIT>' + path <EOL> def upload_avatar_image_local ( user_file , user_profile , email ) : <EOL> email_hash = user_avatar_hash ( email ) <EOL> image_data = user_file . read ( ) <EOL> write_local_file ( '<STR_LIT>' , email_hash + '<STR_LIT>' , image_data ) <EOL> resized_data = resize_avatar ( image_data ) <EOL> write_local_file ( '<STR_LIT>' , email_hash + '<STR_LIT>' , resized_data ) <EOL> if settings . LOCAL_UPLOADS_DIR is not None : <EOL> upload_message_image = upload_message_image_local <EOL> upload_avatar_image = upload_avatar_image_local <EOL> else : <EOL> upload_message_image = upload_message_image_s3 <EOL> upload_avatar_image = upload_avatar_image_s3 <EOL> def upload_message_image_through_web_client ( request , user_file , user_profile ) : <EOL> uploaded_file_name , content_type = get_file_info ( request , user_file ) <EOL> return upload_message_image ( uploaded_file_name , content_type , user_file . read ( ) , user_profile ) </s>
