在Python中实现带进度跟踪的交互式压缩，核心是利用zipfile模块完成压缩操作，同时通过自定义回调机制捕获每个文件的压缩进度，再结合进度条库将进度可视化。这种方式能让用户清晰看到每个文件的压缩状态，避免等待时的不确定性。
核心依赖库说明
实现该功能需要用到两个核心库，分别是Python标准库和第三方进度条库：
- zipfile：Python内置的压缩文件处理库，支持创建、读取、写入ZIP格式压缩包，无需额外安装。
- tqdm：用于生成进度条的第三方库，可通过pip install tqdm命令安装，支持迭代过程进度展示。
基础压缩实现思路
要实现单个文件的压缩并跟踪进度，首先需要了解zipfile的写入逻辑。zipfile模块的ZipFile类在写入文件时，会将文件分块处理，我们可以在分块写入的过程中统计已处理的字节数，从而计算当前文件的压缩进度。
自定义进度跟踪回调
我们可以定义一个回调函数，在每次写入文件块时触发，更新当前已处理的字节数，再结合文件总大小计算进度比例。以下是基础的实现框架：
import zipfile
import os
from tqdm import tqdm
def compress_single_file_with_progress(zip_obj, file_path, arcname, progress_bar):
"""
压缩单个文件并跟踪进度
:param zip_obj: ZipFile对象
:param file_path: 要压缩的文件路径
:param arcname: 压缩包内的文件名称
:param progress_bar: tqdm进度条对象
"""
file_size = os.path.getsize(file_path)
# 打开文件准备读取
with open(file_path, 'rb') as f:
# 将文件分块写入压缩包
while True:
chunk = f.read(1024 * 1024) # 每次读取1MB
if not chunk:
break
zip_obj.writestr(arcname, chunk)
# 更新进度条
progress_bar.update(len(chunk))
progress_bar.close()
# 使用示例
if __name__ == '__main__':
target_zip = 'test.zip'
file_to_compress = 'example.txt'
# 创建ZipFile对象，模式为写入
with zipfile.ZipFile(target_zip, 'w', zipfile.ZIP_DEFLATED) as zf:
# 创建进度条，总数为文件大小
pbar = tqdm(total=os.path.getsize(file_to_compress), desc=f'压缩 {file_to_compress}')
compress_single_file_with_progress(zf, file_to_compress, file_to_compress, pbar)
批量文件压缩进度跟踪
实际应用中通常需要批量压缩多个文件，此时需要同时跟踪每个文件的压缩进度，以及整体压缩的进度。我们可以遍历文件列表，为每个文件创建独立的进度条，同时维护一个整体进度统计。
以下是批量压缩的实现代码：
import zipfile
import os
from tqdm import tqdm
def batch_compress_with_progress(file_list, output_zip):
"""
批量压缩文件并跟踪每个文件的压缩进度
:param file_list: 要压缩的文件路径列表
:param output_zip: 输出的压缩包路径
"""
# 计算所有文件的总大小，用于整体进度参考
total_size = sum(os.path.getsize(f) for f in file_list if os.path.exists(f))
# 创建整体进度条
overall_pbar = tqdm(total=total_size, desc='整体压缩进度')
with zipfile.ZipFile(output_zip, 'w', zipfile.ZIP_DEFLATED) as zf:
for file_path in file_list:
if not os.path.exists(file_path):
print(f'文件 {file_path} 不存在，跳过')
continue
# 获取文件大小
file_size = os.path.getsize(file_path)
# 创建单个文件的进度条
file_pbar = tqdm(total=file_size, desc=f'压缩 {os.path.basename(file_path)}', leave=False)
try:
with open(file_path, 'rb') as f:
while True:
chunk = f.read(1024 * 1024) # 每次读取1MB
if not chunk:
break
# 写入压缩包，arcname使用相对路径避免绝对路径被写入
arcname = os.path.relpath(file_path, start=os.path.commonprefix(file_list))
zf.writestr(arcname, chunk)
# 更新单个文件进度条
file_pbar.update(len(chunk))
# 更新整体进度条
overall_pbar.update(len(chunk))
except Exception as e:
print(f'压缩文件 {file_path} 失败: {e}')
finally:
file_pbar.close()
overall_pbar.close()
# 使用示例
if __name__ == '__main__':
# 待压缩的文件列表，可自行替换
files_to_compress = ['file1.txt', 'file2.jpg', 'file3.pdf']
output_zip_path = 'batch_output.zip'
batch_compress_with_progress(files_to_compress, output_zip_path)
注意事项说明
- 使用
writestr方法分块写入时，需要注意如果同一个arcname多次调用writestr，会覆盖之前的内容，因此实际实现中可以先读取完整文件内容再分块写入，或者调整逻辑使用write方法结合回调（zipfile的write方法本身不支持直接回调，因此分块读取写入是更灵活的方式）。 - 如果压缩的文件较大，分块大小可以根据实际情况调整，比如设置为2MB或4MB，减少循环次数。
- 进度条的
leave=False参数可以让单个文件的进度条在完成之后不保留在终端，避免终端输出过于杂乱。 - 处理文件路径时，建议使用
os.path.relpath生成压缩包内的相对路径，避免将本地绝对路径写入压缩包，导致解压时出现路径问题。
上述实现方式可以灵活适配不同的压缩需求，开发者可以根据实际场景调整进度条的样式、分块大小以及异常处理逻辑，实现更符合需求的交互式压缩功能。