Python代码片段-断点任务

使用Python处理一堆长耗时任务的时候，为了防止异常退出程序或者手动退出程序后丢失任务进度，可用使用断点的方式记录任务进度，下次重载任务后，继续运行上次未完成的任务即可。

这里用json文件作为数据持久化的方式，免去了使用数据库来作为持久的依赖问题，为了一个小的任务来搭建一套数据库，耗时耗力，得不偿失，不如选择本地文件来做简单的持久化，就算丢失部分任务进度，少数任务重新执行所消耗的时间完全处于容忍范围内。另外使用json文件记录任务进度，可用修改json文件来人为干预任务的执行情况，也是不错的一个优点

import json
import time
import os


class Task:
    def __init__(self, task_file='task_progress.json'):
        self.task_file = task_file
        self.progress = []
        if os.path.exists(self.task_file):
            self.load_progress()

    def load_progress(self):
        """加载任务进度"""
        with open(self.task_file, 'r') as f:
            data = json.load(f)
            self.progress = data.get('progress', 0)

    def save_progress(self):
        """保存任务进度"""
        data = {'progress': self.progress}
        with open(self.task_file, 'w') as f:
            json.dump(data, f,indent=4, separators=(',', ': '))
            f.flush()
            os.fsync(f.fileno())  # 确保数据已写入磁盘

    def execute_task(self):

        legacy_tasks = [t  for t in self.progress if t['finish'] ==0 ]

        for i, v in enumerate(legacy_tasks):
            print(f"执行任务 {v['key']} ...")
            time.sleep(3)  # 模拟任务执行的时间
            self.mark_finished( v['key']) # 更行进度
            self.save_progress()  # 每次执行后保存进度
            print(f"任务 {v['key']} 完成。")

        print("任务完成！")

    def flush_progresses(self):
        tasks = [
            {"key":"xxx1"},
            {"key":"xxx2"},
            {"key":"xxx3"}
        ]

        self.progress = [ {"key" : t['key'], 'finish':0} for t in tasks]
        self.save_progress()

    def mark_finished(self, key):
        """标记完成"""
        print(key)
        for i, t in enumerate(self.progress) :
            if t ['key'] ==  key :
                self.progress[i]['finish'] = 1
                break
        self.save_progress()
    
    def count(self):
        """查询未完成"""
        return len( [t for t in self.progress if t['finish'] == 0])



if __name__ == "__main__":
    task = Task()
    # 第一次生成任务列表，下次任务时候注释掉此行
    task.flush_progresses()
    try:
        task.execute_task()
    except KeyboardInterrupt:
        print("\n任务被中断，进度已保存。")