文件监控循环重启后无法断点续传的解决方案咨询

开发了一个File Watcher Loop程序，运行时会扫描指定文件夹中的.json文件，将数据追加到output.xlsx文件中，之后持续循环扫描新增文件并重复该流程。但当程序因电脑关机等原因停止后，文件夹中新增的文件在重启程序后无法被继续处理，必须删除output.xlsx重新开始。请问如何让程序记录已处理文件的历史，重启后仅处理未追加的文件？

原代码如下：

import os
import glob
import json
import pandas as pd
import time
from datetime import datetime
import openpyxl
from openpyxl.utils.dataframe import dataframe_to_rows

def jsonFilesInDirectory(my_dir: str):
    # 获取目录中的JSON文件列表
    json_files = glob.glob(os.path.join(my_dir, "*.json"))
    return json_files

def clean_value(value):
    # 清理数值，移除不必要的字符
    return float(value.replace('\xa0s', '').replace('\xa0ms', '').replace(',', ''))

def doThingsWithNewFiles(fileDiff: list, my_dir: str, workbook):
    for file_name in fileDiff:
        file_path = os.path.join(my_dir, file_name)
        with open(file_path, 'r', encoding='utf-8') as file:
            try:
                json_data = json.load(file)

                # 从JSON文件中提取数据
                url = json_data["finalUrl"]
                fetch_time = json_data["fetchTime"]

                audits = json_data["audits"]
                fcp_metric = audits["first-contentful-paint"]["id"]
                fcp_value = audits["first-contentful-paint"]["displayValue"]
                fcp_score = audits["first-contentful-paint"]["score"]
                lcp_metric = audits["largest-contentful-paint"]["id"]
                lcp_value = audits["largest-contentful-paint"]["displayValue"]
                lcp_score = audits["largest-contentful-paint"]["score"]
                fmp_metric = audits["first-meaningful-paint"]["id"]
                fmp_value = audits["first-meaningful-paint"]["displayValue"]
                fmp_score = audits["first-meaningful-paint"]["score"]
                si_metric = audits["speed-index"]["id"]
                si_value = audits["speed-index"]["displayValue"]
                si_score = audits["speed-index"]["score"]
                tbt_metric = audits["total-blocking-time"]["id"]
                tbt_value = audits["total-blocking-time"]["displayValue"]
                tbt_score = audits["total-blocking-time"]["score"]
                cls_metric = audits["cumulative-layout-shift"]["id"]
                cls_value = audits["cumulative-layout-shift"]["displayValue"]
                cls_score = audits["cumulative-layout-shift"]["score"]

                categories = json_data["categories"]
                perf_metric = categories["performance"]["id"]
                perf_value = 0
                perf_score = categories["performance"]["score"]

                # 清理数值并格式化抓取时间
                cleaned_fcp_value = clean_value(fcp_value)
                cleaned_lcp_value = clean_value(lcp_value)
                cleaned_fmp_value = clean_value(fmp_value)
                cleaned_si_value = clean_value(si_value)
                cleaned_tbt_value = clean_value(tbt_value)
                datetime_obj = datetime.strptime(fetch_time, "%Y-%m-%dT%H:%M:%S.%fZ")
                cleaned_fetch_time = datetime_obj.strftime("%Y-%m-%d %H:%M:%S")

                # 为DataFrame创建数据字典
                data_dict = {
                    "fetch_time": [cleaned_fetch_time] * 7,
                    "url": [url] * 7,
                    "metric": ["performance","first_contentful_paint", "largest_contentful_paint",
                               "first-meaningful-paint", "speed-index", "total-blocking-time",
                               "cumulative-layout-shift"],
                    "value": [perf_value, cleaned_fcp_value, cleaned_lcp_value,
                              cleaned_fmp_value, cleaned_si_value, cleaned_tbt_value,
                              cls_value],
                    "score": [perf_score, fcp_score, lcp_score, fmp_score, si_score, tbt_score, cls_score]
                }

                df = pd.DataFrame(data_dict)

                # 将DataFrame追加到Excel文件
                sheet_name = "Sheet1"
                if sheet_name in workbook.sheetnames:
                    sheet = workbook[sheet_name]
                    startrow = sheet.max_row
                    for row in dataframe_to_rows(df, index=False, header=False):
                        sheet.append(row)
                else:
                    sheet = workbook.create_sheet(sheet_name)
                    for row in dataframe_to_rows(df, index=False, header=True):
                        sheet.append(row)

                print(f"已从{file_name}提取数据并追加到Excel文件")

            except KeyError as e:
                print(f"处理文件'{file_name}'时发生KeyError: {e}")
            except json.JSONDecodeError as e:
                print(f"处理文件'{file_name}'时发生JSONDecodeError: {e}")
            except Exception as e:
                print(f"处理文件'{file_name}'时发生错误: {e}")

def fileWatcher(my_dir: str, pollTime: int):
    excel_file_path = os.path.join(my_dir, 'output.xlsx')
    existingFiles = []

    # 检查输出文件是否已存在
    if os.path.isfile(excel_file_path):
        try:
            workbook = openpyxl.load_workbook(excel_file_path)
            existingFiles = jsonFilesInDirectory(my_dir)
            # 处理已存在的JSON文件并将数据追加到Excel文件
            doThingsWithNewFiles(existingFiles, my_dir, workbook)
            print("已处理现有JSON文件并将数据追加到Excel文件")
        except openpyxl.utils.exceptions.InvalidFileException:
            workbook = openpyxl.Workbook()
    else:
        workbook = openpyxl.Workbook()

    # 启动时检查新文件
    newFileList = jsonFilesInDirectory(my_dir)
    fileDiff = listComparison(existingFiles, newFileList)
    existingFiles = newFileList

    if len(fileDiff) > 0:
        # 处理新文件并将数据追加到Excel文件
        doThingsWithNewFiles(fileDiff, my_dir, workbook)

        # 保存Excel文件
        workbook.save(excel_file_path)
        print(f"DataFrame已导出到{excel_file_path}")

    while True:
        time.sleep(pollTime)

        # 获取目录中更新后的JSON文件列表
        newFileList = jsonFilesInDirectory(my_dir)

        # 比较新旧文件列表的差异
        fileDiff = listComparison(existingFiles, newFileList)
        existingFiles = newFileList

        if len(fileDiff) > 0:
            # 处理新文件并将数据追加到Excel文件
            doThingsWithNewFiles(fileDiff, my_dir, workbook)

            # 保存Excel文件
            workbook.save(excel_file_path)
            print(f"DataFrame已导出到{excel_file_path}")


def listComparison(originalList: list, newList: list):
    # 比较两个列表并返回差异
    differencesList = [x for x in newList if x not in originalList]
    return differencesList

my_dir = r"Z:"
pollTime = 60

fileWatcher(my_dir, pollTime)

核心思路是维护一个已处理文件记录文件，持久化保存已经处理过的JSON文件路径，程序启动时读取该记录，只处理未在记录中的文件，处理完成后更新记录文件。

修改后的完整代码如下：

import os
import glob
import json
import pandas as pd
import time
from datetime import datetime
import openpyxl
from openpyxl.utils.dataframe import dataframe_to_rows

def jsonFilesInDirectory(my_dir: str):
    # 获取目录中的JSON文件列表（返回绝对路径）
    json_files = glob.glob(os.path.join(my_dir, "*.json"))
    # 转为绝对路径，避免路径格式不一致导致的判断错误
    return [os.path.abspath(file) for file in json_files]

def clean_value(value):
    # 清理数值，移除不必要的字符
    return float(value.replace('\xa0s', '').replace('\xa0ms', '').replace(',', ''))

def load_processed_files(record_path):
    # 加载已处理文件记录
    if os.path.exists(record_path):
        with open(record_path, 'r', encoding='utf-8') as f:
            try:
                return json.load(f)
            except json.JSONDecodeError:
                return []
    return []

def save_processed_files(record_path, processed_files):
    # 保存已处理文件记录
    with open(record_path, 'w', encoding='utf-8') as f:
        json.dump(processed_files, f, indent=2)

def doThingsWithNewFiles(fileDiff: list, my_dir: str, workbook):
    processed_files = []
    for file_path in fileDiff:
        file_name = os.path.basename(file_path)
        try:
            with open(file_path, 'r', encoding='utf-8') as file:
                json_data = json.load(file)

                # 从JSON文件中提取数据
                url = json_data["finalUrl"]
                fetch_time = json_data["fetchTime"]

                audits = json_data["audits"]
                fcp_metric = audits["first-contentful-paint"]["id"]
                fcp_value = audits["first-contentful-paint"]["displayValue"]
                fcp_score = audits["first-contentful-paint"]["score"]
                lcp_metric = audits["largest-contentful-paint"]["id"]
                lcp_value = audits["largest-contentful-paint"]["displayValue"]
                lcp_score = audits["largest-contentful-paint"]["score"]
                fmp_metric = audits["first-meaningful-paint"]["id"]
                fmp_value = audits["first-meaningful-paint"]["displayValue"]
                fmp_score = audits["first-meaningful-paint"]["score"]
                si_metric = audits["speed-index"]["id"]
                si_value = audits["speed-index"]["displayValue"]
                si_score = audits["speed-index"]["score"]
                tbt_metric = audits["total-blocking-time"]["id"]
                tbt_value = audits["total-blocking-time"]["displayValue"]
                tbt_score = audits["total-blocking-time"]["score"]
                cls_metric = audits["cumulative-layout-shift"]["id"]
                cls_value = audits["cumulative-layout-shift"]["displayValue"]
                cls_score = audits["cumulative-layout-shift"]["score"]

                categories = json_data["categories"]
                perf_metric = categories["performance"]["id"]
                perf_value = 0
                perf_score = categories["performance"]["score"]

                # 清理数值并格式化抓取时间
                cleaned_fcp_value = clean_value(fcp_value)
                cleaned_lcp_value = clean_value(lcp_value)
                cleaned_fmp_value = clean_value(fmp_value)
                cleaned_si_value = clean_value(si_value)
                cleaned_tbt_value = clean_value(tbt_value)
                datetime_obj = datetime.strptime(fetch_time, "%Y-%m-%dT%H:%M:%S.%fZ")
                cleaned_fetch_time = datetime_obj.strftime("%Y-%m-%d %H:%M:%S")

                # 为DataFrame创建数据字典
                data_dict = {
                    "fetch_time": [cleaned_fetch_time] * 7,
                    "url": [url] * 7,
                    "metric": ["performance","first_contentful_paint", "largest_contentful_paint",
                               "first-meaningful-paint", "speed-index", "total-blocking-time",
                               "cumulative-layout-shift"],
                    "value": [perf_value, cleaned_fcp_value, cleaned_lcp_value,
                              cleaned_fmp_value, cleaned_si_value, cleaned_tbt_value,
                              cls_value],
                    "score": [perf_score, fcp_score, lcp_score, fmp_score, si_score, tbt_score, cls_score]
                }

                df = pd.DataFrame(data_dict)

                # 将DataFrame追加到Excel文件
                sheet_name = "Sheet1"
                if sheet_name in workbook.sheetnames:
                    sheet = workbook[sheet_name]
                    startrow = sheet.max_row
                    for row in dataframe_to_rows(df, index=False, header=False):
                        sheet.append(row)
                else:
                    sheet = workbook.create_sheet(sheet_name)
                    for row in dataframe_to_rows(df, index=False, header=True):
                        sheet.append(row)

                print(f"已从{file_name}提取数据并追加到Excel文件")
                processed_files.append(file_path)

        except KeyError as e:
            print(f"处理文件'{file_name}'时发生KeyError: {e}")
        except json.JSONDecodeError as e:
            print(f"处理文件'{file_name}'时发生JSONDecodeError: {e}")
        except Exception as e:
            print(f"处理文件'{file_name}'时发生错误: {e}")
    return processed_files

def fileWatcher(my_dir: str, pollTime: int):
    excel_file_path = os.path.join(my_dir, 'output.xlsx')
    record_file_path = os.path.join(my_dir, 'processed_files.json')
    
    # 加载已处理文件记录
    processed_files = load_processed_files(record_file_path)

    # 初始化Excel工作簿
    if os.path.isfile(excel_file_path):
        try:
            workbook = openpyxl.load_workbook(excel_file_path)
        except openpyxl.utils.exceptions.InvalidFileException:
            workbook = openpyxl.Workbook()
    else:
        workbook = openpyxl.Workbook()

    # 启动时扫描所有JSON文件，筛选未处理的
    all_json_files = jsonFilesInDirectory(my_dir)
    unprocessed_files = [file for file in all_json_files if file not in processed_files]

    if len(unprocessed_files) > 0:
        # 处理未处理的文件
        newly_processed = doThingsWithNewFiles(unprocessed_files, my_dir, workbook)
        # 更新已处理记录
        processed_files.extend(newly_processed)
        save_processed_files(record_file_path, processed_files)
        # 保存Excel
        workbook.save(excel_file_path)
        print(f"已处理{len(unprocessed_files)}个未处理文件，数据已保存到{excel_file_path}")

    # 进入循环扫描
    while True:
        time.sleep(pollTime)

        # 获取当前所有JSON文件
        current_json_files = jsonFilesInDirectory(my_dir)
        # 筛选新增的未处理文件
        new_files = [file for file in current_json_files if file not in processed_files]

        if len(new_files) > 0:
            # 处理新增文件
            newly_processed = doThingsWithNewFiles(new_files, my_dir, workbook)
            # 更新记录
            processed_files.extend(newly_processed)
            save_processed_files(record_file_path, processed_files)
            # 保存Excel
            workbook.save(excel_file_path)
            print(f"已处理{len(new_files)}个新增文件，数据已保存到{excel_file_path}")

def listComparison(originalList: list, newList: list):
    # 比较两个列表并返回差异（保留原函数，兼容可能的其他调用）
    differencesList = [x for x in newList if x not in originalList]
    return differencesList

my_dir = r"Z:"
pollTime = 60

fileWatcher(my_dir, pollTime)

关键改动说明 #

• 新增已处理文件记录功能：添加load_processed_files和save_processed_files函数，通过processed_files.json文件持久化保存已处理的JSON文件绝对路径，避免程序重启后丢失记录。
• 修改文件路径处理：jsonFilesInDirectory返回文件的绝对路径，防止相对路径、不同格式路径导致的判断错误。
• 调整启动逻辑：程序启动时不再默认处理所有文件，而是读取已处理记录，仅处理未在记录中的文件。
• 更新处理流程：每次处理完文件后，立即将成功处理的文件路径添加到记录中并保存，确保即使程序意外终止，已处理记录也不会丢失。

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