Python shutil 模块，unpack_archive() 实例源码
我们从Python开源项目中，提取了以下18个代码示例，用于说明如何使用shutil.unpack_archive()。
def wishbone_my_setup(setup_dir):
# install GSEA, diffusion components, and download data.
tools_dir = setup_dir + '/tools'
if not os.path.exists(tools_dir + '/DiffusionGeometry/'):
shutil.unpack_archive(tools_dir + '/DiffusionGeometry.zip', tools_dir +
'/DiffusionGeometry/')
if not os.path.exists(tools_dir + '/mouse/'):
shutil.unpack_archive(tools_dir + '/mouse_gene_sets.tar.gz', tools_dir)
if not os.path.exists(tools_dir + '/human/'):
shutil.unpack_archive(tools_dir + '/human_gene_sets.tar.gz', tools_dir)
if not os.path.exists( setup_dir +'/data/GSE72857_umitab.txt.gz'):
# downloads mouse UMI from GSE72857 Transcriptional heterogeneity and lineage commitment in myeloid progenitors [single cell RNA-seq]
os.system("wget -m -nH -nd -P "+ setup_dir + "/data/ ftp://ftp.ncbi.nlm.nih.gov/geo/series/GSE72nnn/GSE72857/suppl/GSE72857%5Fumitab%2Etxt%2Egz")
x=pd.read_csv(setup_dir + '/data/GSE72857_umitab.txt.gz', sep = '\t', compression="gzip")
y=pd.read_csv(setup_dir + '/data/sample_scseq_data.csv', index_col=[0])
scdata_raw= x.T.loc[y.index]
scdata_raw = wishbone.wb.SCData(scdata_raw.astype('float'), data_type='sc-seq')
return(scdata_raw)
def _restore_from_backup(self, version):
logger.debug('Restoring from backup for {}'.format(version))
for file_path in self.files_to_preserve:
try:
shutil.copy2(os.path.join(self.backup_target, file_path),
os.path.join(self.tmp_dir, file_path))
except IOError as e:
logger.warning('Copying {} failed due to {}'
.format(file_path, e))
shutil.unpack_archive(self._backup_name_ext(
version), self.backup_target, self.backup_format)
for file_path in self.files_to_preserve:
try:
shutil.copy2(os.path.join(self.tmp_dir, file_path),
os.path.join(self.backup_target, file_path))
except IOError as e:
logger.warning('Copying {} failed due to {}'
.format(file_path, e))
shutil.rmtree(self.tmp_dir, ignore_errors=True)
def execute(self, context):
name = os.path.basename(self.filepath)
#TODO: put on another file and module
def checkProjectFile(path):
if os.path.basename(path) != "project.json": return False
if not os.path.isfile(path): return False
utils.loadProjectFile(path)
if not 'bge-project' in utils.project_data: return False
return True
def getMainBlend(path):
path = os.path.dirname(path) + os.sep + "project" + os.sep + "main.blend"
if os.path.isfile(path): return path
#If we have a zip
if os.path.isfile(self.filepath):
for x in [".zip", ".tar", ".gztar", ".bztar", ".xztar"]:
if self.filepath.endswith(x):
tpath = tempfile.gettempdir() + os.sep + os.path.basename(self.filepath)[:-len(x)] + os.sep
shutil.unpack_archive(self.filepath, tpath)
self.filepath = tpath
print("Extracted to: ", tpath)
#Once we have a folder...
list=[self.filepath, self.filepath + "project.json", os.path.dirname(self.filepath) + os.sep + "project.json", os.path.dirname(self.filepath) + os.sep + "../" + "project.json"]
endpath=None
for path in list:
if checkProjectFile(path): endpath=getMainBlend(path)
if endpath==None:
self.report({'ERROR_INVALID_INPUT'}, "Error loading project, project file not found.")
return {'CANCELLED'}
bpy.ops.wm.open_mainfile(filepath=endpath)
return {'FINISHED'}
def untar(path, fname, deleteTar=True):
"""Unpacks the given archive file to the same directory, then (by default)
deletes the archive file.
"""
print('unpacking ' + fname)
fullpath = os.path.join(path, fname)
shutil.unpack_archive(fullpath, path)
if deleteTar:
os.remove(fullpath)
def extract_neo4j(database_name, archive_path):
database_directory = os.path.join(settings.POLYGLOT_DATA_DIRECTORY, database_name)
neo4j_directory = os.path.join(database_directory, 'neo4j')
if os.path.exists(neo4j_directory):
return False
shutil.unpack_archive(archive_path, database_directory)
for d in os.listdir(database_directory):
if d.startswith('neo4j'):
os.rename(os.path.join(database_directory, d), neo4j_directory)
return True
def extract_influxdb(database_name, archive_path):
database_directory = os.path.join(settings.POLYGLOT_DATA_DIRECTORY, database_name)
influxdb_directory = os.path.join(database_directory, 'influxdb')
if os.path.exists(influxdb_directory):
return False
shutil.unpack_archive(archive_path, database_directory)
for d in os.listdir(database_directory):
if d.startswith('influxdb'):
os.rename(os.path.join(database_directory, d), influxdb_directory)
return True
def zipfile_unzip(file_path=None, extract_dir=None, **filedialog_kwargs):
if file_path is None:
file_path = QtGui.QFileDialog.getOpenFileName(**filedialog_kwargs)[0]
if extract_dir is None:
extract_dir = os.path.dirname(file_path)
shutil.unpack_archive(file_path, extract_dir=extract_dir)
def unpack_archive(file, extract_dir, format_, msg):
try:
shutil.unpack_archive(file, extract_dir=extract_dir, format=format_)
except (ValueError, OSError) as err:
logging.debug(traceback.format_exc())
logging.critical(err)
logging.critical(msg)
raise SystemExit
def _get_url_unpacked_path_or_null(url):
parsed = urllib.parse.urlparse(url)
if parsed.scheme == "file" and parsed.path.endswith(".whl"):
return Path("/dev/null")
try:
cache_dir, packed_path = _get_url_impl(url)
except CalledProcessError:
return Path("/dev/null")
if packed_path.is_file(): # pip://
shutil.unpack_archive(str(packed_path), cache_dir.name)
unpacked_path, = (
path for path in Path(cache_dir.name).iterdir() if path.is_dir())
return unpacked_path
def untar(fname):
print('unpacking ' + fname)
fullpath = os.path.join(fname)
shutil.unpack_archive(fullpath)
os.remove(fullpath)
def untar(path, fname):
print('unpacking ' + fname)
fullpath = os.path.join(path, fname)
shutil.unpack_archive(fullpath, path)
os.remove(fullpath)
def untar(fname):
print('unpacking ' + fname)
fullpath = os.path.join(fname)
shutil.unpack_archive(fullpath)
os.remove(fullpath)
def untar(fname):
print('unpacking ' + fname)
fullpath = os.path.join(fname)
shutil.unpack_archive(fullpath)
os.remove(fullpath)
def untar(fname):
print('unpacking ' + fname)
fullpath = os.path.join(fname)
shutil.unpack_archive(fullpath)
os.remove(fullpath)
def maybe_download_and_extract(data_root: str, url: str) -> None:
"""
Maybe download the specified file to ``data_root`` and try to unpack it with ``shutil.unpack_archive``.
:param data_root: data root to download the files to
:param url: url to download from
"""
# make sure data_root exists
os.makedirs(data_root, exist_ok=True)
filename = os.path.basename(url)
# check whether the archive already exists
filepath = os.path.join(data_root, filename)
if os.path.exists(filepath):
logging.info('\t`%s` already exists; skipping', filepath)
return
# download with progressbar
logging.info('\tdownloading %s', filepath)
req = requests.get(url, stream=True)
expected_size = int(req.headers.get('content-length'))
chunk_size = 1024
with open(filepath, 'wb') as f_out,\
click.progressbar(req.iter_content(chunk_size=chunk_size), length=expected_size/chunk_size) as bar:
for chunk in bar:
if chunk:
f_out.write(chunk)
f_out.flush()
# extract
try:
shutil.unpack_archive(filepath, data_root)
except (shutil.ReadError, ValueError):
logging.info('File `%s` could not be extracted by `shutil.unpack_archive`. Please process it manually.',
filepath)
def execute(self, context):
import html.parser
import urllib.request
remote_platforms = []
ps = context.scene.ge_publish_settings
# create lib folder if not already available
lib_path = bpy.path.abspath(ps.lib_path)
if not os.path.exists(lib_path):
os.makedirs(lib_path)
print("Retrieving list of platforms from blender.org...", end=" ", flush=True)
class AnchorParser(html.parser.HTMLParser):
def handle_starttag(self, tag, attrs):
if tag == 'a':
for key, value in attrs:
if key == 'href' and value.startswith('blender'):
remote_platforms.append(value)
url = 'http://download.blender.org/release/Blender' + bpy.app.version_string.split()[0]
parser = AnchorParser()
data = urllib.request.urlopen(url).read()
parser.feed(str(data))
print("done", flush=True)
print("Downloading files (this will take a while depending on your internet connection speed).", flush=True)
for i in remote_platforms:
src = '/'.join((url, i))
dst = os.path.join(lib_path, i)
dst_dir = '.'.join([i for i in dst.split('.') if i not in {'zip', 'tar', 'bz2'}])
if not os.path.exists(dst) and not os.path.exists(dst.split('.')[0]):
print("Downloading " + src + "...", end=" ", flush=True)
urllib.request.urlretrieve(src, dst)
print("done", flush=True)
else:
print("Reusing existing file: " + dst, flush=True)
print("Unpacking " + dst + "...", end=" ", flush=True)
if os.path.exists(dst_dir):
shutil.rmtree(dst_dir)
shutil.unpack_archive(dst, dst_dir)
print("done", flush=True)
print("Creating platform from libs...", flush=True)
bpy.ops.scene.publish_auto_platforms()
return {'FINISHED'}
def execute(self, context):
import html.parser
import urllib.request
remote_platforms = []
ps = context.scene.ge_publish_settings
# create lib folder if not already available
lib_path = bpy.path.abspath(ps.lib_path)
if not os.path.exists(lib_path):
os.makedirs(lib_path)
print("Retrieving list of platforms from blender.org...", end=" ", flush=True)
class AnchorParser(html.parser.HTMLParser):
def handle_starttag(self, tag, attrs):
if tag == 'a':
for key, value in attrs:
if key == 'href' and value.startswith('blender'):
remote_platforms.append(value)
url = 'http://download.blender.org/release/Blender' + bpy.app.version_string.split()[0]
parser = AnchorParser()
data = urllib.request.urlopen(url).read()
parser.feed(str(data))
print("done", flush=True)
print("Downloading files (this will take a while depending on your internet connection speed).", flush=True)
for i in remote_platforms:
src = '/'.join((url, i))
dst = os.path.join(lib_path, i)
dst_dir = '.'.join([i for i in dst.split('.') if i not in {'zip', 'tar', 'bz2'}])
if not os.path.exists(dst) and not os.path.exists(dst.split('.')[0]):
print("Downloading " + src + "...", end=" ", flush=True)
urllib.request.urlretrieve(src, dst)
print("done", flush=True)
else:
print("Reusing existing file: " + dst, flush=True)
print("Unpacking " + dst + "...", end=" ", flush=True)
if os.path.exists(dst_dir):
shutil.rmtree(dst_dir)
shutil.unpack_archive(dst, dst_dir)
print("done", flush=True)
print("Creating platform from libs...", flush=True)
bpy.ops.scene.publish_auto_platforms()
return {'FINISHED'}
def fetch(source, destination, allow_file=False):
"""
Fetch a group of files either from a file archive or version control
repository.
Supported URL schemes:
- file
- ftp
- ftps
- git
- git+http
- git+https
- git+ssh
- http
- https
:param str source: The source URL to retrieve.
:param str destination: The directory into which the files should be placed.
:param bool allow_file: Whether or not to permit the file:// URL for processing local resources.
:return: The destination directory that was used.
:rtype: str
"""
source = source.strip()
if os.path.exists(destination):
raise ValueError('destination must not be an existing directory')
parsed_url = urllib.parse.urlparse(source, scheme='file')
if parsed_url.username is not None:
# if the username is not none, then the password will be a string
creds = Creds(parsed_url.username, parsed_url.password or '')
else:
creds = Creds(None, None)
parsed_url = collections.OrderedDict(zip(('scheme', 'netloc', 'path', 'params', 'query', 'fragment'), parsed_url))
parsed_url['netloc'] = parsed_url['netloc'].split('@', 1)[-1]
parsed_url['scheme'] = parsed_url['scheme'].lower()
if parsed_url['scheme'] == 'file':
if not allow_file:
raise RuntimeError('file: URLs are not allowed to be processed')
tmp_path = parsed_url['path']
if os.path.isdir(tmp_path):
shutil.copytree(tmp_path, destination, symlinks=True)
elif os.path.isfile(tmp_path):
shutil.unpack_archive(tmp_path, destination)
else:
tmp_fd, tmp_path = tempfile.mkstemp(suffix='_' + os.path.basename(parsed_url['path']))
os.close(tmp_fd)
tmp_file = open(tmp_path, 'wb')
try:
_fetch_remote(source, destination, parsed_url, creds, tmp_file, tmp_path)
if os.stat(tmp_path).st_size:
shutil.unpack_archive(tmp_path, destination)
finally:
tmp_file.close()
os.remove(tmp_path)
return