寻求自动提取嵌套JSON至多个DataFrame的Python实现方案

需求说明 #

需要一段Python代码，自动识别API返回的嵌套JSON中的主对象、子对象及对象列表，将信息分别提取为多个DataFrame，用于后续构建数据库。

JSON结构示例 #

{
    "ns0:sfobject": {
        "@xmlns:ns0": "urn:sfobject.sfapi.api.com",
        "ns0:id": "xxxx",
        "ns0:type": "table",
        "ns0:person": {
            "ns0:country_of_birth": "xxx",
            "ns0:date_of_birth": "xxx",
            "ns0:last_modified_by": "xxx",
            "ns0:last_modified_on": "xxx",
            "ns0:identity_information": {
                "ns0:account_uuid": "xxx",
                "ns0:created_on_timestamp": "xxx",
                "ns0:last_modified_on": "xxx"
            },
            "ns0:personal_information": [
                {
                    "ns0:created_by": "xxx",
                    "ns0:last_name": "xxx",
                    "ns0:nationality": "xxx"
                },
                {
                    "ns0:created_by": "yyy",
                    "ns0:last_name": "yyy",
                    "ns0:nationality": "yyy"
                }
            ],
            "ns0:address_information": [
                {
                    "ns0:address1": "xxx",
                    "ns0:country": "xxx",
                    "ns0:zip_code": "xxx"
                },
                {
                    "ns0:address1": "xxx",
                    "ns0:country": "xxx",
                    "ns0:zip_code": "xxx"
                }
            ],
            "ns0:employment_information": [
                {
                    "ns0:assignment_class": "xxx",
                    "ns0:assignment_uuid": "xxx",
                    "ns0:user_id": "xxx",
                    "ns0:job_information": [
                        {
                            "ns0:business_unit": "xxx",
                            "ns0:calc_method_indicator": "xxx",
                            "ns0:company": "xxx"
                        },
                        {
                            "ns0:business_unit": "yyy",
                            "ns0:calc_method_indicator": "yyy",
                            "ns0:company": "yyy"
                        }
                    ],
                    "ns0:compensation_information": [
                        {
                            "ns0:created_by": "xxx",
                            "ns0:created_on_timestamp": "xxx",
                            "ns0:paycompensation_recurring": [
                                {
                                    "ns0:annualizationFactor": "xxx",
                                    "ns0:calculated_amount": "xxx"
                                },
                                {
                                    "ns0:annualizationFactor": "yyy",
                                    "ns0:calculated_amount": "yyy"
                                }
                            ]
                        }
                    ],
                    "ns0:associated_employee_information": [
                        {
                            "ns0:country_of_birth": "xxx",
                            "ns0:created_by": "xxx",
                            "ns0:associated_employee_employment_information": {
                                "ns0:assignment_class": "xxx",
                                "ns0:user_id": "xxx"
                            }
                        }
                    ]
                }
            ]
        }
    }
}

原有代码问题 #

你的代码是硬编码指定字段提取，存在几个核心问题：

• 无法处理深层嵌套的列表结构（比如employment_information下的job_information、compensation_information等）
• 若JSON中不存在指定字段（比如phone_information）会直接抛出KeyError
• 没有添加父级关联ID，后续无法在数据库中建立表间关联

可行解决方案代码 #

下面的代码通过递归遍历JSON结构，自动识别所有对象和列表，清理字段名中的命名空间前缀（ns0:），并为子表添加父级ID用于关联：

import json
import pandas as pd
from typing import Dict, List, Any

def clean_key(key: str) -> str:
    """清理字段名，去掉ns0:前缀和@xmlns这类属性"""
    if key.startswith('@'):
        return key[1:]
    return key.replace('ns0:', '')

def process_json(data: Any, parent_id: str = None, parent_name: str = None, result: Dict[str, List[Dict]] = None) -> Dict[str, List[Dict]]:
    """递归处理嵌套JSON，提取所有对象和列表为DataFrame格式的数据"""
    if result is None:
        result = {}
    
    if isinstance(data, dict):
        # 处理单个对象，清理字段名
        cleaned_data = {clean_key(k): v for k, v in data.items()}
        # 添加父级关联ID
        if parent_id is not None and parent_name is not None:
            cleaned_data[f"{parent_name}_id"] = parent_id
        
        # 提取当前对象的唯一ID（优先用id字段）
        current_id = cleaned_data.get('id', parent_id)
        
        # 分离出子对象和列表字段
        children = {}
        list_fields = {}
        
        for k, v in cleaned_data.items():
            if isinstance(v, dict):
                children[k] = v
            elif isinstance(v, list) and len(v) > 0 and isinstance(v[0], (dict, list)):
                list_fields[k] = v
        
        # 保存当前对象的非嵌套字段数据
        current_name = parent_name if parent_name else 'sfobject'
        if current_name not in result:
            result[current_name] = []
        flat_data = {k: v for k, v in cleaned_data.items() if k not in children and k not in list_fields}
        result[current_name].append(flat_data)
        
        # 递归处理子对象
        for child_name, child_data in children.items():
            process_json(child_data, current_id, child_name, result)
        
        # 递归处理列表字段中的每个元素
        for list_name, list_data in list_fields.items():
            for item in list_data:
                process_json(item, current_id, list_name, result)
    
    return result

def main():
    # JSON文件路径
    json_file_path = r'json_rout'
    # CSV输出目录
    output_dir = r'C:\Users\MariaGironaOrts\PycharmProjects\vy-data-corporate-people-employee\data'
    
    # 加载JSON数据
    with open(json_file_path, 'r') as file:
        data = json.load(file)
    
    # 从顶层sfobject开始处理
    sf_object = data.get('ns0:sfobject', {})
    processed_data = process_json(sf_object)
    
    # 转换为DataFrame并保存为CSV
    for table_name, records in processed_data.items():
        if records:
            df = pd.DataFrame(records)
            csv_path = f"{output_dir}\\{table_name}.csv"
            df.to_csv(csv_path, index=False)
            print(f"已保存表: {table_name} 到 {csv_path}")

if __name__ == "__main__":
    main()

代码说明 #

1. clean_key函数：自动清理字段名中的ns0:前缀和@开头的属性名，让字段名更符合数据库表字段的命名习惯
2. process_json函数：递归遍历整个JSON结构，自动识别单个对象和列表，为子表添加父级ID，确保后续数据库可以建立关联关系
3. main函数：加载JSON文件，启动处理流程，将所有提取出的表保存为CSV文件

运行这段代码后，你会得到所有嵌套结构对应的CSV文件，比如person.csv、identity_information.csv、personal_information.csv等，每个子表都会包含父级ID用于关联。

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