Python脚本处理AWS Transcribe JSON文件时JSONDecodeError问题及优化

核心问题分析 #

1. JSON解码错误：watchdog的on_created事件会在文件刚创建时立即触发，但此时文件可能还在下载/写入过程中，内容为空或不完整，导致json.load()读取失败。
2. 时间记录错误：dt_string是脚本启动时生成的固定时间，所有后续处理的记录都会使用这个时间，不符合“记录文件处理时实际时间”的需求。
3. 转录文本提取逻辑脆弱：通过df1.iloc[3]['results']提取数据依赖DataFrame的行索引，一旦JSON结构微调就会失效，应该直接操作JSON字典。

修复方案 #

1. 等待文件写入完成 #

添加重试机制，尝试读取文件直到成功：

def read_json_with_retry(file_path, max_retries=5, delay=1):
    for _ in range(max_retries):
        try:
            with open(file_path, 'r') as f:
                return json.load(f)
        except json.JSONDecodeError:
            time.sleep(delay)
    raise Exception(f"Failed to read {file_path} after {max_retries} retries")

2. 实时获取处理时间 #

每次处理文件时重新生成当前时间，替换脚本启动时的固定时间：

dt_string = datetime.now().strftime("%m/%d/%Y %H:%M:%S")

3. 正确提取转录文本 #

直接从JSON字典中获取目标字段，无需转换为DataFrame：

transcript = data['results']['transcripts'][0]['transcript']

代码优化建议 #

• 过滤文件类型：只处理.json后缀的文件，避免无关文件触发错误。
• 简化Excel操作：使用pandas的mode='a'模式追加数据，无需手动操作openpyxl。
• 移除冗余导入：删除未使用的threading模块。
• 增强异常处理：捕获更多可能的异常（如文件读取失败、JSON字段缺失），避免脚本崩溃。
• 清理冗余代码：移除with语句内的file.close()（with会自动关闭文件）。

完整修正代码 #

import time
import json
from datetime import datetime
import pandas as pd
from watchdog.events import FileSystemEventHandler, FileCreatedEvent
from watchdog.observers import Observer

def read_json_with_retry(file_path, max_retries=5, delay=1):
    """重试读取JSON文件，直到文件写入完成"""
    for _ in range(max_retries):
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                return json.load(f)
        except json.JSONDecodeError:
            time.sleep(delay)
        except FileNotFoundError:
            time.sleep(delay)
    raise Exception(f"无法读取文件 {file_path}，已重试{max_retries}次")

class CustomHandler(FileSystemEventHandler):
    def on_created(self, event):
        if isinstance(event, FileCreatedEvent) and event.src_path.endswith('.json'):
            print(f"检测到新文件: {event.src_path}")
            
            try:
                # 读取JSON数据
                data = read_json_with_retry(event.src_path)
                
                # 提取转录文本
                transcript = data['results']['transcripts'][0]['transcript']
                
                # 获取当前处理时间
                dt_string = datetime.now().strftime("%m/%d/%Y %H:%M:%S")
                
                # 构造数据
                new_row = pd.DataFrame({
                    'Date & Time': [dt_string],
                    'Radio Transmission': [transcript]
                })
                
                # 生成当日Excel文件名
                now = datetime.now()
                daily_file = f"D:/AWS Transcribe/2 Daily Transcriptions/{now.strftime('%m-%d-%Y')}.xlsx"
                
                # 写入/追加Excel
                try:
                    # 如果文件存在，追加数据
                    with pd.ExcelWriter(daily_file, engine='openpyxl', mode='a', if_sheet_exists='overlay') as writer:
                        start_row = writer.sheets['Sheet1'].max_row
                        new_row.to_excel(writer, index=False, header=False, startrow=start_row)
                    print(f"已追加数据到 {daily_file}")
                except FileNotFoundError:
                    # 文件不存在，创建新文件
                    new_row.to_excel(daily_file, index=False)
                    print(f"已创建新文件 {daily_file}")
                    
            except Exception as e:
                print(f"处理文件失败: {str(e)}")

def main():
    path_to_monitor = "D:/AWS Transcribe/1 JSON From AWS/"
    observer = Observer()
    observer.schedule(CustomHandler(), path=path_to_monitor, recursive=True)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == '__main__':
    main()

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