Python文件监控与JSON转XLSX代码调试求助

我需要实现一个循环程序，每60秒检查指定文件夹是否新增.json文件，提取文件中的FCP、LCP等性能指标并导出到XLSX文件。代码运行无报错，但始终不生成output.xlsx；单独测试各代码模块均正常，整合后出现异常，请求帮忙调试。

原代码 #

import os
from os import listdir
from os.path import isfile, join, splitext
import json
import pandas as pd
import time

# 获取目录下的所有JSON文件
def jsonFilesInDirectory(my_dir: str):
    onlyfiles = [f for f in listdir(my_dir) if isfile(join(my_dir, f)) and splitext(f)[1].lower() == '.json']
    return onlyfiles

# 对比两个列表，返回新列表中独有的元素（新增文件）
def listComparison(originalList: list, newList: list):
    differencesList = [x for x in newList if x not in originalList]  # 注：无法检测文件删除
    return differencesList

def doThingsWithNewFiles(fileDiff: list, my_dir: str):
    for file_name in fileDiff:
        file_path = os.path.join(my_dir, file_name)
        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"]
        fcp_value = json_data["audits"]["first-contentful-paint"]["displayValue"]
        fcp_score = json_data["audits"]["first-contentful-paint"]["score"]
        lcp_value = json_data["audits"]["largest-contentful-paint"]["displayValue"]
        lcp_score = json_data["audits"]["largest-contentful-paint"]["score"]
        fmp_value = json_data["audits"]["first-meaningful-paint"]["displayValue"]
        fmp_score = json_data["audits"]["first-meaningful-paint"]["score"]
        si_value = json_data["audits"]["speed-index"]["displayValue"]
        si_score = json_data["audits"]["speed-index"]["score"]
        tbt_value = json_data["audits"]["total-blocking-time"]["displayValue"]
        tbt_score = json_data["audits"]["total-blocking-time"]["score"]
        cls_value = json_data["audits"]["cumulative-layout-shift"]["displayValue"]
        cls_score = json_data["audits"]["cumulative-layout-shift"]["score"]

        # 清洗指标值
        cleaned_fcp_value = fcp_value.replace('\xa0s', '')
        cleaned_lcp_value = lcp_value.replace('\xa0s', '')
        cleaned_fmp_value = fmp_value.replace('\xa0s', '')
        cleaned_si_value = si_value.replace('\xa0s', '')
        cleaned_tbt_value = tbt_value.replace('\xa0ms', '')

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

        df = pd.DataFrame(data_dict)

        # 导出到Excel
        excel_file_path = os.path.join(my_dir, 'output.xlsx')
        if os.path.exists(excel_file_path):
            with pd.ExcelWriter(excel_file_path, engine='openpyxl', mode='a') as writer:
                df.to_excel(writer, sheet_name='Sheet1', index=False, header=False,
                            startrow=writer.sheets['Sheet1'].max_row)
        else:
            df.to_excel(excel_file_path, index=False)

        print(f"DataFrame exported to {excel_file_path}")

def fileWatcher(my_dir: str, pollTime: int):
    while True:
        if 'watching' not in locals():  # 判断是否首次运行
            previousFileList = jsonFilesInDirectory(my_dir)
            watching = 1
            print('First Time')
            print(previousFileList)

        time.sleep(pollTime)

        newFileList = jsonFilesInDirectory(my_dir)

        fileDiff = listComparison(previousFileList, newFileList)

        previousFileList = newFileList
        if len(fileDiff) == 0:
            continue
        doThingsWithNewFiles(fileDiff, my_dir)

my_dir = r"C:\Users\84948\Desktop\EC\Project\Test_Folder"
pollTime = 60

fileWatcher(my_dir, pollTime)

问题排查与修复方案 #

1. 修复文件监控的首次执行逻辑 #

原代码通过locals()判断首次运行的方式不够可靠，且首次初始化文件列表后直接进入60秒休眠，会导致启动后立即新增的文件无法及时被检测。修改后的监控函数更直观且可靠：

def fileWatcher(my_dir: str, pollTime: int):
    # 初始化文件列表
    previousFileList = jsonFilesInDirectory(my_dir)
    print('首次运行，当前JSON文件列表:', previousFileList)
    
    # 可选：如果需要处理启动时已存在的文件，取消下方注释
    # if previousFileList:
    #     doThingsWithNewFiles(previousFileList, my_dir)
    
    while True:
        time.sleep(pollTime)
        newFileList = jsonFilesInDirectory(my_dir)
        fileDiff = listComparison(previousFileList, newFileList)
        
        # 更新文件列表
        previousFileList = newFileList
        
        if len(fileDiff) == 0:
            print(f'无新增JSON文件，继续监控...')
            continue
        
        print(f'检测到新增文件: {fileDiff}，开始处理')
        doThingsWithNewFiles(fileDiff, my_dir)

2. 增强Excel写入的可靠性 #

原代码未处理Excel文件存在但Sheet1不存在的情况，且缺少异常捕获，添加后可以避免静默失败：

# 替换doThingsWithNewFiles中的Excel导出部分
excel_file_path = os.path.join(my_dir, 'output.xlsx')
if os.path.exists(excel_file_path):
    try:
        with pd.ExcelWriter(excel_file_path, engine='openpyxl', mode='a') as writer:
            # 检查Sheet1是否存在，不存在则创建
            if 'Sheet1' not in writer.sheets:
                df.to_excel(writer, sheet_name='Sheet1', index=False)
            else:
                start_row = writer.sheets['Sheet1'].max_row
                df.to_excel(writer, sheet_name='Sheet1', index=False, header=False, startrow=start_row)
        print(f'已追加数据到 {excel_file_path}')
    except Exception as e:
        print(f'追加Excel数据时出错: {str(e)}')
else:
    try:
        df.to_excel(excel_file_path, index=False, engine='openpyxl')
        print(f'已创建新Excel文件: {excel_file_path}')
    except Exception as e:
        print(f'创建Excel文件时出错: {str(e)}')

3. 检查依赖安装 #

确保已安装所有必要的依赖包：

pip install pandas openpyxl

4. 新增调试日志 #

在关键步骤添加日志输出，方便定位问题，比如在doThingsWithNewFiles中打印正在处理的文件名：

print(f'正在处理文件: {file_name}')

完整修正后的代码 #

import os
from os import listdir
from os.path import isfile, join, splitext
import json
import pandas as pd
import time

# 获取目录下的所有JSON文件
def jsonFilesInDirectory(my_dir: str):
    onlyfiles = [f for f in listdir(my_dir) if isfile(join(my_dir, f)) and splitext(f)[1].lower() == '.json']
    return onlyfiles

# 对比两个列表，返回新列表中独有的元素（新增文件）
def listComparison(originalList: list, newList: list):
    differencesList = [x for x in newList if x not in originalList]  # 注：无法检测文件删除
    return differencesList

def doThingsWithNewFiles(fileDiff: list, my_dir: str):
    for file_name in fileDiff:
        print(f'正在处理文件: {file_name}')
        file_path = os.path.join(my_dir, file_name)
        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"]
        fcp_value = json_data["audits"]["first-contentful-paint"]["displayValue"]
        fcp_score = json_data["audits"]["first-contentful-paint"]["score"]
        lcp_value = json_data["audits"]["largest-contentful-paint"]["displayValue"]
        lcp_score = json_data["audits"]["largest-contentful-paint"]["score"]
        fmp_value = json_data["audits"]["first-meaningful-paint"]["displayValue"]
        fmp_score = json_data["audits"]["first-meaningful-paint"]["score"]
        si_value = json_data["audits"]["speed-index"]["displayValue"]
        si_score = json_data["audits"]["speed-index"]["score"]
        tbt_value = json_data["audits"]["total-blocking-time"]["displayValue"]
        tbt_score = json_data["audits"]["total-blocking-time"]["score"]
        cls_value = json_data["audits"]["cumulative-layout-shift"]["displayValue"]
        cls_score = json_data["audits"]["cumulative-layout-shift"]["score"]

        # 清洗指标值
        cleaned_fcp_value = fcp_value.replace('\xa0s', '')
        cleaned_lcp_value = lcp_value.replace('\xa0s', '')
        cleaned_fmp_value = fmp_value.replace('\xa0s', '')
        cleaned_si_value = si_value.replace('\xa0s', '')
        cleaned_tbt_value = tbt_value.replace('\xa0ms', '')

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

        df = pd.DataFrame(data_dict)

        # 导出到Excel
        excel_file_path = os.path.join(my_dir, 'output.xlsx')
        if os.path.exists(excel_file_path):
            try:
                with pd.ExcelWriter(excel_file_path, engine='openpyxl', mode='a') as writer:
                    # 检查Sheet1是否存在，不存在则创建
                    if 'Sheet1' not in writer.sheets:
                        df.to_excel(writer, sheet_name='Sheet1', index=False)
                    else:
                        start_row = writer.sheets['Sheet1'].max_row
                        df.to_excel(writer, sheet_name='Sheet1', index=False, header=False, startrow=start_row)
                print(f'已追加数据到 {excel_file_path}')
            except Exception as e:
                print(f'追加Excel数据时出错: {str(e)}')
        else:
            try:
                df.to_excel(excel_file_path, index=False, engine='openpyxl')
                print(f'已创建新Excel文件: {excel_file_path}')
            except Exception as e:
                print(f'创建Excel文件时出错: {str(e)}')

def fileWatcher(my_dir: str, pollTime: int):
    # 初始化文件列表
    previousFileList = jsonFilesInDirectory(my_dir)
    print('首次运行，当前JSON文件列表:', previousFileList)
    
    # 可选：如果需要处理启动时已存在的文件，取消下方注释
    # if previousFileList:
    #     doThingsWithNewFiles(previousFileList, my_dir)
    
    while True:
        time.sleep(pollTime)
        newFileList = jsonFilesInDirectory(my_dir)
        fileDiff = listComparison(previousFileList, newFileList)
        
        # 更新文件列表
        previousFileList = newFileList
        
        if len(fileDiff) == 0:
            print(f'无新增JSON文件，继续监控...')
            continue
        
        print(f'检测到新增文件: {fileDiff}，开始处理')
        doThingsWithNewFiles(fileDiff, my_dir)

my_dir = r"C:\Users\84948\Desktop\EC\Project\Test_Folder"
pollTime = 60

fileWatcher(my_dir, pollTime)

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