本教程详细介绍了如何利用 Python tqdm 库有效监控文件操作进度，特别是在批量处理（如加密/解密）场景下。我们将探讨如何计算总进度并为每个文件操作提供更新回调，从而实现对整个文件处理过程的直观进度条显示，提升用户体验。
引言：理解文件操作进度监控的挑战
在 python 中进行文件操作时，尤其是涉及大量文件或大文件时，提供一个进度条可以显著提升用户体验。tqdm 是一个功能强大的库，能够轻松为循环或迭代器添加进度条。然而，许多 tqdm 的示例都集中在网络下载场景，例如使用 requests 库的 iter_content 方法，它能以分块的方式迭代数据流，从而方便地更新进度条。
对于本地文件操作，特别是像 file.write(data) 这样一次性写入大量数据的场景，直接跟踪 write() 内部的进度会变得复杂，因为 write() 通常是一个原子操作，它会尝试一次性将所有数据写入。这意味着我们无法在单次 write() 调用过程中获得细粒度的进度更新。
本教程的目标是解决一个更常见但同样重要的场景：如何在使用 file.read() 读取整个文件并进行处理（如编码/加密），然后使用 file.write() 写回时，监控 整个文件处理过程 或 批量文件处理过程 的总进度。我们将通过计算所有文件的总大小来初始化进度条，并在每个文件处理完成后更新进度。
tqdm 库基础回顾
tqdm 库的核心思想是为可迭代对象提供一个进度条。它的基本用法非常简单：
from tqdm import tqdmimport timefor i in tqdm(range(100)): time.sleep(0.01) # 模拟耗时操作
对于更复杂的场景，我们需要手动控制进度条：
tqdm(total=…): 初始化一个进度条，并指定总工作量。pbar.update(n): 将进度条向前推进 n 个单位。pbar.close(): 关闭进度条。
在文件操作中，通常我们将文件大小作为进度单位，total 就是所有文件的总字节数，而 update(n) 则是在处理完一个大小为 n 的文件后调用。
实现批量文件处理进度监控
为了实现对批量文件处理的进度监控，我们需要两个关键步骤：首先，计算出所有待处理文件的总大小，作为 tqdm 进度条的 total 值；其次，设计一个机制，在每个文件处理完成后，能够通知进度条进行更新。
我们将通过两个辅助函数来实现这一目标：
1. 计算总文件大小：iter_files 函数
这个函数负责遍历指定文件夹及其子文件夹中的所有文件，并生成每个文件的路径和大小。这是为了在初始化 tqdm 进度条之前，能够计算出所有文件的总字节数。
import osfrom tqdm import tqdmdef iter_files(folder): """ 遍历指定文件夹及其子文件夹中的所有文件，并生成文件的路径和大小。 """ for root, _, files in os.walk(folder): for file in files: file_path = os.path.join(root, file) try: # 确保文件存在且可访问，避免PermissionError file_size = os.path.getsize(file_path) yield file_size, file_path except OSError: print(f"警告: 无法获取文件大小或访问文件: {file_path}") continue
iter_files 函数是一个生成器，它不会一次性将所有文件信息加载到内存中，这对于处理大量文件非常高效。
2. 构建带进度的迭代器：iter_with_progress 函数
这个函数是核心部分，它接收一个文件夹路径，并返回一个迭代器。这个迭代器在每次迭代时，会提供一个 done 回调函数、当前文件的大小和文件路径。done 回调函数的作用是，在当前文件处理完成后，通知 tqdm 进度条更新相应的字节数。
def iter_with_progress(folder): """ 为指定文件夹中的文件处理创建一个带进度的迭代器。 初始化一个tqdm进度条，并为每个文件提供一个更新进度的回调函数。 """ # 计算所有文件的总大小，作为tqdm的total参数 total_size = sum(s for s, _ in iter_files(folder)) # 初始化tqdm进度条 # unit='B' 表示单位是字节 # unit_scale=True 自动缩放单位（B, KB, MB, GB） # unit_divisor=1024 使用1024作为单位除数 progress_bar = tqdm(unit='B', total=total_size, unit_scale=True, unit_divisor=1024, desc="总进度") # 遍历文件，并为每个文件提供一个更新进度的回调 for size, file_path in iter_files(folder): # 定义一个闭包函数，用于在文件处理完成后更新进度条 # 每次调用done()，进度条就会增加当前文件的大小 done_callback = lambda: progress_bar.update(size) yield done_callback, size, file_path # 确保进度条在所有文件处理完成后关闭 progress_bar.close()
iter_with_progress 函数首先调用 iter_files 来计算 total_size，然后创建一个全局的 tqdm 进度条。在每次迭代时，它返回一个 done_callback，当这个回调被调用时，它会告诉 progress_bar 增加当前文件的大小。
3. 整合到文件处理逻辑
现在，我们可以将上述函数整合到您的文件加密/解密逻辑中。假设您的加密/解密操作发生在循环内部，并且每次处理一个文件。
import osfrom base64 import b85encode, b85decode # 导入base85编码/解码函数import time # 模拟耗时操作# 假设的输入目录input_dir = 'C:Python311test_files' # 请替换为您的实际目录# 确保测试目录存在并创建一些示例文件if not os.path.exists(input_dir): os.makedirs(input_dir) with open(os.path.join(input_dir, 'file1.txt'), 'w') as f: f.write("This is a test file for encryption and decryption. " * 100) with open(os.path.join(input_dir, 'file2.bin'), 'wb') as f: f.write(os.urandom(50000)) # 50KB random data print(f"Created sample files in {input_dir}")print(f"开始处理目录: {input_dir}")# 使用 iter_with_progress 迭代文件，并获取进度更新回调for done_callback, file_size, file_path in iter_with_progress(input_dir): file_name = os.path.basename(file_path) print(f"正在处理: {file_name} ({file_size} 字节)...", end='') try: # 模拟文件读取、编码/加密、写入过程 with open(file_path, 'rb') as encrypting_file: original_bytes = encrypting_file.read() # 模拟加密操作：这里使用b85encode作为示例 # 注意：b85encode会增加数据大小，但tqdm是基于原始文件大小更新的 # 如果需要精确跟踪加密后写入的字节数，需要调整total和update逻辑 encoded_bytes = b85encode(original_bytes) with open(file_path, 'wb') as output_file: output_file.write(encoded_bytes) print(f"完成处理: {file_name} ") # 清除行尾的进度信息 # 调用 done_callback 通知进度条更新当前文件大小 done_callback() except PermissionError: print(f"跳过 (权限不足): {file_name} ") # 对于跳过的文件，如果不想计入总进度，则不调用done_callback # 如果跳过也算作处理完成，则可以调用 done_callback() done_callback() # 也可以选择不调用，这样进度条会显示未完成 except Exception as e: print(f"处理 {file_name} 时发生错误: {e} ") done_callback() # 即使出错也更新进度，表示该文件已尝试处理print("所有文件处理完毕。")
在这个整合示例中，iter_with_progress(input_dir) 提供了遍历所有文件的能力，并且在每次循环迭代时，我们获得了 done_callback。在每个文件的加密（或任何处理）完成后，我们调用 done_callback()，这将通知 tqdm 进度条增加相应的文件大小，从而实现对整个批处理过程的进度监控。
注意事项与优化
内存消耗与大文件处理：您原始代码中的 b85encode(encryptingfile.read()) 会将整个文件内容读入内存。对于非常大的文件（例如几GB），这可能会导致内存溢出。优化建议： 对于大文件，应采用分块读写的方式。这意味着您需要以固定大小的块读取文件，对每个块进行处理，然后将处理后的块写入新文件。在这种情况下，tqdm.update() 的参数应该是每次写入的块大小，而不是整个文件的大小。例如：
# 伪代码：分块读写chunk_size = 4096with open(file_path, 'rb') as infile, open(output_path, 'wb') as outfile: # pbar for single file, total = file_size with tqdm(total=file_size, unit='B', unit_scale=True, desc=file_name) as pbar_file: while True: chunk = infile.read(chunk_size) if not chunk: break processed_chunk = b85encode(chunk) # 假设处理 outfile.write(processed_chunk) pbar_file.update(len(chunk)) # 更新当前文件的进度# 外部的 done_callback() 仍然可以在这里调用，表示整个文件处理完成
这种分块处理方式将提供更细粒度的单个文件内部进度。
file.write() 的原子性：再次强调，本教程的方案是监控 文件完成 或 总字节数处理 的进度，而不是单次 file.write() 操作内部的进度。如果您希望监控单个 write() 操作的内部进度，那意味着您需要手动将要写入的数据分成小块，然后循环写入并更新进度条，但这通常只在处理非常大的单一数据流时才有意义。
错误处理：在实际应用中，务必加入健壮的错误处理机制。例如，当文件不存在、权限不足或在读写过程中发生其他I/O错误时，能够优雅地处理并记录问题。示例代码中已加入了 PermissionError 的处理。
tqdm 参数的灵活运用：tqdm 提供了丰富的参数来自定义进度条的显示。例如：
desc: 进度条前缀描述。ncols: 进度条的宽度。bar_format: 自定义进度条的显示格式。leave: 控制进度条在完成后是否保留在终端。根据您的需求调整这些参数，可以使进度条更加符合您的应用场景。
总结
通过本教程，我们学习了如何利用 tqdm 库有效地监控 Python 中批量文件处理的进度。核心思想是计算所有待处理文件的总大小作为 tqdm 的 total 值，并为每个文件处理提供一个回调函数来更新进度。这种方法尤其适用于文件加密、解密、压缩、转换等需要遍历文件夹并处理每个文件的场景，极大地提升了脚本的用户友好性。同时，我们也讨论了处理大文件时的内存优化策略，即采用分块读写，以及理解 file.write() 操作的原子性对进度条实现的影响。掌握这些技术，将使您的文件处理脚本更加专业和高效。
以上就是如何使用 tqdm 监控文件批量读写与处理进度的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1363387.html
微信扫一扫 
支付宝扫一扫 
