<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import os <EOL> import re <EOL> from gslib . exception import InvalidUrlError <EOL> PROVIDER_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> BUCKET_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> OBJECT_REGEX = re . compile ( <EOL> r'<STR_LIT>' ) <EOL> GS_GENERATION_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> S3_VERSION_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> FILE_OBJECT_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> BUCKET_NAME_RE = re . compile ( r'<STR_LIT>' ) <EOL> TOO_LONG_DNS_NAME_COMP = re . compile ( r'<STR_LIT>' ) <EOL> WILDCARD_REGEX = re . compile ( r'<STR_LIT>' ) <EOL> class StorageUrl ( object ) : <EOL> """<STR_LIT>""" <EOL> def Clone ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def IsFileUrl ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def IsCloudUrl ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def IsStream ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def CreatePrefixUrl ( self , wildcard_suffix = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> @ property <EOL> def url_string ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> @ property <EOL> def versionless_url_string ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def __eq__ ( self , other ) : <EOL> return isinstance ( other , StorageUrl ) and self . url_string == other . url_string <EOL> def __hash__ ( self ) : <EOL> return hash ( self . url_string ) <EOL> class _FileUrl ( StorageUrl ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url_string , is_stream = False ) : <EOL> self . scheme = '<STR_LIT:file>' <EOL> self . bucket_name = '<STR_LIT>' <EOL> match = FILE_OBJECT_REGEX . match ( url_string ) <EOL> if match and match . lastindex == <NUM_LIT:2> : <EOL> self . object_name = match . group ( <NUM_LIT:2> ) <EOL> else : <EOL> self . object_name = url_string <EOL> self . generation = None <EOL> self . is_stream = is_stream <EOL> self . delim = os . sep <EOL> def Clone ( self ) : <EOL> return _FileUrl ( self . url_string ) <EOL> def IsFileUrl ( self ) : <EOL> return True <EOL> def IsCloudUrl ( self ) : <EOL> return False <EOL> def IsStream ( self ) : <EOL> return self . is_stream <EOL> def IsDirectory ( self ) : <EOL> return not self . IsStream ( ) and os . path . isdir ( self . object_name ) <EOL> def CreatePrefixUrl ( self , wildcard_suffix = None ) : <EOL> return self . url_string <EOL> @ property <EOL> def url_string ( self ) : <EOL> return '<STR_LIT>' % ( self . scheme , self . object_name ) <EOL> @ property <EOL> def versionless_url_string ( self ) : <EOL> return self . url_string <EOL> def __str__ ( self ) : <EOL> return self . url_string <EOL> class _CloudUrl ( StorageUrl ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url_string ) : <EOL> self . scheme = None <EOL> self . bucket_name = None <EOL> self . object_name = None <EOL> self . generation = None <EOL> self . delim = '<STR_LIT:/>' <EOL> provider_match = PROVIDER_REGEX . match ( url_string ) <EOL> bucket_match = BUCKET_REGEX . match ( url_string ) <EOL> if provider_match : <EOL> self . scheme = provider_match . group ( '<STR_LIT>' ) <EOL> elif bucket_match : <EOL> self . scheme = bucket_match . group ( '<STR_LIT>' ) <EOL> self . bucket_name = bucket_match . group ( '<STR_LIT>' ) <EOL> if ( not ContainsWildcard ( self . bucket_name ) and <EOL> ( not BUCKET_NAME_RE . match ( self . bucket_name ) or <EOL> TOO_LONG_DNS_NAME_COMP . search ( self . bucket_name ) ) ) : <EOL> raise InvalidUrlError ( '<STR_LIT>' % url_string ) <EOL> else : <EOL> object_match = OBJECT_REGEX . match ( url_string ) <EOL> if object_match : <EOL> self . scheme = object_match . group ( '<STR_LIT>' ) <EOL> self . bucket_name = object_match . group ( '<STR_LIT>' ) <EOL> self . object_name = object_match . group ( '<STR_LIT:object>' ) <EOL> if self . scheme == '<STR_LIT>' : <EOL> generation_match = GS_GENERATION_REGEX . match ( self . object_name ) <EOL> if generation_match : <EOL> self . object_name = generation_match . group ( '<STR_LIT:object>' ) <EOL> self . generation = generation_match . group ( '<STR_LIT>' ) <EOL> elif self . scheme == '<STR_LIT>' : <EOL> version_match = S3_VERSION_REGEX . match ( self . object_name ) <EOL> if version_match : <EOL> self . object_name = version_match . group ( '<STR_LIT:object>' ) <EOL> self . generation = version_match . group ( '<STR_LIT>' ) <EOL> else : <EOL> raise InvalidUrlError ( <EOL> '<STR_LIT>' % url_string ) <EOL> def Clone ( self ) : <EOL> return _CloudUrl ( self . url_string ) <EOL> def IsFileUrl ( self ) : <EOL> return False <EOL> def IsCloudUrl ( self ) : <EOL> return True <EOL> def IsStream ( self ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def IsBucket ( self ) : <EOL> return bool ( self . bucket_name and not self . object_name ) <EOL> def IsObject ( self ) : <EOL> return bool ( self . bucket_name and self . object_name ) <EOL> def HasGeneration ( self ) : <EOL> return bool ( self . generation ) <EOL> def IsProvider ( self ) : <EOL> return bool ( self . scheme and not self . bucket_name ) <EOL> def CreatePrefixUrl ( self , wildcard_suffix = None ) : <EOL> prefix = StripOneSlash ( self . versionless_url_string ) <EOL> if wildcard_suffix : <EOL> prefix = '<STR_LIT>' % ( prefix , wildcard_suffix ) <EOL> return prefix <EOL> @ property <EOL> def bucket_url_string ( self ) : <EOL> return '<STR_LIT>' % ( self . scheme , self . bucket_name ) <EOL> @ property <EOL> def url_string ( self ) : <EOL> url_str = self . versionless_url_string <EOL> if self . HasGeneration ( ) : <EOL> url_str += '<STR_LIT>' % self . generation <EOL> return url_str <EOL> @ property <EOL> def versionless_url_string ( self ) : <EOL> if self . IsProvider ( ) : <EOL> return '<STR_LIT>' % self . scheme <EOL> elif self . IsBucket ( ) : <EOL> return self . bucket_url_string <EOL> return '<STR_LIT>' % ( self . scheme , self . bucket_name , self . object_name ) <EOL> def __str__ ( self ) : <EOL> return self . url_string <EOL> def _GetSchemeFromUrlString ( url_str ) : <EOL> """<STR_LIT>""" <EOL> end_scheme_idx = url_str . find ( '<STR_LIT>' ) <EOL> if end_scheme_idx == - <NUM_LIT:1> : <EOL> return '<STR_LIT:file>' <EOL> else : <EOL> return url_str [ <NUM_LIT:0> : end_scheme_idx ] . lower ( ) <EOL> def _GetPathFromUrlString ( url_str ) : <EOL> """<STR_LIT>""" <EOL> end_scheme_idx = url_str . find ( '<STR_LIT>' ) <EOL> if end_scheme_idx == - <NUM_LIT:1> : <EOL> return url_str <EOL> else : <EOL> return url_str [ end_scheme_idx + <NUM_LIT:3> : ] <EOL> def IsFileUrlString ( url_str ) : <EOL> """<STR_LIT>""" <EOL> return _GetSchemeFromUrlString ( url_str ) == '<STR_LIT:file>' <EOL> def StorageUrlFromString ( url_str ) : <EOL> """<STR_LIT>""" <EOL> scheme = _GetSchemeFromUrlString ( url_str ) <EOL> if scheme not in ( '<STR_LIT:file>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> raise InvalidUrlError ( '<STR_LIT>' % scheme ) <EOL> if scheme == '<STR_LIT:file>' : <EOL> path = _GetPathFromUrlString ( url_str ) <EOL> is_stream = ( path == '<STR_LIT:->' ) <EOL> return _FileUrl ( url_str , is_stream = is_stream ) <EOL> return _CloudUrl ( url_str ) <EOL> def StripOneSlash ( url_str ) : <EOL> if url_str and url_str . endswith ( '<STR_LIT:/>' ) : <EOL> return url_str [ : - <NUM_LIT:1> ] <EOL> return url_str <EOL> def ContainsWildcard ( url_string ) : <EOL> """<STR_LIT>""" <EOL> return bool ( WILDCARD_REGEX . search ( url_string ) ) </s>
