Python读取多JSON文件，按db、loginID和日期分组统计生成CSV

核心思路 #

• 自动遍历目标目录下所有JSON文件，无需手动指定单个文件名（如json1、json2）
• 解析每条数据的StartTime字段，提取纯日期部分作为分组维度之一
• 用嵌套字典实现高效分组统计：日期 -> (db, loginID) -> 统计数量
• 按日期生成对应CSV文件，支持增量追加（后续同日期的新数据可直接追加到已有文件）

代码实现（Python） #

import json
import os
import csv
from datetime import datetime
from collections import defaultdict

def process_json_files(json_dir):
    # 初始化统计结构：日期键对应嵌套字典，存储(db, loginID)的计数
    stats = defaultdict(lambda: defaultdict(int))

    # 遍历目录下所有JSON文件
    for filename in os.listdir(json_dir):
        if not filename.lower().endswith('.json'):
            continue
        file_path = os.path.join(json_dir, filename)
        
        # 读取并解析JSON文件
        with open(file_path, 'r', encoding='utf-8') as f:
            try:
                data = json.load(f)
                # 统一转为列表格式（兼容单条数据的JSON文件）
                data_list = data if isinstance(data, list) else [data]
            except json.JSONDecodeError as e:
                print(f"跳过损坏的文件 {filename}: {str(e)}")
                continue
        
        # 逐条处理数据
        for item in data_list:
            # 提取必填字段，缺失则跳过
            db = item.get('db')
            login_id = item.get('loginID')
            start_time = item.get('StartTime')
            if not all([db, login_id, start_time]):
                continue
            
            # 解析日期（兼容两种常见格式：带时间/纯日期）
            try:
                # 优先尝试解析带时间的格式
                date_obj = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S').date()
            except ValueError:
                try:
                    # 尝试纯日期格式
                    date_obj = datetime.strptime(start_time, '%Y-%m-%d').date()
                except ValueError:
                    print(f"无法解析日期格式 {start_time}，跳过该条数据")
                    continue
            date_str = date_obj.isoformat()
            
            # 更新统计计数
            stats[date_str][(db, login_id)] += 1
    
    # 将统计结果写入对应CSV文件
    for date_str, group_stats in stats.items():
        csv_path = f"{date_str}.csv"
        file_exists = os.path.isfile(csv_path)
        
        with open(csv_path, 'a', encoding='utf-8', newline='') as csv_file:
            writer = csv.writer(csv_file)
            # 首次写入时添加表头
            if not file_exists:
                writer.writerow(['db', 'loginID', 'count'])
            # 写入分组统计数据
            for (db, login_id), count in group_stats.items():
                writer.writerow([db, login_id, count])
    
    print("所有JSON文件处理完成")

# 调用示例：替换为你的JSON文件所在目录
if __name__ == "__main__":
    process_json_files("./json_data")

方案优势 #

• 自动化批量处理：无需手动维护文件名列表，新增JSON文件直接放入目录即可处理
• 高效低内存：边读取边统计，无需一次性加载所有数据，适合处理大体积JSON文件
• 容错性强：捕获JSON解析错误、字段缺失、日期格式异常等情况，避免程序中断
• 增量支持：CSV采用追加模式，重复运行脚本可合并同日期的新数据，无需重新生成文件
• 灵活兼容：适配两种常见的StartTime格式，可根据实际格式调整日期解析规则

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