使用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}")
                continue

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()

# 假设我们有一个名为 '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')