<s> """<STR_LIT>""" <EOL> import os . path <EOL> import posixpath <EOL> from flask import current_app , Module , send_from_directory , abort , url_for <EOL> from itertools import chain <EOL> from werkzeug import secure_filename , FileStorage <EOL> try : <EOL> from flask import Blueprint <EOL> except ImportError : <EOL> from flask import Module <EOL> using_blueprints = False <EOL> else : <EOL> using_blueprints = True <EOL> TEXT = ( '<STR_LIT>' , ) <EOL> DOCUMENTS = tuple ( '<STR_LIT>' . split ( ) ) <EOL> IMAGES = tuple ( '<STR_LIT>' . split ( ) ) <EOL> AUDIO = tuple ( '<STR_LIT>' . split ( ) ) <EOL> DATA = tuple ( '<STR_LIT>' . split ( ) ) <EOL> SCRIPTS = tuple ( '<STR_LIT>' . split ( ) ) <EOL> ARCHIVES = tuple ( '<STR_LIT>' . split ( ) ) <EOL> EXECUTABLES = tuple ( '<STR_LIT>' . split ( ) ) <EOL> DEFAULTS = TEXT + DOCUMENTS + IMAGES + DATA <EOL> class UploadNotAllowed ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def tuple_from ( * iters ) : <EOL> return tuple ( itertools . chain ( * iters ) ) <EOL> def extension ( filename ) : <EOL> return filename . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ - <NUM_LIT:1> ] <EOL> def lowercase_ext ( filename ) : <EOL> if '<STR_LIT:.>' in filename : <EOL> main , ext = filename . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> return main + '<STR_LIT:.>' + ext . lower ( ) <EOL> else : <EOL> return filename . lower ( ) <EOL> def addslash ( url ) : <EOL> if url . endswith ( '<STR_LIT:/>' ) : <EOL> return url <EOL> return url + '<STR_LIT:/>' <EOL> def patch_request_class ( app , size = <NUM_LIT:64> * <NUM_LIT> * <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> if size is None : <EOL> if isinstance ( app . request_class . __dict__ [ '<STR_LIT>' ] , <EOL> property ) : <EOL> return <EOL> size = app . config . get ( '<STR_LIT>' ) <EOL> reqclass = app . request_class <EOL> patched = type ( reqclass . __name__ , ( reqclass , ) , <EOL> { '<STR_LIT>' : size } ) <EOL> app . request_class = patched <EOL> def config_for_set ( uset , app , defaults = None ) : <EOL> """<STR_LIT>""" <EOL> config = app . config <EOL> prefix = '<STR_LIT>' % uset . name . upper ( ) <EOL> using_defaults = False <EOL> if defaults is None : <EOL> defaults = dict ( dest = None , url = None ) <EOL> allow_extns = tuple ( config . get ( prefix + '<STR_LIT>' , ( ) ) ) <EOL> deny_extns = tuple ( config . get ( prefix + '<STR_LIT>' , ( ) ) ) <EOL> destination = config . get ( prefix + '<STR_LIT>' ) <EOL> base_url = config . get ( prefix + '<STR_LIT>' ) <EOL> if destination is None : <EOL> if uset . default_dest : <EOL> destination = uset . default_dest ( app ) <EOL> if destination is None : <EOL> if defaults [ '<STR_LIT>' ] is not None : <EOL> using_defaults = True <EOL> destination = os . path . join ( defaults [ '<STR_LIT>' ] , uset . name ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" % uset . name ) <EOL> if base_url is None and using_defaults and defaults [ '<STR_LIT:url>' ] : <EOL> base_url = addslash ( defaults [ '<STR_LIT:url>' ] ) + uset . name + '<STR_LIT:/>' <EOL> return UploadConfiguration ( destination , base_url , allow_extns , deny_extns ) <EOL> def configure_uploads ( app , upload_sets ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( upload_sets , UploadSet ) : <EOL> upload_sets = ( upload_sets , ) <EOL> if not hasattr ( app , '<STR_LIT>' ) : <EOL> app . upload_set_config = { } <EOL> set_config = app . upload_set_config <EOL> defaults = dict ( dest = app . config . get ( '<STR_LIT>' ) , <EOL> url = app . config . get ( '<STR_LIT>' ) ) <EOL> for uset in upload_sets : <EOL> config = config_for_set ( uset , app , defaults ) <EOL> set_config [ uset . name ] = config <EOL> should_serve = any ( s . base_url is None for s in set_config . itervalues ( ) ) <EOL> if using_blueprints : <EOL> if '<STR_LIT>' not in app . blueprints and should_serve : <EOL> app . register_blueprint ( uploads_mod ) <EOL> else : <EOL> if '<STR_LIT>' not in app . modules and should_serve : <EOL> app . register_module ( uploads_mod ) <EOL> class UploadsManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app = None ) : <EOL> self . app = app <EOL> def init_app ( self , app ) : <EOL> self . app = app <EOL> def register ( self , upload_sets ) : <EOL> configure_uploads ( self . app , upload_sets ) <EOL> class All ( object ) : <EOL> """<STR_LIT>""" <EOL> def __contains__ ( self , item ) : <EOL> return True <EOL> ALL = All ( ) <EOL> class AllExcept ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , items ) : <EOL> self . items = items <EOL> def __contains__ ( self , item ) : <EOL> return item not in self . items <EOL> class UploadConfiguration ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , destination , base_url = None , allow = ( ) , deny = ( ) ) : <EOL> self . destination = destination <EOL> self . base_url = base_url <EOL> self . allow = allow <EOL> self . deny = deny <EOL> @ property <EOL> def tuple ( self ) : <EOL> return ( self . destination , self . base_url , self . allow , self . deny ) <EOL> def __eq__ ( self , other ) : <EOL> return self . tuple == other . tuple <EOL> class UploadSet ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name = '<STR_LIT>' , extensions = DEFAULTS , default_dest = None ) : <EOL> if not name . isalnum ( ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . name = name <EOL> self . extensions = extensions <EOL> self . _config = None <EOL> self . default_dest = default_dest <EOL> @ property <EOL> def config ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _config is not None : <EOL> return self . _config <EOL> try : <EOL> return current_app . upload_set_config [ self . name ] <EOL> except AttributeError : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def url ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> base = self . config . base_url <EOL> if base is None : <EOL> return url_for ( '<STR_LIT>' , setname = self . name , <EOL> filename = filename , _external = True ) <EOL> else : <EOL> return base + filename <EOL> def path ( self , filename , folder = None ) : <EOL> """<STR_LIT>""" <EOL> if folder : <EOL> target_folder = os . path . join ( self . config . destination , folder ) <EOL> else : <EOL> target_folder = self . config . destination <EOL> return os . path . join ( target_folder , filename ) <EOL> def file_allowed ( self , storage , basename ) : <EOL> """<STR_LIT>""" <EOL> return self . extension_allowed ( extension ( basename ) ) <EOL> def extension_allowed ( self , ext ) : <EOL> """<STR_LIT>""" <EOL> return ( ( ext in self . config . allow ) or <EOL> ( ext in self . extensions and ext not in self . config . deny ) ) <EOL> def save ( self , storage , folder = None , name = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( storage , FileStorage ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if folder is None and name is not None and "<STR_LIT:/>" in name : <EOL> folder , name = name . rsplit ( "<STR_LIT:/>" , <NUM_LIT:1> ) <EOL> basename = lowercase_ext ( secure_filename ( storage . filename ) ) <EOL> if name : <EOL> if name . endswith ( '<STR_LIT:.>' ) : <EOL> basename = name + extension ( basename ) <EOL> else : <EOL> basename = name <EOL> if not self . file_allowed ( storage , basename ) : <EOL> raise UploadNotAllowed ( ) <EOL> if folder : <EOL> target_folder = os . path . join ( self . config . destination , folder ) <EOL> else : <EOL> target_folder = self . config . destination <EOL> if not os . path . exists ( target_folder ) : <EOL> os . makedirs ( target_folder ) <EOL> if os . path . exists ( os . path . join ( target_folder , basename ) ) : <EOL> basename = self . resolve_conflict ( target_folder , basename ) <EOL> target = os . path . join ( target_folder , basename ) <EOL> storage . save ( target ) <EOL> if folder : <EOL> return posixpath . join ( folder , basename ) <EOL> else : <EOL> return basename <EOL> def resolve_conflict ( self , target_folder , basename ) : <EOL> """<STR_LIT>""" <EOL> name , ext = basename . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> count = <NUM_LIT:0> <EOL> while True : <EOL> count = count + <NUM_LIT:1> <EOL> newname = '<STR_LIT>' % ( name , count , ext ) <EOL> if not os . path . exists ( os . path . join ( target_folder , newname ) ) : <EOL> return newname <EOL> if using_blueprints : <EOL> uploads_mod = Blueprint ( '<STR_LIT>' , __name__ , url_prefix = '<STR_LIT>' ) <EOL> else : <EOL> uploads_mod = Module ( __name__ , name = '<STR_LIT>' , url_prefix = '<STR_LIT>' ) <EOL> @ uploads_mod . route ( '<STR_LIT>' ) <EOL> def uploaded_file ( setname , filename ) : <EOL> config = current_app . upload_set_config . get ( setname ) <EOL> if config is None : <EOL> abort ( <NUM_LIT> ) <EOL> return send_from_directory ( config . destination , filename ) <EOL> class TestingFileStorage ( FileStorage ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stream = None , filename = None , name = None , <EOL> content_type = '<STR_LIT>' , content_length = - <NUM_LIT:1> , <EOL> headers = None ) : <EOL> FileStorage . __init__ ( self , stream , filename , name = name , <EOL> content_type = content_type , content_length = content_length , <EOL> headers = None ) <EOL> self . saved = None <EOL> def save ( self , dst , buffer_size = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( dst , basestring ) : <EOL> self . saved = dst <EOL> else : <EOL> self . saved = dst . name </s>
