如何为持续运行的Python脚本实现CLI底部常驻仪表盘？

这个需求我太熟了——之前写后台监控脚本的时候专门折腾过类似的控制台固定仪表盘，完全明白你想要的效果：普通日志正常往上滚动，底部永远固定一块区域显示关键统计，不会被新输出的日志冲乱。下面给你两种靠谱的实现思路，都是纯命令行方案，完全符合你的要求。

核心思路：ANSI控制字符+线程定时重绘 #

其实wget这类工具的进度条本质就是用ANSI终端控制字符来操纵光标位置，覆盖旧内容实现的。关键是要解决两个问题：一是让仪表盘始终固定在底部，二是和Python的logger输出兼容，不互相干扰。

完整代码示例 #

先上可直接运行的代码，后面再拆解细节：

import logging
import os
import time
import threading
import signal
import ctypes

# 处理Windows终端的ANSI支持
if os.name == 'nt':
    try:
        kernel32 = ctypes.WinDLL('kernel32')
        kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7)
    except:
        pass  # 旧版本Windows可能不支持，忽略即可

class ConsoleDashboard:
    def __init__(self, stats_callback):
        self.stats_callback = stats_callback  # 获取最新统计数据的回调
        self.running = False
        self.thread = None
        self._lock = threading.Lock()
        # 监听终端窗口大小变化
        signal.signal(signal.SIGWINCH, self._handle_resize)
        self.terminal_height = os.get_terminal_size().lines

    def _handle_resize(self, signum, frame):
        self.terminal_height = os.get_terminal_size().lines

    def _redraw(self):
        with self._lock:
            # 获取最新统计数据
            stats = self.stats_callback()
            # 定义仪表盘内容（可以自定义行数和格式）
            dash_lines = [
                "+" * 80,
                f"+ Statistics: Critical errors: {stats['errors']} | Processing: {stats['speed']} units/sec +",
                f"+ Runtime: {stats['runtime']}s | Foos barred: {stats['foos']} | Bars fooed: {stats['bars']} +",
                "+" * 80
            ]
            num_dash_lines = len(dash_lines)
            # 计算仪表盘起始行：终端高度 - 仪表盘行数
            start_line = max(1, self.terminal_height - num_dash_lines)
            
            # ANSI控制码：移动光标到仪表盘起始行开头
            print(f"\033[{start_line};0H", end="")
            # 清除从当前行到屏幕底部的内容（避免残留旧数据）
            print("\033[J", end="")
            # 输出仪表盘内容
            for line in dash_lines:
                print(line)
            # 把光标移回仪表盘上方的日志输出位置
            print(f"\033[{start_line - 1};0H", end="")
            # 强制刷新输出缓冲区
            os.sys.stdout.flush()

    def start(self, interval=1):
        self.running = True
        def update_loop():
            while self.running:
                self._redraw()
                time.sleep(interval)
        # 守护线程，主程序退出时自动停止
        self.thread = threading.Thread(target=update_loop, daemon=True)
        self.thread.start()

    def stop(self):
        self.running = False
        if self.thread:
            self.thread.join()
        # 清除仪表盘，恢复正常终端输出
        print("\033[J", end="")
        os.sys.stdout.flush()

# 自定义Logger Handler，确保日志输出后立即重绘仪表盘
class DashboardStreamHandler(logging.StreamHandler):
    def __init__(self, dashboard):
        super().__init__()
        self.dashboard = dashboard

    def emit(self, record):
        try:
            # 保存当前光标位置
            print("\033[s", end="")
            # 输出日志内容
            super().emit(record)
            # 恢复光标位置并重绘仪表盘
            print("\033[u", end="")
            self.dashboard._redraw()
        except Exception:
            self.handleError(record)

# ---------------------- 主程序示例 ----------------------
if __name__ == "__main__":
    # 统计数据存储（实际项目中替换成你的业务数据）
    start_time = time.time()
    error_count = 0
    current_speed = 12
    foos_barred = 3
    bars_fooed = 11

    def get_current_stats():
        """返回最新统计数据的回调函数"""
        runtime = int(time.time() - start_time)
        return {
            "errors": error_count,
            "speed": current_speed,
            "runtime": runtime,
            "foos": foos_barred,
            "bars": bars_fooed
        }

    # 配置Logger
    formatter = logging.Formatter("%(asctime)s - Logger info - %(message)s")
    dashboard = ConsoleDashboard(get_current_stats)
    handler = DashboardStreamHandler(dashboard)
    handler.setFormatter(formatter)
    
    logging.basicConfig(level=logging.INFO, handlers=[handler])

    # 启动仪表盘（每秒更新一次）
    dashboard.start(interval=1)

    try:
        # 模拟脚本运行，输出日志并更新统计数据
        while True:
            logging.info("Minor status update 1")
            time.sleep(2)
            logging.info("Status has changed")
            current_speed += 1
            foos_barred += 1
            time.sleep(3)
            logging.info("Successfully opened file")
            bars_fooed += 2
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nStopping script...")
    finally:
        dashboard.stop()

关键细节说明 #

1. ANSI控制字符：

   • \033[<行>;<列>H：移动光标到指定位置
   • \033[J：清除从当前行到屏幕底部的内容
   • \033[s/\033[u：保存/恢复光标位置
     这些控制码几乎所有现代终端（Linux/macOS终端、Windows Terminal/PowerShell）都支持，旧版Windows CMD需要开启虚拟终端模式（代码里已经处理了）。

2. Logger兼容：
   自定义了DashboardStreamHandler，每次输出日志前先保存光标位置，输出后恢复位置并重绘仪表盘，彻底避免日志冲掉仪表盘的问题。

3. 终端自适应：
   监听SIGWINCH信号（终端窗口大小变化），动态调整仪表盘的位置，保证始终在底部。

4. 线程安全：
   用threading.Lock确保多线程下的输出不会混乱，仪表盘更新线程是守护线程，主程序退出时自动停止。

备选方案：用tqdm实现单行仪表盘 #

如果你只需要单行的统计信息，tqdm其实也能满足需求，而且代码更简洁：

from tqdm import tqdm
import logging
import time

# 让Logger输出不干扰tqdm进度条
class TqdmLoggingHandler(logging.StreamHandler):
    def emit(self, record):
        try:
            msg = self.format(record)
            tqdm.write(msg)
            self.flush()
        except Exception:
            self.handleError(record)

# 配置Logger
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - Logger info - %(message)s",
    handlers=[TqdmLoggingHandler()]
)

# 初始化统计数据
start_time = time.time()
error_count = 0
current_speed = 12
foos_barred = 3
bars_fooed = 11

# 用tqdm作为单行仪表盘，total设为None表示无限循环
with tqdm(total=None, bar_format="{desc}") as pbar:
    while True:
        logging.info("Minor status update")
        time.sleep(2)
        logging.info("Status has changed")
        
        # 更新统计数据
        runtime = int(time.time() - start_time)
        current_speed += 1
        foos_barred += 1
        
        # 更新仪表盘描述
        pbar.set_description(
            f"Statistics: Errors={error_count} | Speed={current_speed} units/sec | Runtime={runtime}s | Foos={foos_barred} | Bars={bars_fooed}"
        )
        pbar.update(0)  # 不更新进度条，只更新描述
        time.sleep(3)

这个方案适合只需要单行统计的场景，优点是代码简洁，缺点是无法支持多行仪表盘。

总结 #

• 如果需要多行、自定义程度高的仪表盘，优先选择ANSI控制字符+线程重绘的方案，完全兼容logger，效果和wget进度条一致。
• 如果只需要单行统计，tqdm是个快速实现的选择。

内容的提问来源于stack exchange，提问作者right2clicky