使用Shell/Python解析非标准类JSON文件并扁平化输出为CSV

实现思路 #

你提供的输入本质是用空行分隔的事件块集合，每个事件块包含若干顶级键值对，其中FileAttributes_s为对象数组，需要将数组中的每个对象与当前事件的顶级字段合并为一行CSV输出。实现逻辑为：

1. 逐行流式读取输入文件，按空行分割为单个事件块，避免一次性加载全量文件
2. 自动识别所有出现的字段（包括顶级字段和FileAttributes_s内的字段），无需硬编码适配新增属性
3. 内置基础格式修复能力，处理输入中的HTML转义、语法小错误等问题
4. 流式写入CSV文件，无临时文件生成

完整Python代码 #

import json
import csv
import re

def fix_json_str(s: str) -> str:
    """修复输入中常见的JSON语法错误，可根据实际情况扩展规则"""
    # 修复漏写左引号的情况，比如 "Name": Plan Growth Number" 转为 "Name": "Plan Growth Number"
    s = re.sub(r':\s*([^"].*?")', r': "\1', s)
    return s

def process_event_block(block_lines: list, csv_writer: csv.DictWriter, headers: set):
    """处理单个事件块，解析后写入CSV"""
    top_fields = {}
    attr_json_lines = []
    in_attr_array = False

    for line in block_lines:
        # 先处理HTML转义
        line = line.replace('"', '"').strip()
        if not line:
            continue
        
        # 判断是否进入FileAttributes_s的数组段
        if line.startswith('FileAttributes_s'):
            in_attr_array = True
            if '[' in line:
                start_idx = line.index('[')
                attr_json_lines.append(line[start_idx:])
            continue
        
        if in_attr_array:
            attr_json_lines.append(line)
            if ']' in line:
                in_attr_array = False
                # 拼接完整的JSON数组字符串并解析
                attr_json_str = ' '.join(attr_json_lines)
                attr_json_str = fix_json_str(attr_json_str)
                try:
                    attributes = json.loads(attr_json_str)
                except Exception as e:
                    print(f"解析属性数组失败，跳过当前块：{e}")
                    return
            continue
        
        # 解析普通顶级键值对
        if ':' in line:
            key, val = line.split(':', 1)
            key = key.strip()
            val = val.strip()
            # 把null转为空字符串
            if val.lower() == 'null':
                val = ''
            top_fields[key] = val

    # 扁平化处理：每个属性对象合并顶级字段为一行记录
    for attr in attributes:
        # 合并字段
        row = top_fields.copy()
        for k, v in attr.items():
            # 处理null值
            row[k] = v if v is not None else ''
        # 新增字段同步到表头
        new_fields = set(row.keys()) - headers
        if new_fields:
            headers.update(new_fields)
            csv_writer.fieldnames = list(headers)
        # 写入行
        csv_writer.writerow(row)

if __name__ == "__main__":
    # 配置输入输出路径
    INPUT_PATH = "full.json"
    OUTPUT_PATH = "output.csv"

    headers = set()
    block_buffer = []

    with open(INPUT_PATH, 'r', encoding='utf-8') as in_f, \
         open(OUTPUT_PATH, 'w', encoding='utf-8-sig', newline='') as out_f:
        # 初始化CSV写入器，表头动态更新
        csv_writer = csv.DictWriter(out_f, fieldnames=[], quoting=csv.QUOTE_MINIMAL)
        # 先写入空表头，后续有新增字段会自动覆盖第一行表头
        csv_writer.writeheader()

        for line in in_f:
            stripped_line = line.strip()
            # 空行代表一个事件块结束
            if not stripped_line and block_buffer:
                process_event_block(block_buffer, csv_writer, headers)
                block_buffer = []
                # 重写最新表头（处理新增字段的情况）
                out_f.seek(0)
                csv_writer.writeheader()
                out_f.seek(0, 2)
            else:
                block_buffer.append(line)
        
        # 处理最后一个没有空行结尾的事件块
        if block_buffer:
            process_event_block(block_buffer, csv_writer, headers)
            out_f.seek(0)
            csv_writer.writeheader()

方案优势 #

• 字段自适应：无需硬编码任何键名，新增顶级字段或FileAttributes_s内字段都可以自动识别输出
• 大文件支持：逐行流式读写，内存占用稳定在KB级，即使后续文件增长到GB级也可以正常处理
• 无临时文件：全程无中间文件生成，仅输入输出两个文件IO
• 容错能力：内置常见格式错误修复逻辑，可根据实际输入场景扩展修复规则
• 输出兼容：生成标准CSV格式，支持Excel、pandas等各类工具直接读取

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