Python tqdm 实践：构建文件处理与写入操作的进度条

import requests
from tqdm import tqdm

url = 'https://wordnetcode.princeton.edu/2.1/WNsnsmap-2.1.tar.gz'
filename = url.split('/')[-1]
resp = requests.get(url, stream=True)
# 初始化进度条，total 为文件总大小，unit='B' 表示单位是字节
pbar = tqdm(desc=filename, total=int(resp.headers.get('content-length', 0)),
            unit='B', unit_scale=True, unit_divisor=1024)

with open(filename, 'wb') as f:
    for data in resp.iter_content(chunk_size=1024): # 按块迭代内容
        f.write(data)
        pbar.update(len(data)) # 每次写入一块数据后更新进度条

pbar.close() # 关闭进度条

import os
from base64 import b85encode # 假设 b85encode 是你的处理逻辑

input_dir = "path/to/your/files" # 替换为你的输入目录

for filefolder, dirs, files in os.walk(input_dir):
    for filename in files:
        file_path = os.path.join(filefolder, filename)
        print(f"正在加密 {filename}..")
        try:
            with open(file_path, 'rb') as encryptingfile:
                # 一次性读取整个文件内容
                original_bytes = encryptingfile.read()
                # 进行加密处理，例如 Base85 编码
                b85_bytes = b85encode(original_bytes)

            # 一次性写入处理后的所有字节
            with open(file_path, 'wb') as output_file:
                output_file.write(b85_bytes)

            # 问题：在哪里调用 pbar.update() 来显示 output_file.write(b85_bytes) 的进度？
            # 因为 write() 是一个原子操作，它不提供内部的字节写入进度。

        except PermissionError:
            print(f"\r跳过: {filename} (权限不足)\n")
        except Exception as e:
            print(f"\r处理 {filename} 时发生错误: {e}\n")

import os
from tqdm import tqdm
import time # 用于模拟文件处理耗时
from base64 import b85encode, b85decode # 模拟加密/解密

def iter_files(folder_path):
    """
    遍历指定文件夹及其子文件夹中的所有文件，并返回文件大小和完整路径。
    """
    for root, _, files in os.walk(folder_path):
        for file_name in files:
            file_full_path = os.path.join(root, file_name)
            try:
                # 确保文件存在且可访问，获取其大小
                if os.path.exists(file_full_path) and os.path.isfile(file_full_path):
                    yield os.path.getsize(file_full_path), file_full_path
            except Exception as e:
                print(f"无法获取文件大小或访问文件: {file_full_path}, 错误: {e}")
                continue

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

    # 初始化 tqdm 进度条
    progress_bar = tqdm(unit='B',
                        total=total_size,
                        unit_scale=True,
                        unit_divisor=1024,
                        desc=description)

    # 遍历文件，并为每个文件提供一个更新进度的回调
    for file_size, file_path in iter_files(folder_path):
        # 使用 lambda 表达式创建一个闭包，当文件处理完成后调用它来更新进度条
        update_callback = lambda size=file_size: progress_bar.update(size)
        yield update_callback, file_size, file_path

    # 在所有文件迭代完毕后，确保进度条关闭
    # 注意：如果循环提前终止，可能需要手动调用 progress_bar.close()
    progress_bar.close()

# 示例：模拟文件加密并更新进度条
# 请将 'path/to/your/input/directory' 替换为你要处理的文件夹路径
input_directory = 'path/to/your/input/directory'

# 确保输入目录存在
if not os.path.isdir(input_directory):
    print(f"错误：目录 '{input_directory}' 不存在。请替换为有效路径。")
else:
    print(f"开始处理目录: {input_directory}")
    for update_progress_callback, file_size, file_path in iter_with_progress(input_directory, description="文件处理"):
        try:
            print(f"\n正在处理文件: {os.path.basename(file_path)} ({file_size} Bytes)")

            # 模拟文件读取和加密操作
            with open(file_path, 'rb') as f_read:
                original_bytes = f_read.read()
                # 假设 b85encode 是你的加密逻辑
                processed_bytes = b85encode(original_bytes) # 或 b85decode(original_bytes) 进行解密

            # 模拟文件写入操作
            # 注意：这里直接覆盖原文件，实际应用中可能写入新文件或进行备份
            with open(file_path, 'wb') as f_write:
                f_write.write(processed_bytes)

            # 文件处理完成后，调用回调函数更新进度条
            update_progress_callback()

        except PermissionError:
            print(f"跳过文件 (权限不足): {os.path.basename(file_path)}")
        except Exception as e:
            print(f"处理文件 {os.path.basename(file_path)} 时发生错误: {e}")

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