You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python中创建带实时时钟、支持双触发的可调用进度条?

实现可主动更新+自动定时刷新的文件处理进度条

你当前的代码存在几个关键问题:

  1. threading.Timer 使用错误:progress_bar(current,total,filename) 是直接执行函数并把返回值传给Timer,而非传递函数对象,导致Timer无法正确触发
  2. 每次调用都会启动新Timer,最终多个线程同时打印,输出混乱
  3. 依赖全局变量 start,代码耦合性高
  4. 无法同时满足「进度推进时主动更新」和「每秒自动刷新」的需求

下面提供两种实用的实现方案:

方案1:用类封装状态+后台刷新线程

这种方式把进度数据封装在类中,启动一个后台线程每秒自动刷新,主线程处理文件时只需要调用update方法更新状态即可,同时update会立即刷新一次进度条。

import time
import threading

class ProgressMonitor:
    def __init__(self, total, bar_length=20):
        self.total = total
        self.current = 0
        self.filename = ""
        self.bar_length = bar_length
        self.start_time = time.monotonic()
        self._stop_flag = threading.Event()
        # 启动后台刷新线程
        self._refresh_thread = threading.Thread(target=self._refresh_loop, daemon=True)
        self._refresh_thread.start()

    def _refresh_loop(self):
        while not self._stop_flag.is_set():
            self._draw_bar()
            time.sleep(1)
        # 任务完成后打印最终状态
        self._draw_bar()

    def _draw_bar(self):
        percent = (self.current / self.total) * 100 if self.total > 0 else 0
        arrow = '-' * int(percent / 100 * self.bar_length - 1) + '>' if self.current > 0 else ''
        spaces = ' ' * (self.bar_length - len(arrow))
        elapsed = time.strftime("%H:%M:%S", time.gmtime(time.monotonic() - self.start_time))

        if self.current >= self.total:
            print(f"\rProgress: [{'-'*(self.bar_length-1)}>] 100% (COMPLETED)\033[K | time passed: {elapsed}")
            self._stop_flag.set()
        else:
            print(f"\rProgress: [{arrow}{spaces}] {percent:.2f}% | file: {self.filename} | time passed: {elapsed}",
                  end='', flush=True)

    def update(self, current, filename=""):
        self.current = current
        self.filename = filename
        # 主动更新时立即刷新进度条
        self._draw_bar()

# 使用示例
if __name__ == "__main__":
    total_tasks = 5
    monitor = ProgressMonitor(total_tasks)

    for idx in range(total_tasks):
        current_file = f"large_file_{idx+1}.dat"
        monitor.update(idx+1, current_file)
        # 模拟文件处理耗时操作
        time.sleep(2)

方案2:用共享状态字典+Timer定时刷新

如果不想用类,可以用一个全局字典保存进度状态,通过Timer实现定时刷新,同时提供更新函数主动触发刷新。

import time
import threading

# 共享进度状态
progress_data = {
    'current': 0,
    'total': 0,
    'filename': '',
    'start_time': None,
    'bar_length': 20,
    'active_timer': None
}

def _draw_progress():
    data = progress_data
    if not data['start_time']:
        return

    percent = (data['current'] / data['total']) * 100 if data['total'] > 0 else 0
    arrow = '-' * int(percent / 100 * data['bar_length'] - 1) + '>' if data['current'] > 0 else ''
    spaces = ' ' * (data['bar_length'] - len(arrow))
    elapsed = time.strftime("%H:%M:%S", time.gmtime(time.monotonic() - data['start_time']))

    if data['current'] >= data['total']:
        print(f"\rProgress: [{'-'*(data['bar_length']-1)}>] 100% (COMPLETED)\033[K | time passed: {elapsed}")
        if data['active_timer']:
            data['active_timer'].cancel()
    else:
        print(f"\rProgress: [{arrow}{spaces}] {percent:.2f}% | file: {data['filename']} | time passed: {elapsed}",
              end='', flush=True)
        # 启动下一次定时刷新
        data['active_timer'] = threading.Timer(1, _draw_progress)
        data['active_timer'].start()

def init_progress(total, bar_length=20):
    progress_data['total'] = total
    progress_data['current'] = 0
    progress_data['filename'] = ''
    progress_data['start_time'] = time.monotonic()
    progress_data['bar_length'] = bar_length
    # 首次触发刷新
    _draw_progress()

def update_progress(current, filename=""):
    progress_data['current'] = current
    progress_data['filename'] = filename
    # 主动更新时立即刷新
    _draw_progress()

# 使用示例
if __name__ == "__main__":
    total_files = 5
    init_progress(total_files)

    for idx in range(total_files):
        current_file = f"large_file_{idx+1}.dat"
        update_progress(idx+1, current_file)
        time.sleep(2)

方案说明

  • 类封装方案:状态管理更清晰,适合复杂的多任务场景,后台线程自动处理定时刷新,主线程只需要关注进度更新
  • 共享字典方案:更轻量,适合简单场景,不需要类的结构,通过全局状态实现数据共享

内容的提问来源于stack exchange,提问作者condosz

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.08 17:43:21