<s> from conans . errors import ConanException , ConanConnectionError <EOL> from requests . exceptions import ConnectionError <EOL> from conans . util . files import build_files_set , save , tar_extract , rmdir <EOL> from conans . util . log import logger <EOL> import traceback <EOL> import os <EOL> from conans . paths import PACKAGE_TGZ_NAME , CONANINFO , CONAN_MANIFEST , CONANFILE , EXPORT_TGZ_NAME <EOL> from io import StringIO , BytesIO <EOL> import tarfile <EOL> from conans . util . files import gzopen_without_timestamps <EOL> from conans . util . files import touch <EOL> class RemoteManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , paths , remote_client , output ) : <EOL> self . _paths = paths <EOL> self . _output = output <EOL> self . _remote_client = remote_client <EOL> def upload_conan ( self , conan_reference , remote ) : <EOL> """<STR_LIT>""" <EOL> basedir = self . _paths . export ( conan_reference ) <EOL> rel_files = self . _paths . export_paths ( conan_reference ) <EOL> the_files = build_files_set ( basedir , rel_files ) <EOL> the_files = compress_export_files ( the_files ) <EOL> return self . _call_remote ( remote , "<STR_LIT>" , conan_reference , the_files ) <EOL> def upload_package ( self , package_reference , remote ) : <EOL> """<STR_LIT>""" <EOL> basedir = self . _paths . package ( package_reference ) <EOL> rel_files = self . _paths . package_paths ( package_reference ) <EOL> the_files = build_files_set ( basedir , rel_files ) <EOL> self . _output . rewrite_line ( "<STR_LIT>" ) <EOL> the_files = compress_package_files ( the_files ) <EOL> return self . _call_remote ( remote , "<STR_LIT>" , package_reference , the_files ) <EOL> def get_conan_digest ( self , conan_reference , remote ) : <EOL> """<STR_LIT>""" <EOL> return self . _call_remote ( remote , "<STR_LIT>" , conan_reference ) <EOL> def get_package_digest ( self , package_reference , remote ) : <EOL> """<STR_LIT>""" <EOL> return self . _call_remote ( remote , "<STR_LIT>" , package_reference ) <EOL> def get_conanfile ( self , conan_reference , remote ) : <EOL> """<STR_LIT>""" <EOL> export_files = self . _call_remote ( remote , "<STR_LIT>" , conan_reference ) <EOL> export_folder = self . _paths . export ( conan_reference ) <EOL> uncompress_files ( export_files , export_folder , EXPORT_TGZ_NAME ) <EOL> rmdir ( self . _paths . source ( conan_reference ) ) <EOL> def get_package ( self , package_reference , remote ) : <EOL> """<STR_LIT>""" <EOL> package_files = self . _call_remote ( remote , "<STR_LIT>" , package_reference ) <EOL> destination_dir = self . _paths . package ( package_reference ) <EOL> uncompress_files ( package_files , destination_dir , PACKAGE_TGZ_NAME ) <EOL> for dirname , _ , files in os . walk ( destination_dir ) : <EOL> for fname in files : <EOL> touch ( os . path . join ( dirname , fname ) ) <EOL> def search ( self , remote , pattern = None , ignorecase = True ) : <EOL> """<STR_LIT>""" <EOL> return self . _call_remote ( remote , "<STR_LIT>" , pattern , ignorecase ) <EOL> def remove ( self , conan_ref , remote ) : <EOL> """<STR_LIT>""" <EOL> return self . _call_remote ( remote , "<STR_LIT>" , conan_ref ) <EOL> def remove_packages ( self , conan_ref , remove_ids , remote ) : <EOL> """<STR_LIT>""" <EOL> return self . _call_remote ( remote , "<STR_LIT>" , conan_ref , remove_ids ) <EOL> def authenticate ( self , remote , name , password ) : <EOL> return self . _call_remote ( remote , '<STR_LIT>' , name , password ) <EOL> def _call_remote ( self , remote , method , * argc , ** argv ) : <EOL> self . _remote_client . remote = remote <EOL> try : <EOL> return getattr ( self . _remote_client , method ) ( * argc , ** argv ) <EOL> except ConnectionError as exc : <EOL> raise ConanConnectionError ( "<STR_LIT>" % ( remote . name , remote . url ) ) <EOL> except ConanException : <EOL> raise <EOL> except Exception as exc : <EOL> logger . error ( traceback . format_exc ( ) ) <EOL> raise ConanException ( exc ) <EOL> def compress_package_files ( files ) : <EOL> return compress_files ( files , PACKAGE_TGZ_NAME , excluded = ( CONANINFO , CONAN_MANIFEST ) ) <EOL> def compress_export_files ( files ) : <EOL> return compress_files ( files , EXPORT_TGZ_NAME , excluded = ( CONANFILE , CONAN_MANIFEST ) ) <EOL> def compress_files ( files , name , excluded ) : <EOL> """<STR_LIT>""" <EOL> tgz_contents = BytesIO ( ) <EOL> tgz = gzopen_without_timestamps ( name , mode = "<STR_LIT:w>" , fileobj = tgz_contents ) <EOL> def addfile ( name , file_info , tar ) : <EOL> info = tarfile . TarInfo ( name = name ) <EOL> the_str = BytesIO ( file_info [ "<STR_LIT>" ] ) <EOL> info . size = len ( file_info [ "<STR_LIT>" ] ) <EOL> info . mode = file_info [ "<STR_LIT>" ] <EOL> tar . addfile ( tarinfo = info , fileobj = the_str ) <EOL> for the_file , info in files . items ( ) : <EOL> if the_file not in excluded : <EOL> addfile ( the_file , info , tgz ) <EOL> tgz . close ( ) <EOL> ret = { } <EOL> for e in excluded : <EOL> if e in files : <EOL> ret [ e ] = files [ e ] [ "<STR_LIT>" ] <EOL> ret [ name ] = tgz_contents . getvalue ( ) <EOL> return ret <EOL> def uncompress_files ( files , folder , name ) : <EOL> for file_name , content in files : <EOL> if os . path . basename ( file_name ) != name : <EOL> save ( os . path . join ( folder , file_name ) , content ) <EOL> else : <EOL> tar_extract ( BytesIO ( content ) , folder ) </s>
