当前位置：首页 > 文章列表 > 文章 > python教程 > tqdm监控批量文件处理进度教程

tqdm监控批量文件处理进度教程

2025-07-11 16:30:33 0浏览收藏

本文详细介绍了如何使用 Python 的 tqdm 库监控批量文件处理的进度，尤其是在文件加密、解密等场景下。针对传统 tqdm 示例在本地文件操作中无法有效追踪进度的问题，提出通过计算文件总大小并为每个文件操作提供更新回调的方法，实现直观的总进度条显示。教程涵盖了 tqdm 库的基础回顾、`iter_files` 函数计算总文件大小、`iter_with_progress` 函数构建带进度的迭代器，以及整合到实际文件处理逻辑的示例。此外，还讨论了内存消耗优化、大文件分块处理、错误处理以及 tqdm 参数的灵活运用，旨在帮助开发者提升文件处理脚本的用户体验和效率，使其更专业、更友好。

如何使用 tqdm 监控文件批量读写与处理进度

本教程详细介绍了如何利用 Python tqdm 库有效监控文件操作进度，特别是在批量处理（如加密/解密）场景下。我们将探讨如何计算总进度并为每个文件操作提供更新回调，从而实现对整个文件处理过程的直观进度条显示，提升用户体验。

引言：理解文件操作进度监控的挑战

在 Python 中进行文件操作时，尤其是涉及大量文件或大文件时，提供一个进度条可以显著提升用户体验。tqdm 是一个功能强大的库，能够轻松为循环或迭代器添加进度条。然而，许多 tqdm 的示例都集中在网络下载场景，例如使用 requests 库的 iter_content 方法，它能以分块的方式迭代数据流，从而方便地更新进度条。

对于本地文件操作，特别是像 file.write(data) 这样一次性写入大量数据的场景，直接跟踪 write() 内部的进度会变得复杂，因为 write() 通常是一个原子操作，它会尝试一次性将所有数据写入。这意味着我们无法在单次 write() 调用过程中获得细粒度的进度更新。

本教程的目标是解决一个更常见但同样重要的场景：如何在使用 file.read() 读取整个文件并进行处理（如编码/加密），然后使用 file.write() 写回时，监控 整个文件处理过程 或 批量文件处理过程 的总进度。我们将通过计算所有文件的总大小来初始化进度条，并在每个文件处理完成后更新进度。

tqdm 库基础回顾

tqdm 库的核心思想是为可迭代对象提供一个进度条。它的基本用法非常简单：

from tqdm import tqdm
import time

for 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 os
from tqdm import tqdm

def 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 os
from base64 import b85encode, b85decode # 导入base85编码/解码函数
import time # 模拟耗时操作

# 假设的输入目录
input_dir = 'C:\\Python311\\test_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"\r正在处理: {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"\r完成处理: {file_name}                               ") # 清除行尾的进度信息

        # 调用 done_callback 通知进度条更新当前文件大小
        done_callback() 

    except PermissionError:
        print(f"\r跳过 (权限不足): {file_name}                      ")
        # 对于跳过的文件，如果不想计入总进度，则不调用done_callback
        # 如果跳过也算作处理完成，则可以调用 done_callback()
        done_callback() # 也可以选择不调用，这样进度条会显示未完成
    except Exception as e:
        print(f"\r处理 {file_name} 时发生错误: {e}                   ")
        done_callback() # 即使出错也更新进度，表示该文件已尝试处理

print("所有文件处理完毕。")

在这个整合示例中，iter_with_progress(input_dir) 提供了遍历所有文件的能力，并且在每次循环迭代时，我们获得了 done_callback。在每个文件的加密（或任何处理）完成后，我们调用 done_callback()，这将通知 tqdm 进度条增加相应的文件大小，从而实现对整个批处理过程的进度监控。

注意事项与优化

内存消耗与大文件处理： 您原始代码中的 b85encode(encryptingfile.read()) 会将整个文件内容读入内存。对于非常大的文件（例如几GB），这可能会导致内存溢出。 优化建议： 对于大文件，应采用分块读写的方式。这意味着您需要以固定大小的块读取文件，对每个块进行处理，然后将处理后的块写入新文件。在这种情况下，tqdm.update() 的参数应该是每次写入的块大小，而不是整个文件的大小。例如：

# 伪代码：分块读写
chunk_size = 4096
with 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监控批量文件处理进度教程》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！