使用tqdm跟踪文件处理进度详解

从现在开始，努力学习吧！本文《用tqdm跟踪文件处理进度方法》主要讲解了等等相关知识点，我会在golang学习网中持续更新相关的系列文章，欢迎大家关注并积极留言建议。下面就先一起来看一下本篇正文内容吧，希望能帮到你！

本文详细介绍了如何利用Python的tqdm库有效地跟踪文件处理（如加密、解密或批量写入）的进度。文章通过自定义迭代器函数，实现了在文件级别而非字节级别对操作总进度进行可视化，解决了传统tqdm示例主要针对下载流式数据的局限性，并提供了清晰的代码示例和集成指导，帮助开发者为文件操作添加直观的进度条。

1. 引言：理解文件处理中的进度跟踪需求

在处理大量文件，例如进行批量加密、解密、格式转换或数据迁移时，我们通常希望能够实时了解操作的进度。Python的tqdm库是一个功能强大的进度条工具，广泛用于迭代过程的可视化。然而，其官方示例多集中于网络下载等流式数据的处理，即按字节流更新进度。对于像file.read()一次性读取整个文件内容，然后通过file.write()一次性写入处理后的数据这种场景，传统的tqdm用法（如iter_content）似乎难以直接套用。

本教程旨在解决这一挑战，展示如何通过巧妙的设计，使tqdm能够跟踪文件级别的处理进度，即每完成一个文件的处理，进度条就相应更新。

2. 核心思想：基于文件总量的进度跟踪

为了在文件处理过程中实现进度跟踪，我们需要解决两个关键问题：

1. 确定总进度： 这通常是所有待处理文件的总大小（字节数）。
2. 更新进度： 在每个文件处理完成后，将该文件的大小累加到已完成的进度中。

与下载文件时按接收到的字节数实时更新不同，文件处理往往是“一次性”完成一个文件的读取、处理和写入。因此，最直观且易于实现的方式是：将每个文件的处理视为一个独立的“任务单元”，并在该任务单元完成后，更新进度条。

3. 实现细节：自定义迭代器与进度回调

为了实现上述核心思想，我们可以设计两个辅助函数：

3.1 iter_files(folder)：遍历并统计文件信息

这个函数负责遍历指定文件夹及其子文件夹中的所有文件，并为每个文件生成其大小和完整路径。这是计算总进度的基础。

import os

def iter_files(folder):
    """
    遍历指定文件夹及其子文件夹中的所有文件，
    并生成每个文件的大小和完整路径。
    """
    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}")
                continue # 跳过无法访问的文件

解释：

• os.walk(folder)：递归遍历目录树。
• os.path.join(root, file)：构建文件的完整路径。
• os.path.getsize(file_path)：获取文件的大小（字节）。
• yield file_size, file_path：将文件大小和路径作为元组生成，使其成为一个生成器，节省内存。

3.2 iter_with_progress(folder)：集成tqdm进度条

这个函数将tqdm进度条与文件遍历结合起来。它首先利用iter_files计算出所有文件的总大小，然后初始化tqdm进度条。接着，它为每个文件生成一个特殊的“完成回调函数”（done），以及文件的大小和路径。当外部调用者完成对某个文件的处理后，只需调用这个done函数，进度条就会相应更新。

from tqdm import tqdm

def iter_with_progress(folder):
    """
    为指定文件夹中的文件处理提供tqdm进度条。
    初始化进度条，并为每个文件生成一个更新进度的回调函数。
    """
    # 1. 计算所有文件的总大小作为tqdm的总量
    total_size = sum(s for s, _ in iter_files(folder))

    # 2. 初始化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="处理文件")

    # 3. 再次遍历文件，并为每个文件提供更新回调
    for size, file_path in iter_files(folder):
        # 定义一个lambda函数作为回调，当调用时更新进度条
        done = lambda current_file_size=size: progress.update(current_file_size)
        yield done, size, file_path

    # 确保循环结束后关闭进度条
    progress.close()

解释：

• total_size = sum(s for s, _ in iter_files(folder))：首先调用iter_files来获取所有文件的大小，并求和，得到总进度。
• tqdm(...)：初始化进度条，设置单位为字节（B），总大小为total_size，并启用单位自动缩放。
• done = lambda current_file_size=size: progress.update(current_file_size)：这是一个关键点。它创建了一个闭包，current_file_size捕获了当前文件的大小。当外部代码完成对file_path的处理后，调用done()即可将size累加到进度条上。
• yield done, size, file_path：生成器每次产出三个值：一个用于更新进度的回调函数、当前文件的大小和路径。
• progress.close()：在所有文件处理完毕后，确保关闭进度条，释放资源。

4. 整合示例：应用于文件加密/解密场景

现在，我们将上述函数整合到实际的文件处理流程中，例如用户最初提到的文件加密/解密场景。

import os
import time
from base64 import b85encode, b85decode # 导入base85编码/解码函数

# 假设的输入目录和输出目录
INPUT_DIR = 'test_input_files'
OUTPUT_DIR = 'test_output_files'

# 创建一些测试文件
def create_dummy_files(directory, num_files=5, avg_size_kb=100):
    if not os.path.exists(directory):
        os.makedirs(directory)
    print(f"创建测试文件于: {directory}")
    for i in range(num_files):
        filename = os.path.join(directory, f"file_{i+1}.txt")
        # 生成随机内容
        content = os.urandom(avg_size_kb * 1024 // 2 + i * 1024) # 稍微变化大小
        with open(filename, 'wb') as f:
            f.write(content)
    print("测试文件创建完成。")

# 确保辅助函数已定义
# (这里省略重复定义 iter_files 和 iter_with_progress，假设它们已在上方定义)
# ... iter_files 和 iter_with_progress 函数定义 ...

def process_files_with_progress(input_folder, output_folder, operation_type='encrypt'):
    """
    带进度条的文件处理函数（加密或解密）。
    """
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)

    print(f"\n开始{operation_type}文件...")
    for done_callback, file_size, input_file_path in iter_with_progress(input_folder):
        relative_path = os.path.relpath(input_file_path, input_folder)
        output_file_path = os.path.join(output_folder, relative_path)

        # 确保输出目录存在
        output_file_dir = os.path.dirname(output_file_path)
        if not os.path.exists(output_file_dir):
            os.makedirs(output_file_dir)

        try:
            print(f"\n正在处理: {relative_path} ({file_size / 1024:.2f} KB)")
            with open(input_file_path, 'rb') as infile:
                original_bytes = infile.read()

            processed_bytes = b''
            if operation_type == 'encrypt':
                processed_bytes = b85encode(original_bytes)
            elif operation_type == 'decrypt':
                processed_bytes = b85decode(original_bytes)
            else:
                raise ValueError("operation_type 必须是 'encrypt' 或 'decrypt'")

            with open(output_file_path, 'wb') as outfile:
                outfile.write(processed_bytes)

            # 模拟处理时间
            # time.sleep(file_size / (1024 * 1024 * 5)) # 模拟每MB处理0.2秒

            done_callback() # 关键：文件处理完成后调用回调函数更新进度条
            print(f"完成处理: {relative_path}")

        except PermissionError:
            print(f"\r跳过 (权限错误): {relative_path}\n")
        except Exception as e:
            print(f"\r处理失败 ({relative_path}): {e}\n")
    print(f"\n所有文件{operation_type}处理完成。")

# --- 运行示例 ---
if __name__ == "__main__":
    # 清理旧目录
    import shutil
    if os.path.exists(INPUT_DIR):
        shutil.rmtree(INPUT_DIR)
    if os.path.exists(OUTPUT_DIR):
        shutil.rmtree(OUTPUT_DIR)

    # 1. 创建测试文件
    create_dummy_files(INPUT_DIR, num_files=10, avg_size_kb=500)

    # 2. 执行加密操作并显示进度
    process_files_with_progress(INPUT_DIR, OUTPUT_DIR + '_encrypted', 'encrypt')

    # 3. 模拟解密操作（可选，需要先有加密文件）
    # 为了演示，我们将加密后的文件作为解密输入
    # process_files_with_progress(OUTPUT_DIR + '_encrypted', OUTPUT_DIR + '_decrypted', 'decrypt')

代码解释：

• create_dummy_files：一个辅助函数，用于创建一些随机内容的测试文件，方便演示。
• process_files_with_progress：
  • 接收input_folder、output_folder和operation_type（'encrypt'或'decrypt'）。
  • 通过iter_with_progress(input_folder)获取一个迭代器，每次迭代得到done_callback、file_size和input_file_path。
  • 在try-except块中执行文件读取、b85encode或b85decode处理，以及文件写入操作。
  • 最重要的是： 在outfile.write(processed_bytes)完成之后，调用done_callback()。这将通知tqdm进度条，当前文件已经处理完毕，并将该文件的大小累加到已完成的总量中。
  • 包含了基本的错误处理，如PermissionError。

5. 注意事项与总结

1. 进度更新粒度： 本方案的进度更新粒度是“文件级别”。这意味着，对于单个大文件的内部处理（例如，如果加密/解密操作是分块进行的），进度条不会在文件内部实时更新。它只在整个文件处理完毕后才向前推进。如果需要文件内部的字节级进度，则需要更复杂的实现，例如为write操作创建一个自定义的文件类，或者将大文件分块读取和写入，并在每次小块写入后更新tqdm。然而，对于大多数批量文件处理场景，文件级别的进度已经足够提供清晰的用户反馈。
2. 错误处理： 在实际应用中，文件操作可能遇到各种错误（如权限不足、磁盘空间不足、文件损坏等）。务必在文件处理逻辑中加入健壮的try-except块，以防止程序崩溃，并确保即使发生错误，进度条也能继续运行（如果跳过该文件）或给出适当的提示。
3. 内存考虑： b85encode(encryptingfile.read())这种模式会将整个文件内容读入内存。对于非常大的文件（GB级别），这可能会导致内存溢出。在处理大文件时，建议采用分块读取和写入的策略，并相应调整进度更新逻辑。
4. tqdm参数： tqdm提供了丰富的参数来自定义进度条的显示，例如desc（描述）、ncols（宽度）、bar_format（格式）等。根据需要调整这些参数以优化用户体验。

通过上述方法，我们可以有效地利用tqdm为各种文件处理任务提供直观且专业的进度条，极大地提升了脚本的用户友好性。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《使用tqdm跟踪文件处理进度详解》文章吧，也可关注golang学习网公众号了解相关技术文章。