Python如何实现Bzip2文件类似extractAll功能解压到指定文件夹
问题原因说明
BZ2是单文件压缩格式,本身没有ZIP/TAR这类归档格式的目录结构,原生没有extractall方法,可分两种场景实现相同效果:
- 如果你的BZ2文件是
xxx.tar.bz2格式(先tar归档再用BZ2压缩),直接用tarfile库即可,原生支持extractall方法 - 如果是单独的
xxx.bz2格式(仅压缩单个文件),修复你现有代码的逻辑问题即可实现将解压文件输出到指定目录的效果
你现有代码的核心问题是:调用os.mkdir(path_zip)创建目录后,直接把目录路径当做文件路径写入内容,属于非法操作会触发报错,只需要把解压后的内容写入到这个新建目录下的文件即可。
适配代码
场景1:处理.tar.bz2格式(和tar.gz逻辑一致)
def tar_bz2_download(*args): import tarfile import requests import os mystring_bz2 = ' '.join(args) file_base_name = mystring_bz2.split('/')[-1].split('.')[0] print(f'{file_base_name} Tar.bz2 file Download Started!!!') r = requests.get(mystring_bz2) filename = mystring_bz2.split('/')[-1] with open(filename, 'wb') as output_file: output_file.write(r.content) print(f' {file_base_name} Tar.bz2 file Download Completed!!!') extract_dir = file_base_name print(f'---------------------------{extract_dir} extraction Started----------------------------------') # 仅修改mode参数为r:bz2即可 thetarfile = tarfile.open(filename, mode="r:bz2") thetarfile.extractall(extract_dir) thetarfile.close() print(f'---------------------------{extract_dir} extraction Complete----------------------------------')
场景2:处理单独的.bz2格式
def single_bz2_download(*args): import _bz2 import requests import os mystring_zip = ' '.join(args) file_full_name = mystring_zip.split('/')[-1] # 去掉.bz2后缀得到解压后的文件名 unzip_file_name = file_full_name.rsplit('.', 1)[0] print(f'{unzip_file_name} Bz2 file Download Started!!!') r = requests.get(mystring_zip) with open(file_full_name, 'wb') as output_file: output_file.write(r.content) print(f'{unzip_file_name} Bz2 file Download Completed!!!') extract_dir = unzip_file_name print(f'--------------------------{extract_dir} extraction Started----------------------------------') # 创建目标目录,exist_ok参数避免目录已存在时报错 os.makedirs(extract_dir, exist_ok=True) # 拼接得到目标目录下的解压文件路径 target_file_path = os.path.join(extract_dir, unzip_file_name) total_write = 0 with open(target_file_path, 'wb') as new_file, open(file_full_name, 'rb') as file: decompressor = _bz2.BZ2Decompressor() for data in iter(lambda: file.read(100 * 1024), b''): total_write += new_file.write(decompressor.decompress(data)) if total_write > 0: print(f'---------------------------{extract_dir} extraction completed----------------------------------') else: print(f'---------------------------{extract_dir} extraction Failed----------------------------------')
内容的提问来源于stack exchange,提问作者aditya
相关产品推荐
相关产品推荐

