Python脚本如何实现运行进度展示

wen python案例 28

本文目录导读:

Python脚本如何实现运行进度展示

  1. 使用 tqdm 库(推荐)
  2. 自定义简单进度条
  3. 基于时间的进度估算
  4. 多线程任务进度(同时处理多个任务)
  5. 带日志的进度条
  6. Web应用中的进度展示
  7. 最佳实践建议
  8. 简单示例:文件下载进度

我来介绍几种Python实现运行进度展示的方法:

使用 tqdm 库(推荐)

最简单强大的方式:

# 安装:pip install tqdm
from tqdm import tqdm
import time
# 方法1:直接包装可迭代对象
for i in tqdm(range(100)):
    time.sleep(0.1)  # 模拟耗时操作
# 方法2:手动更新进度
with tqdm(total=100) as pbar:
    for i in range(100):
        time.sleep(0.1)
        pbar.update(1)  # 更新进度
# 方法3:处理文件
from tqdm import tqdm
import os
file_size = os.path.getsize("large_file.txt")
with open("large_file.txt", "rb") as f:
    with tqdm(total=file_size, unit="B", unit_scale=True) as pbar:
        while True:
            chunk = f.read(1024)
            if not chunk:
                break
            pbar.update(len(chunk))

自定义简单进度条

def progress_bar(current, total, bar_length=50):
    """自定义进度条"""
    percent = current / total
    arrow = '█' * int(round(percent * bar_length))
    spaces = '░' * (bar_length - len(arrow))
    print(f'\r进度: |{arrow}{spaces}| {percent:.1%} ({current}/{total})', end='', flush=True)
# 使用示例
import time
total = 100
for i in range(total + 1):
    progress_bar(i, total)
    time.sleep(0.1)
print()  # 换行

基于时间的进度估算

import time
from datetime import datetime
class TimeProgressBar:
    def __init__(self, total, desc="处理中"):
        self.total = total
        self.desc = desc
        self.start_time = time.time()
    def update(self, current):
        """更新进度"""
        percent = current / self.total
        elapsed = time.time() - self.start_time
        # 估算剩余时间
        if percent > 0:
            eta = elapsed / percent - elapsed
            eta_str = str(datetime.fromtimestamp(int(eta)).strftime('%H:%M:%S'))
        else:
            eta_str = "计算中..."
        # 构建进度条
        bar_length = 30
        filled = int(bar_length * percent)
        bar = '█' * filled + '░' * (bar_length - filled)
        # 格式化输出
        speed = current / elapsed if elapsed > 0 else 0
        print(f'\r{self.desc}: |{bar}| {percent:.1%} '
              f'[{current}/{self.total}] '
              f'速度: {speed:.2f}/秒 '
              f'ETA: {eta_str}', end='', flush=True)
# 使用示例
total = 100
pbar = TimeProgressBar(total, "下载文件")
for i in range(total + 1):
    pbar.update(i)
    time.sleep(0.1)
print()

多线程任务进度(同时处理多个任务)

from concurrent.futures import ThreadPoolExecutor
from tqdm import tqdm
import time
def process_item(item):
    """处理单个任务"""
    time.sleep(0.5)  # 模拟耗时操作
    return item * 2
# 方法1:简单列表处理
items = range(20)
results = []
with tqdm(total=len(items), desc="处理任务") as pbar:
    for item in items:
        result = process_item(item)
        results.append(result)
        pbar.update(1)
# 方法2:使用线程池 + tqdm
def process_with_progress():
    with ThreadPoolExecutor(max_workers=5) as executor:
        items = list(range(20))
        # 提交所有任务
        futures = [executor.submit(process_item, item) for item in items]
        # 使用tqdm跟踪进度
        with tqdm(total=len(items), desc="并行处理") as pbar:
            for future in futures:
                future.add_done_callback(lambda x: pbar.update(1))
            # 等待所有任务完成
            results = [future.result() for future in futures]
    return results
results = process_with_progress()

带日志的进度条

import logging
from tqdm import tqdm
import time
class LoggingTqdm(tqdm):
    """支持日志输出的tqdm"""
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.logger = logging.getLogger(__name__)
    def display(self, msg=None, pos=None):
        # 重写display方法
        super().display(msg, pos)
        if self.logger:
            self.logger.debug(str(self))
# 配置日志
logging.basicConfig(level=logging.DEBUG)
# 使用示例
for i in LoggingTqdm(range(100), desc="处理"):
    time.sleep(0.1)

Web应用中的进度展示

# 使用 Streamlit(需要安装:pip install streamlit)
import streamlit as st
import time
# 创建进度条
progress_bar = st.progress(0)
status_text = st.empty()
for i in range(100):
    # 更新进度
    progress_bar.progress(i + 1)
    status_text.text(f"进度: {i + 1}%")
    time.sleep(0.1)
st.success("任务完成!")

最佳实践建议

  1. 使用 tqdm:对于大多数场景,tqdm 是最佳选择,功能丰富且代码简洁
  2. 设置合适的描述信息:让用户知道当前在进行什么操作
  3. 显示速度信息:让用户了解处理速度
  4. 估算剩余时间:提供 ETA 信息,让用户知道还需等待多久
  5. 适当更新频率:避免过于频繁的更新导致性能下降(特别是处理大量数据时)

选择哪种方式取决于你的具体需求:

  • 命令行工具:使用 tqdm
  • Web应用:使用前端框架或 WebSocket 实现
  • 数据处理:使用 tqdm + pandas 集成
  • 移动端/桌面应用:使用 GUI 框架的进度条组件

简单示例:文件下载进度

import requests
from tqdm import tqdm
def download_file(url, filename):
    response = requests.get(url, stream=True)
    total_size = int(response.headers.get('content-length', 0))
    with open(filename, 'wb') as file:
        with tqdm(total=total_size, unit='B', unit_scale=True, 
                  desc=filename) as pbar:
            for chunk in response.iter_content(chunk_size=1024):
                file.write(chunk)
                pbar.update(len(chunk))
# 使用示例
url = "https://example.com/large_file.zip"
download_file(url, "large_file.zip")

选择最适合你场景的实现方式即可!

抱歉,评论功能暂时关闭!