Python

概述

使用 python 的rich库做一些漂亮的数据展示，比如任务列表。

最终效果看着比较直观，可以清晰的看见每个任务的处理状态。

代码示例

1
pip3 install rich
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from rich.console import Console
from rich.table import Table
from rich.live import Live
from rich import box

import time

# 定义状态
STATUS_READY = "[yellow]就绪[/]"
STATUS_RUNNING = "[cyan]进行中[/]"
STATUS_DONE = "[green]完成[/]"
STATUS_FAILED = "[bold red]失败[/]"
STATUS_UNFINISHED = "[grey50]未完成[/]"


class TaskManager:
    def __init__(self, task_names: list[str], fail_fast: bool = True):
        """
        初始化任务管理器
        :param task_names: 任务名称列表
        :param fail_fast: 失败时是否中断所有后续任务
        """
        self.console = Console()
        self.fail_fast = fail_fast
        self.tasks = [{"name": name, "status": STATUS_READY} for name in task_names]
        self._live = None
        self._current_task_idx = 0
        self._failed = False

    def _render_table(self):
        table = Table(title="任务进度", box=box.SIMPLE_HEAVY)
        table.add_column("步骤", justify="center", style="bold")
        table.add_column("任务", justify="left")
        table.add_column("状态", justify="center")

        for idx, task in enumerate(self.tasks, start=1):
            table.add_row(str(idx), task["name"], task["status"])
        return table

    def start(self):
        """开始任务实时渲染"""
        self._live = Live(
            self._render_table(), refresh_per_second=10, console=self.console
        )
        self._live.__enter__()

    def stop(self):
        """停止任务实时渲染"""
        if self._live:
            self._live.__exit__(None, None, None)

    def update(self):
        """刷新显示"""
        if self._live:
            self._live.update(self._render_table())

    def set_running(self, idx: int):
        """设置任务进行中"""
        self.tasks[idx]["status"] = STATUS_RUNNING
        self._current_task_idx = idx
        self.update()

    def set_done(self, idx: int):
        """设置任务完成"""
        self.tasks[idx]["status"] = STATUS_DONE
        self.update()

    def set_failed(self, idx: int):
        """设置任务失败，并根据 fail_fast 控制后续任务"""
        self.tasks[idx]["status"] = STATUS_FAILED
        self._failed = True
        self.update()

        if self.fail_fast:
            for next_idx in range(idx + 1, len(self.tasks)):
                if self.tasks[next_idx]["status"] == STATUS_READY:
                    self.tasks[next_idx]["status"] = STATUS_UNFINISHED
            self.update()

    def set_ready(self, idx: int):
        """手动设置任务为就绪状态"""
        self.tasks[idx]["status"] = STATUS_READY
        self.update()

    def print_final_status(self):
        """打印最终总结信息"""
        self.console.print()
        if self._failed:
            self.console.print("[bold red]部分任务失败！[/] ❌")
        else:
            self.console.print("[bold green]所有任务完成！[/] ✅")


# 简单示例用法
# if __name__ == "__main__":
#     task_list = ["检测系统版本", "寻找源", "开始更换", "更新源", "验证源", "完成"]

#     manager = TaskManager(task_list, fail_fast=False)
#     manager.start()

#     try:
#         for idx, task in enumerate(task_list):
#             manager.set_running(idx)
#             time.sleep(1)  # 模拟任务执行

#             if task == "开始更换":
#                 # 模拟故障
#                 manager.set_failed(idx)
#             else:
#                 manager.set_done(idx)
#                 time.sleep(0.5)
#     finally:
#         manager.stop()
#         manager.print_final_status()

效果

效果大概长这样，能看见定义的每个任务的状态。

其实可以做的更漂亮一点，比如放到表格里面去更新，但是懒得做了。