1. 理解进度条的需求与tqdm的适用性
在进行文件操作，尤其是批量处理文件（如加密、解密、压缩、传输等）时，用户通常希望看到操作的实时进度，而非仅仅等待程序完成。tqdm是一个功能强大且易于使用的python库，能够为循环迭代器添加智能进度条。
原始需求是追踪file.write()操作的进度。然而，需要明确的是，Python的file.write(data)通常是一个原子操作，它一次性将所有数据写入文件。tqdm本身无法在单次write()调用内部追踪字节级别的进度。tqdm更适用于迭代过程，即在每次迭代完成时更新进度。
因此，对于文件操作的进度追踪，更实际和常见的做法是：
- 追踪多文件的整体处理进度：当处理一个文件夹中的多个文件时，每完成一个文件的处理，就更新一次进度条，累计已处理的文件大小。
- 追踪大文件的分块写入进度：如果一个文件非常大，且其写入操作可以被分解为多次小块写入，那么可以在每次小块写入后更新进度条。
本文将主要聚焦于第一种场景，即如何利用tqdm有效追踪文件夹内多个文件的处理进度。
2. 构建文件迭代器以获取总大小
为了使用tqdm显示总进度，我们首先需要知道所有待处理文件的总大小。这可以通过遍历目标文件夹及其子文件夹中的所有文件来实现。
import os
import time
from tqdm import tqdm
def iter_files(folder):
"""
递归遍历指定文件夹中的所有文件，并生成每个文件的路径和大小。
Args:
folder (str): 待遍历的根文件夹路径。
Yields:
tuple: (文件大小, 文件路径)
"""
for root, _, files in os.walk(folder):
for file in files:
file_path = os.path.join(root, file)
# 尝试获取文件大小，如果遇到权限问题或其他错误则跳过
try:
file_size = os.path.getsize(file_path)
yield file_size, file_path
except OSError as e:
print(f"警告: 无法访问文件 {file_path}，已跳过。错误: {e}")
continueiter_files函数是一个生成器，它会逐个返回文件的大小和完整路径。这不仅可以用于计算总大小，也可以作为后续处理的输入。
3. 集成tqdm实现进度追踪
有了获取文件大小的迭代器，我们就可以将其与tqdm结合，创建一个能够追踪整体进度的迭代器。
def iter_with_progress(folder):
"""
为指定文件夹中的文件处理过程添加tqdm进度条。
Args:
folder (str): 待处理的根文件夹路径。
Yields:
tuple: (完成回调函数, 文件大小, 文件路径)
"""
# 第一次遍历，计算所有文件的总大小
all_files_data = list(iter_files(folder))
total_size = sum(s for s, _ in all_files_data)
# 初始化tqdm进度条
# unit='B' 表示单位是字节
# total=total_size 设置总进度为所有文件总大小
# unit_scale=True 自动缩放单位 (B, KB, MB, GB)
# unit_divisor=1024 使用1024作为单位除数
progress = tqdm(unit='B',
total=total_size,
unit_scale=True,
unit_divisor=1024,
desc="处理文件") # 进度条描述
for size, file_path in all_files_data:
# 定义一个lambda函数作为“完成回调”，用于在文件处理完成后更新进度条
# 每次调用done()，进度条就会增加当前文件的大小
done = lambda current_file_size=size: progress.update(current_file_size)
yield done, size, file_path
# 确保在所有文件处理完毕后关闭进度条
progress.close()iter_with_progress函数是核心。它首先通过调用iter_files获取所有文件信息并计算出总大小，然后用这个总大小初始化tqdm进度条。最关键的是，它在每次迭代时返回一个done回调函数。这个done函数是一个闭包，它捕获了当前文件的大小。当外部代码完成对当前文件的处理后，调用done()即可更新进度条。
4. 实际应用示例
现在，我们可以将上述函数应用于一个模拟的文件处理场景，例如文件加密或解密。
# 假设我们有一个名为 'test_data' 的文件夹，里面包含一些文件
# 在实际使用中，请替换为您的目标文件夹路径
# 例如: target_folder = 'C:\Users\YourUser\Documents\MyFiles'
# 为了演示，我们创建一个虚拟文件夹和文件
if not os.path.exists('test_data'):
os.makedirs('test_data')
with open('test_data/file1.txt', 'w') as f:
f.write('This is file 1 content.' * 100)
with open('test_data/file2.bin', 'wb') as f:
f.write(os.urandom(5000)) # 5KB random data
with open('test_data/file3.log', 'w') as f:
f.write('Log entry
' * 200)
target_folder = 'test_data' # 替换为你的目标文件夹
print(f"开始处理文件夹: {target_folder}")
# 遍历并处理文件
for done_callback, file_size, file_path in iter_with_progress(target_folder):
print(f"
正在模拟处理文件: {os.path.basename(file_path)} ({file_size} 字节)")
# 在这里执行您的实际文件操作，例如加密、解密、复制等
# 为了演示，我们用 time.sleep() 模拟耗时操作
time.sleep(file_size / 10000.0) # 根据文件大小模拟处理时间
# 当文件处理完成后，调用 done_callback 来更新进度条
done_callback()
print(f"文件 {os.path.basename(file_path)} 处理完成。")
print("
所有文件处理完毕！")
# 清理演示文件 (可选)
# import shutil
# shutil.rmtree('test_data')在这个示例中，iter_with_progress返回的done_callback在每次文件处理完成后被调用。这使得tqdm能够精确地根据已处理文件的总大小来更新进度条，给用户提供清晰的整体进度反馈。
5. 注意事项与进阶
权限问题：在遍历文件时，可能会遇到权限不足导致无法访问文件的情况。在iter_files函数中，我们已经添加了try-except块来捕获OSError并跳过此类文件，同时打印警告信息。
内存消耗：如果文件夹中包含数百万个小文件，list(iter_files(folder))可能会消耗大量内存来存储所有文件路径和大小。对于极端情况，可以考虑分批处理或更复杂的迭代策略。
-
单文件内部进度：如前所述，如果需要追踪单个大文件的字节级写入进度（例如，将一个大文件分块写入），你需要修改你的文件写入逻辑，将其分解为多次小块写入，并在每次小块写入后手动调用pbar.update(chunk_size)。例如：
# 示例：追踪大文件分块写入进度 # 假设 b85_bytes 是一个非常大的字节串 # output_file 是一个已打开的文件对象 chunk_size = 1024 * 1024 # 1MB total_bytes = len(b85_bytes) with tqdm(total=total_bytes, unit='B', unit_scale=True, desc="写入文件") as pbar: for i in range(0, total_bytes, chunk_size): chunk = b85_bytes[i:i + chunk_size] output_file.write(chunk) pbar.update(len(chunk))这种方式更符合原始问题中希望追踪output_file.write(b85_bytes)进度的意图，但它要求你能够将待写入的数据分块。如果b85encode一次性返回所有数据，那么你需要手动将其分块后迭代写入。
6. 总结
tqdm是一个强大的工具，可以显著提升命令行应用程序的用户体验。通过本文介绍的方法，你可以有效地为批量文件处理任务添加专业的进度条。核心思想是计算所有待处理项的总量，并确保在每次处理完成后更新tqdm实例。理解tqdm的工作原理以及其在不同场景下的适用性，是编写高效且用户友好的Python脚本的关键。
