如何使用Python将复杂嵌套JSON转换为CSV并优化字段提取

需求说明 #

• 提取JSON中dataRows第一层的entityId、Id字段
• 处理columnValues内的嵌套字段时，需将指定字段拆分为独立列：
  • lastName：提取name、Id，对应列名lastName_name、lastName_Id
  • companyName：提取name、Id、profileType，对应列名companyName_name、companyName_Id、companyName_profileType
• 其他字段提取核心值（如value、name、asOfdate等），排除columnValueType、accessStatus、unpublished这类无关字段
• 多值字段（如institution）用逗号分隔合并为单个单元格值

原始JSON数据 #

{
    "page": {
        "page": 1,
        "pageSize": 250
    },
    "dataRows": [
        {
            "entityId": 349255,
            "Id": "41432-95P",
            "disabled": false,
            "followed": false,
            "suggestion": false,
            "inactive": false,
            "pinned": false,
            "highlighted": false,
            "columnValues": {
                "lastName": [
                    {
                        "columnValueType": "ENTITY",
                        "accessStatus": "OK",
                        "columnValueType": "ENTITY",
                        "name": "McBrady",
                        "Id": "41432-95P",
                        "unpublished": false
                    }
                ],
                "gender": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Male"
                    }
                ],
                "hqCity": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Seattle"
                    }
                ],
                "prefix": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Dr."
                    }
                ],
                "lastUpdateDate": [
                    {
                        "columnValueType": "DATE",
                        "accessStatus": "OK",
                        "columnValueType": "DATE",
                        "expected": false,
                        "asOfdate": "2023-06-26"
                    }
                ],
                "companyName": [
                    {
                        "columnValueType": "BUSINESS_ENTITY",
                        "accessStatus": "OK",
                        "columnValueType": "BUSINESS_ENTITY",
                        "name": "Global Partnerships",
                        "Id": "56347-39",
                        "unpublished": false,
                        "profileType": "INVESTOR"
                    }
                ],
                "roles": [
                    {
                        "columnValueType": "INT_COLUMN_VALUE",
                        "accessStatus": "OK",
                        "columnValueType": "INT_COLUMN_VALUE",
                        "marked": false,
                        "value": 3
                    }
                ],
                "dailyUpdates": [],
                "assetClass": [],
                "hqCountry": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "United States"
                    }
                ],
                "latestNoteAuthor": [],
                "primaryPosition": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Chair, Enterprise Risk, Compliance and Audit Committee and Member of the Board of Directors"
                    }
                ],
                "boardSeats": [
                    {
                        "columnValueType": "INT_COLUMN_VALUE",
                        "accessStatus": "OK",
                        "columnValueType": "INT_COLUMN_VALUE",
                        "marked": false,
                        "value": 2
                    }
                ],
                "fundRoles": [],
                "institution": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Harvard University"
                    },
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "University of Oxford"
                    }
                ],
                "latestNote": [],
                "Id": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "41432-95P"
                    }
                ],
                "hqRegion": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Americas"
                    }
                ],
                "email": [],
                "dealRoles": [],
                "PrimaryCompanyType": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Not-For-Profit Venture Capital"
                    }
                ],
                "mgtRoles": [],
                "hqStateProvince": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Washington"
                    }
                ],
                "fullName": [
                    {
                        "columnValueType": "ENTITY_WITH_NOTE",
                        "accessStatus": "OK",
                        "columnValueType": "ENTITY_WITH_NOTE",
                        "name": "Matthew McBrady Ph.D",
                        "Id": "41432-95P",
                        "unpublished": false
                    }
                ],
                "hqLocation": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Seattle, WA"
                    }
                ],
                "biography": [
                    {
                        "columnValueType": "DESCRIPTION_WITH_SOURCE",
                        "accessStatus": "OK",
                        "columnValueType": "DESCRIPTION_WITH_SOURCE",
                        "value": "Dr. Matthew McBrady serves as Chair, of the Enterprise Risk, Compliance, and Audit Committee.",
                        "morningstarSource": true
                    }
                ],
                "firstName": [
                    {
                        "columnValueType": "ENTITY",
                        "accessStatus": "OK",
                        "columnValueType": "ENTITY",
                        "name": "Matthew",
                        "Id": "41432-95P",
                        "unpublished": false
                    }
                ],
                "phone": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "+1 (206) 652-8773"
                    }
                ],
                "hqSubRegion": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "North America"
                    }
                ],
                "hqAddressLine2": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "Suite 410"
                    }
                ],
                "hqAddressLine1": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "1201 Western Avenue"
                    }
                ],
                "hqFax": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "+1 (206) 456-7877"
                    }
                ],
                "middleName": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "R."
                    }
                ],
                "companyWebsite": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "www.globalpartnerships.org"
                    }
                ],
                "hqZipCode": [
                    {
                        "columnValueType": "STRING",
                        "accessStatus": "OK",
                        "columnValueType": "STRING",
                        "value": "98101"
                    }
                ],
                "weeklyUpdates": []
            }
        }
    ]
}

现有代码的问题 #

• 未提取dataRows第一层的entityId、Id字段
• 无法将嵌套字段（如lastName的name和Id）拆分为独立列，只能合并为单一字符串
• 部分特殊字段（如lastUpdateDate的asOfdate）未正确提取

修正后的Python代码 #

import csv
import json

# 定义需要展开为多列的字段及其子字段映射
EXPAND_FIELDS = {
    "lastName": ["name", "Id"],
    "companyName": ["name", "Id", "profileType"]
}

# 定义需要排除的无关字段
EXCLUDE_FIELDS = {"columnValueType", "accessStatus", "unpublished"}

def process_column_value(column_name, items):
    """处理单个columnValues字段，返回对应的值或键值对"""
    if column_name in EXPAND_FIELDS:
        # 处理需要展开的字段
        result = {}
        sub_fields = EXPAND_FIELDS[column_name]
        for item in items:
            for sub_field in sub_fields:
                if sub_field in item:
                    result[f"{column_name}_{sub_field}"] = str(item[sub_field])
        return result
    else:
        # 处理普通字段，提取核心值
        values = []
        for item in items:
            # 优先提取value，其次是name、asOfdate，再处理其他非排除字段
            if "value" in item:
                values.append(str(item["value"]))
            elif "name" in item:
                values.append(str(item["name"]))
            elif "asOfdate" in item:
                values.append(str(item["asOfdate"]))
            else:
                # 收集其他非排除字段的值
                for k, v in item.items():
                    if k not in EXCLUDE_FIELDS:
                        values.append(str(v))
        return {column_name: ", ".join(values) if values else ""}

def main():
    # 读取JSON数据
    with open('data.json', 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    data_rows = data.get("dataRows", [])
    if not data_rows:
        print("无数据可处理")
        return
    
    # 收集所有表头
    headers = ["entityId", "Id"]
    processed_rows = []
    
    for row in data_rows:
        # 提取第一层字段
        row_data = {
            "entityId": row.get("entityId", ""),
            "Id": row.get("Id", "")
        }
        
        # 处理columnValues
        column_values = row.get("columnValues", {})
        for col_name, items in column_values.items():
            processed = process_column_value(col_name, items)
            row_data.update(processed)
        
        # 收集表头（第一次循环时初始化）
        if not headers:
            headers = list(row_data.keys())
        else:
            # 补充新增的表头（如果有）
            for key in row_data.keys():
                if key not in headers:
                    headers.append(key)
        
        processed_rows.append(row_data)
    
    # 写入CSV文件
    with open('data_converted.csv', 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=headers)
        writer.writeheader()
        writer.writerows(processed_rows)
    
    print("CSV转换完成，文件名为data_converted.csv")

if __name__ == "__main__":
    main()

代码说明 #

1. 字段映射配置：通过EXPAND_FIELDS定义需要拆分为多列的字段，方便后续扩展
2. 字段提取逻辑：
   • 展开字段会生成如lastName_name、lastName_Id的独立列
   • 普通字段优先提取value、name、asOfdate，其他非排除字段合并为字符串
   • 空列表字段会生成空字符串值
3. 表头动态生成：自动收集所有出现的字段作为CSV表头，适配不同数据结构
4. 编码处理：使用utf-8编码读取和写入文件，避免中文乱码问题

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