如何在Python中将JSON转为对应行结构的扁平化JSON以生成CSV

问题背景 #

需要将嵌套JSON转换为适合生成CSV的扁平化结构，核心要求是把orderLines列表中的每一项展开为独立行，同时保留其他公共字段。

示例输入JSON #

{
  "transportOrder": {
    "customerId":"877299" ,
    "customerOrder": "155564649",
    "customerReference": "reference2",
    "creationDateTime": "2022-08-26T16:30:56.000Z",
    "orderDetail": {
      "AdditionalInfo": {
        "info1": "abc",
        "info2": "cds",
        "name1": "Jonathan",
        "name2": "Grulich"
      }},
    "orderLines": [
      {
        "amount": 7,
        "code": "EUP"
      },
      {
        "amount": 8,
        "code": "ENP"
      },
      {
        "amount": 17,
        "code": "ERP"
      }
    ]
  }
}

期望扁平化结果（对应CSV行） #

customerId    customerOrder    customerReference    creationDateTime    info1    info2    name1    name2    amount    code
877299    155564649    reference2    26.08.2022    abc    cds    Jonathan    Grulich    7    EUP
877299    155564649    reference2    26.08.2022    abc    cds    Jonathan    Grulich    8    ENP
877299    155564649    reference2    26.08.2022    abc    cds    Jonathan    Grulich    17    ERP

当前实现代码（使用flatdict） #

import flatdict

data = {
  "Order": {
    "customerId":"877299" ,
    "customerOrder": "155564649",
    "customerReference": "reference2",
    "creationDateTime": "2022-08-26T16:30:56.000Z",
    "orderDetail": {
      "AdditionalInfo": {
        "info1": "abc",
        "info2": "cds",
        "name1": "Jonathan",
        "name2": "Grulich"
      }},
    "orderLines": [
      {
        "amount": 7,
        "code": "EUP"
      },
      {
        "amount": 8,
        "code": "ENP"
      },
      {
        "amount": 17,
        "code": "ERP"
      }
    ]
  }
}
flat = flatdict.FlatDict(data, delimiter='.')

result_list = []
j=0

for line in flat['Order.orderLines']:
    temp = flat
    temp['amount'] = line['amount']
    temp['code'] = line['code']
    result_list.append(temp)

print(result_list)

当前方案存在冗余，寻求更优实现方式。

可以不依赖第三方库，手动处理嵌套结构扁平化与列表展开，逻辑更直接且灵活可控。

实现思路 #

• 提取公共字段（排除orderLines）并递归扁平化嵌套字典；
• 转换日期格式为目标样式；
• 遍历orderLines，将公共字段与每行数据合并生成独立扁平化字典；
• 收集所有行后可直接输出或生成CSV。

代码实现 #

from datetime import datetime
import csv

def flatten_nested_dict(nested_dict, parent_key='', sep='_'):
    """递归扁平化嵌套字典，自定义分隔符"""
    items = []
    for key, value in nested_dict.items():
        new_key = f"{parent_key}{sep}{key}" if parent_key else key
        if isinstance(value, dict):
            items.extend(flatten_nested_dict(value, new_key, sep=sep).items())
        else:
            items.append((new_key, value))
    return dict(items)

def process_order_data(input_data):
    # 提取主订单数据，分离orderLines列表
    transport_order = input_data['transportOrder'].copy()
    order_lines = transport_order.pop('orderLines')
    
    # 扁平化公共字段
    flattened_common = flatten_nested_dict(transport_order)
    
    # 转换日期格式
    if 'creationDateTime' in flattened_common:
        dt = datetime.fromisoformat(flattened_common['creationDateTime'].replace('Z', '+00:00'))
        flattened_common['creationDateTime'] = dt.strftime('%d.%m.%Y')
    
    # 生成每行数据
    processed_rows = []
    for line in order_lines:
        row = flattened_common.copy()
        row.update(line)
        processed_rows.append(row)
    
    return processed_rows

# 处理示例数据
sample_data = {
  "transportOrder": {
    "customerId":"877299" ,
    "customerOrder": "155564649",
    "customerReference": "reference2",
    "creationDateTime": "2022-08-26T16:30:56.000Z",
    "orderDetail": {
      "AdditionalInfo": {
        "info1": "abc",
        "info2": "cds",
        "name1": "Jonathan",
        "name2": "Grulich"
      }},
    "orderLines": [
      {
        "amount": 7,
        "code": "EUP"
      },
      {
        "amount": 8,
        "code": "ENP"
      },
      {
        "amount": 17,
        "code": "ERP"
      }
    ]
  }
}

rows = process_order_data(sample_data)

# 打印结果（模拟CSV格式）
headers = rows[0].keys()
print('\t'.join(headers))
for row in rows:
    print('\t'.join(str(v) for v in row.values()))

# 生成CSV文件
with open('transport_orders.csv', 'w', newline='', encoding='utf-8') as csv_file:
    writer = csv.DictWriter(csv_file, fieldnames=headers, delimiter='\t')
    writer.writeheader()
    writer.writerows(rows)

方案优势 #

• 无第三方依赖：仅使用Python标准库，无需额外安装包；
• 灵活性强：可自定义扁平化分隔符、日期格式，适配不同业务需求；
• 性能更优：避免第三方库的封装开销，逻辑直接高效；
• 扩展性好：后续新增嵌套结构或特殊字段处理，可直接在核心函数中扩展。

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