Python自动化脚本进度条实现：tqdm库高级技巧与可视化

weixin_44118318

18人浏览 · 2026-05-05 12:52:41

weixin_44118318 · 2026-05-05 12:52:41 发布

大家好，我是扣扣。今天来聊聊如何给自动化脚本加上漂亮的进度条。

为什么进度条很重要？

想想这个场景：你写了一个批量处理1000个文件的脚本，运行时什么都看不到，只能干等。你会不会怀疑程序卡死了？

进度条不仅仅是装饰，它能：

给用户反馈：知道程序在运行，没有卡死
预估时间：大概知道还要等多久
调试定位：哪个步骤慢一目了然

一、tqdm基础用法

tqdm是Python最流行的进度条库，"tqdm"在阿拉伯语中是"进度"的意思。

安装

pip install tqdm

基础用法

from tqdm import tqdm
import time

# 最简单的用法
for i in tqdm(range(100)):
    time.sleep(0.1)

# 带描述
for i in tqdm(range(100), desc="处理中"):
    time.sleep(0.1)

# 带单位
for char in tqdm(["a", "b", "c", "d"], desc="处理字符", unit="char"):
    time.sleep(0.1)

二、实用场景实战

场景一：批量文件处理

"""批量文件处理进度条"""

from tqdm import tqdm
from pathlib import Path
import shutil
import time

def batch_process_files(source_dir: str, dest_dir: str, extensions: list = None):
    """批量处理文件，带进度条"""
    
    source = Path(source_dir)
    dest = Path(dest_dir)
    dest.mkdir(parents=True, exist_ok=True)
    
    # 获取文件列表
    if extensions:
        files = [f for f in source.rglob('*') if f.is_file() and f.suffix in extensions]
    else:
        files = [f for f in source.rglob('*') if f.is_file()]
    
    print(f"找到 {len(files)} 个文件待处理")
    
    # 带进度条处理
    success_count = 0
    error_files = []
    
    with tqdm(files, desc="复制文件", unit="file") as pbar:
        for file_path in pbar:
            try:
                # 计算相对路径，保留目录结构
                relative = file_path.relative_to(source)
                target = dest / relative
                target.parent.mkdir(parents=True, exist_ok=True)
                
                # 复制文件
                shutil.copy2(file_path, target)
                success_count += 1
                
            except Exception as e:
                error_files.append((file_path, str(e)))
            
            # 更新描述（动态显示统计信息）
            pbar.set_postfix({
                "成功": success_count,
                "失败": len(error_files)
            })
    
    return {
        "total": len(files),
        "success": success_count,
        "errors": error_files
    }

# 使用示例
if __name__ == '__main__':
    result = batch_process_files('./source', './dest', extensions=['.txt', '.pdf'])
    print(f"处理完成: {result['success']}/{result['total']}")

场景二：网络请求进度

"""带进度条的网络请求"""

import requests
from tqdm import tqdm
import time

def download_with_progress(urls: list, save_dir: str = './downloads') -> dict:
    """下载多个文件，带进度条"""
    
    from pathlib import Path
    save_path = Path(save_dir)
    save_path.mkdir(exist_ok=True)
    
    results = {"success": 0, "failed": 0, "files": []}
    
    # 创建进度条
    pbar = tqdm(urls, desc="下载文件", unit="file")
    
    for url in pbar:
        filename = url.split('/')[-1].split('?')[0]
        filepath = save_path / filename
        
        try:
            # 使用流式请求获取文件大小
            response = requests.get(url, stream=True, timeout=30)
            total_size = int(response.headers.get('Content-Length', 0))
            
            # 下载文件
            downloaded = 0
            with open(filepath, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)
                        downloaded += len(chunk)
                        
                        # 更新进度条
                        pbar.update(0)  # 保持当前位置
                        pbar.set_postfix({"当前": f"{downloaded/1024:.1f}KB / {total_size/1024:.1f}KB"})
            
            results["success"] += 1
            results["files"].append({"url": url, "path": str(filepath), "size": total_size})
            
        except Exception as e:
            results["failed"] += 1
            pbar.set_postfix({"错误": str(e)[:20]})
    
    pbar.close()
    return results

场景三：嵌套进度条

处理多个阶段任务时，需要嵌套进度条：

"""嵌套进度条示例"""

from tqdm import tqdm
import time

def process_data_pipeline():
    """模拟数据处理管道"""
    
    # 外部进度条：处理多个批次
    datasets = [f"dataset_{i}" for i in range(5)]
    
    with tqdm(datasets, desc="处理数据集", position=0) as pbar_datasets:
        for dataset in pbar_datasets:
            # 显示当前处理的 dataset
            pbar_datasets.set_description(f"处理 {dataset}")
            
            # 模拟：读取数据
            time.sleep(0.5)
            
            # 内部进度条：处理数据行
            rows = range(100)
            with tqdm(rows, desc="  处理行", position=1, leave=False) as pbar_rows:
                for row in pbar_rows:
                    time.sleep(0.02)  # 模拟处理
                    pbar_rows.set_postfix({"行号": row})
            
            # 内部进度条：保存数据
            time.sleep(0.3)

if __name__ == '__main__':
    process_data_pipeline()

三、自定义进度条样式

自定义颜色和格式

"""自定义进度条样式"""

from tqdm import tqdm
import time

def custom_progress_demo():
    """自定义进度条演示"""
    
    # 基础自定义
    bar = tqdm(
        range(100),
        desc="基础样式",
        bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]"
    )
    for i in bar:
        time.sleep(0.05)
    
    # 带颜色的自定义
    print("\n带颜色进度条：")
    bar = tqdm(
        range(100),
        desc="处理中",
        bar_format="{desc}: {percentage:3.0f}%|{bar}| {n_fmt}/{total_fmt} [{elapsed}] {postfix}",
        colour="green"
    )
    for i in bar:
        time.sleep(0.05)
        bar.set_postfix({"速度": f"{i*10} item/s", "状态": "正常"})
    
    # 彩色进度条（彩虹色）
    print("\n彩虹进度条：")
    colors = ['red', 'green', 'yellow', 'blue', 'magenta', 'cyan']
    for i, color in enumerate(colors):
        bar = tqdm(
            range(20),
            desc=color,
            bar_format="{l_bar}{bar}|",
            colour=color,
            position=i,
            leave=True
        )
        for _ in bar:
            time.sleep(0.1)

if __name__ == '__main__':
    custom_progress_demo()

ASCII艺术进度条

"""ASCII艺术进度条"""

from tqdm import tqdm
import time

def ascii_progress():
    """ASCII字符进度条"""
    
    # 使用不同字符
    bar_format = '{l_bar}{bar} {n_fmt}/{total_fmt} [{elapsed}]'
    
    # 方块字符
    print("方块字符:")
    for i in tqdm(range(50), bar_format=bar_format, ascii="█"):
        time.sleep(0.05)
    
    # 点字符
    print("\n点字符:")
    for i in tqdm(range(50), bar_format=bar_format, ascii="·"):
        time.sleep(0.05)
    
    # 等号字符
    print("\n等号字符:")
    for i in tqdm(range(50), bar_format=bar_format, ascii="="):
        time.sleep(0.05)

if __name__ == '__main__':
    ascii_progress()

四、trange快捷函数

对于简单的整数循环，trange更简洁：

"""trange快捷函数"""

from tqdm import trange
import time

# trange = tqdm(range(...))
for i in trange(100, desc="使用trange"):
    time.sleep(0.05)

# 嵌套使用
for i in trange(3, desc="外层循环"):
    for j in trange(100, desc=f"内层 {i}", leave=False):
        time.sleep(0.02)

五、tqdm_notebook（Jupyter Notebook）

在Jupyter中使用：

"""Jupyter Notebook 进度条"""

# 方式1：使用 tqdm.notebook
from tqdm.notebook import tqdm, trange

for i in tqdm(range(100), desc="Notebook进度条"):
    pass

# 方式2：自动检测
from tqdm import tqdm
# tqdm 会自动选择合适的实现

六、进度条与日志的配合

长时间运行的任务，进度条和日志如何和平共处？

"""进度条与日志配合"""

from tqdm import tqdm
import logging
import time
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    stream=sys.stdout
)
logger = logging.getLogger(__name__)

def process_with_logging():
    """同时输出日志和进度条"""
    
    items = range(100)
    
    # 方法1：使用 position 和 leave 控制显示位置
    with tqdm(items, desc="处理", position=0, leave=True) as pbar:
        for i in pbar:
            if i % 20 == 0:  # 每20个输出一条日志
                logger.info(f"处理进度: {i}/100")
            
            if i == 50:  # 关键节点日志
                logger.warning("处理到一半了！")
            
            if i == 90:  # 错误恢复
                logger.error("遇到问题，但已恢复")
            
            time.sleep(0.1)
            
            # 动态更新进度条描述
            pbar.set_description(f"处理 {i}")
            pbar.set_postfix({"状态": "运行中"})

if __name__ == '__main__':
    process_with_logging()

七、pandas进度条集成

处理DataFrame时：

"""pandas进度条"""

import pandas as pd
from tqdm import tqdm

# 启用pandas进度条
tqdm.pandas()

# 创建测试数据
df = pd.DataFrame({'value': range(1000)})

# 使用 progress_apply
print("使用 progress_apply:")
result = df['value'].progress_apply(lambda x: x ** 2)

# 分组处理
print("\n分组处理:")
for name, group in df.groupby(df.index // 100):
    processed = group['value'].apply(lambda x: x * 2)
    tqdm.write(f"处理组 {name}: {len(processed)} 条")

八、自定义进度条组件

"""自定义进度条组件"""

from tqdm import tqdm
import time

class MultiProgress:
    """多任务进度管理器"""
    
    def __init__(self, tasks: list):
        self.tasks = tasks
        self.pbars = {}
        self._init_bars()
    
    def _init_bars(self):
        """初始化进度条"""
        for i, task in enumerate(self.tasks):
            self.pbars[task['name']] = tqdm(
                total=task['total'],
                desc=task['name'][:20],
                position=i,
                leave=True
            )
    
    def update(self, task_name: str, n: int = 1, **kwargs):
        """更新进度"""
        pbar = self.pbars.get(task_name)
        if pbar:
            pbar.update(n)
            if kwargs:
                pbar.set_postfix(kwargs)
    
    def close(self):
        """关闭所有进度条"""
        for pbar in self.pbars.values():
            pbar.close()

def multi_task_demo():
    """多任务进度演示"""
    
    tasks = [
        {"name": "下载数据", "total": 100},
        {"name": "处理图片", "total": 50},
        {"name": "写入数据库", "total": 200}
    ]
    
    manager = MultiProgress(tasks)
    
    # 模拟处理
    for i in range(100):
        manager.update("下载数据", 1)
        if i % 2 == 0:
            manager.update("处理图片", 1)
        if i % 0.5 == 0:
            manager.update("写入数据库", 2)
        time.sleep(0.1)
    
    manager.close()

if __name__ == '__main__':
    multi_task_demo()

九、CLI进度条（不使用tqdm）

有时候不想依赖tqdm，可以用标准库实现简单版本：

"""简单进度条实现（不依赖tqdm）"""

import sys
import time

def simple_progress(current: int, total: int, width: int = 50, prefix: str = ""):
    """简单的进度条"""
    percent = current / total
    filled = int(width * percent)
    bar = '█' * filled + '░' * (width - filled)
    
    sys.stdout.write(f'\r{prefix} [{bar}] {percent*100:.1f}% ')
    sys.stdout.flush()

# 使用
for i in range(100):
    simple_progress(i+1, 100, prefix="处理中")
    time.sleep(0.1)
print()  # 换行

总结

基础用法：tqdm(range()) 就够了
文件处理：结合 pathlib/shutil，带统计信息
网络请求：流式下载 + 分块更新
嵌套场景：使用 position 参数
自定义：颜色、字符、格式
日志配合：用 tqdm.write 避免干扰

好了，今天的分享就到这里。给脚本加上进度条，用户体验会好很多。我是扣扣，有问题欢迎留言~🙃

亚马逊云科技技术品牌专区

更多推荐

Kiro Editor 开发实战：使用 Cargo 构建、测试与性能优化指南

欢迎来到这篇终极指南，我们将深入探索如何使用Rust构建高性能的终端文本编辑器Kiro Editor。无论你是Rust新手还是经验丰富的开发者，这篇完整教程将带你了解如何利用Cargo工具链进行高效的开发、测试和性能优化，打造一款快速、轻量且功能强大的UTF-8文本编辑器。## 什么是Kiro Editor？Kiro Editor是一款使用Rust编写的极简终端文本编辑器，它最初是著名编辑