You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何通用化实现多层嵌套JSON转CSV格式的Python方案?

通用递归实现多层嵌套JSON转数据库风格CSV

需求

需要将任意层级嵌套的JSON转换为数据库风格的CSV,无需预先定义所有字段键。现有代码仅支持两层嵌套结构转换,需改为递归实现以适配任意层级。

示例输入JSON

{"uls":{"equ1-L1-u": {"D": 1.10, "La": 1.50, "Lb": 1.50},"equ1-L2-u": {"D": 1.10, "La": 1.50, "Lb": 1.50},},"sls":{"cha-L1": {"Ld": 1.00, "Le": 1.00, "Lf": 1.00, "Lg": 1.00, "Lh": 1.00},"cha-L2": {"D": 1.00, "Df": 1.00},}}

期望输出CSV

Criteria,Name,D,Df,La,Lb,Ld,Le,Lf,Lg,Lh
uls,equ1-L1-u,1.10,,1.50,1.50,,,,
uls,equ1-L2-u,1.10,,1.50,1.50,,,,
sls,cha-L1,,,,,1.00,1.00,1.00,1.00,1.00
sls,cha-L2,1.00,1.00,,,,,,

现有代码(仅支持两层嵌套)

# Function to convert json objects to csv
import json
import csv

def make_csv_dict(data, key_headers):
    csv_dict = []
    for i in data:
        for j in data[i]:
            csv_dict.append({
                key_headers[0]: i,
                key_headers[1]: j,
                **data[i][j]
                })
    return csv_dict

### ENTER DATA HERE ###
key_headers = ["Criteria", "Name"]
path = "File.json"
### ENTER DATA HERE ###

# Read json
with open(path) as json_file:
    data = json.load(json_file)

# make csv_dict from .json data
csv_dict = make_csv_dict(data, key_headers)

# writing to csv file
fieldnames = ["Criteria", "Name", "D", "Df", "La", "Lb", "Lc", "Ld", "Le", "Lf", "Lg", "Lh", "Sl", "Sh", "W", "T", "A", "E"]

with open(path.replace(".json",".csv"), 'w', newline="") as f:
    writer = csv.DictWriter(f, fieldnames)
    writer.writeheader()
    writer.writerows(csv_dict)

改进后的通用递归实现

以下代码通过递归遍历JSON结构,自动收集所有层级路径和字段,无需预先定义字段名:

import json
import csv
from collections import defaultdict

def flatten_json(data, path=None, result=None):
    """递归扁平化JSON,收集所有路径和键值对"""
    if result is None:
        result = []
    if path is None:
        path = []
    
    if isinstance(data, dict):
        for key, value in data.items():
            new_path = path.copy()
            new_path.append(key)
            # 如果值是字典,继续递归;否则记录当前路径和键值
            if isinstance(value, dict):
                flatten_json(value, new_path, result)
            else:
                # 路径的前n-1项作为层级标识,最后一项是字段名
                row = {}
                # 为每个层级生成列名(如Level_1, Level_2...)
                for idx, level in enumerate(new_path[:-1]):
                    row[f"Level_{idx+1}"] = level
                # 叶子节点的键和值
                row[new_path[-1]] = value
                result.append(row)
    return result

def merge_rows(flattened_data):
    """将同一路径的键值对合并为一行"""
    merged = defaultdict(dict)
    for row in flattened_data:
        # 提取层级部分作为唯一键
        level_keys = tuple((k, v) for k, v in row.items() if k.startswith("Level_"))
        # 合并同层级的所有字段
        merged[level_keys].update(row)
    return list(merged.values())

def json_to_csv(input_path, output_path=None, level_headers=None):
    """
    将任意层级JSON转换为CSV
    :param input_path: JSON文件路径
    :param output_path: 输出CSV路径,默认替换.json为.csv
    :param level_headers: 自定义层级列名,如["Criteria", "Name"],默认用Level_1, Level_2...
    """
    if output_path is None:
        output_path = input_path.replace(".json", ".csv")
    
    # 读取JSON数据
    with open(input_path, 'r') as f:
        data = json.load(f)
    
    # 扁平化JSON
    flattened = flatten_json(data)
    if not flattened:
        return
    
    # 合并同层级的行
    merged_rows = merge_rows(flattened)
    
    # 收集所有字段名
    all_fields = set()
    for row in merged_rows:
        all_fields.update(row.keys())
    
    # 自定义层级列名(如果提供)
    if level_headers:
        # 获取所有层级列
        level_fields = [f for f in all_fields if f.startswith("Level_")]
        # 按层级排序
        level_fields.sort(key=lambda x: int(x.split("_")[1]))
        # 替换层级列名
        field_mapping = dict(zip(level_fields, level_headers))
        # 更新所有行的列名
        for row in merged_rows:
            for old_key, new_key in field_mapping.items():
                row[new_key] = row.pop(old_key)
        # 更新字段集合
        all_fields = (all_fields - set(level_fields)) | set(level_headers)
    
    # 排序字段:先层级列,再其他字段
    sorted_fields = []
    if level_headers:
        sorted_fields.extend(level_headers)
    else:
        sorted_fields.extend(sorted([f for f in all_fields if f.startswith("Level_")], key=lambda x: int(x.split("_")[1])))
    # 其他字段按字母排序
    sorted_fields.extend(sorted([f for f in all_fields if not f.startswith("Level_")]))
    
    # 写入CSV
    with open(output_path, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=sorted_fields)
        writer.writeheader()
        writer.writerows(merged_rows)

# 使用示例
if __name__ == "__main__":
    input_path = "File.json"
    # 自定义层级列名,对应示例中的Criteria和Name
    json_to_csv(input_path, level_headers=["Criteria", "Name"])

代码说明

  1. flatten_json:递归遍历JSON结构,将每个叶子节点的路径和键值对记录为单独的行,路径的每一级作为Level_1、Level_2等列。
  2. merge_rows:将同一层级路径的所有键值对合并为一行,确保每个层级对应唯一的CSV行。
  3. json_to_csv:整合扁平化和合并逻辑,支持自定义层级列名(如示例中的Criteria、Name),自动收集所有字段并写入CSV。

使用效果

运行上述代码后,示例JSON将转换为符合需求的CSV,且无需预先定义字段列表,可自动适配任意层级的嵌套JSON结构。

内容的提问来源于stack exchange,提问作者user2817017

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.20 16:44:51