如何递归优先处理JSON最深层级，拆分含[split]的键为$ref引用

问题背景 #

现有含[split]标识键的JSON数据，需求如下：

• 将所有带[split]标识的键对应内容，拆分到以[split]后字符串命名的独立JSON文件中
• 原位置替换为指向该文件的$ref引用
• 必须优先处理最深层级的拆分（例如先拆分invoice对象内部的marketPartner[split]Invoice.MarketPartner，再拆分invoice[split]Invoice自身）
• 已实现第一层拆分逻辑，但无法处理递归深层级的拆分，数组部分可自行扩展处理

示例JSON结构：

{
    "version": 16,
    "customer[split]Invoice.Customer": {
        "authorizedAt": null,
        "createdAt": "2022-07-27 19:27:22",
        "zip": "3141"
    },
    "invoice[split]Invoice": {
        "externalInvoice": 0,
        "invoiceTriggeredByUser": true,
        "billingAddress": {
            "company": null
        },
        "marketPartner[split]Invoice.MarketPartner": {
            "partnerId": "123"
        }
    },
    "contracts": [
       {
           "contract[split]Invoice.Contract": {
               "startDate": "2007-05-10",
               "endDate": null
           }
       }
    ]
}

核心实现思路：深度优先遍历（DFS） #

要实现"先处理最深层级"的要求，必须采用深度优先遍历逻辑，核心步骤：

• 遍历当前对象的所有键值对
• 如果值是对象类型，先递归调用拆分函数处理该子对象，确保子对象内部的[split]键先完成拆分
• 处理当前键：若键包含[split]，执行拆分（保存文件、替换为$ref）；否则直接保留处理后的内容
• 递归处理子对象时，需返回处理后的结果，确保当前层级处理时，子内容已经是完成深层拆分的状态

修改后的递归实现代码 #

import json
import os

class JsonSplitter:
    def __init__(self, input_event):
        self.__input_event = input_event
        # 确保输出目录存在，避免写入失败
        os.makedirs('output/models', exist_ok=True)

    def split_all(self):
        # 启动递归拆分
        return self.__split_recursive(self.__input_event)

    def __split_recursive(self, obj):
        # 非对象类型直接返回（数组部分可自行扩展处理逻辑）
        if not isinstance(obj, dict):
            return obj

        processed_obj = {}
        for key, value in obj.items():
            # 第一步：先递归处理子对象，保证深层级优先拆分
            processed_value = self.__split_recursive(value)

            # 第二步：处理当前键的拆分逻辑
            if '[split]' in key:
                # 拆分键名，获取对象别名和文件名
                key_parts = key.split('[split]')
                object_name = key_parts[0]
                file_name = key_parts[1]

                # 将处理后的子内容（已完成深层拆分）写入文件
                with open(f'output/models/{file_name}.json', 'w') as f:
                    json.dump(processed_value, f, indent=4)

                # 原位置替换为$ref引用
                processed_obj[object_name] = {'$ref': f'./models/{file_name}.json'}
            else:
                # 无split标识，直接保留处理后的值
                processed_obj[key] = processed_value

        return processed_obj

# 使用示例
if __name__ == '__main__':
    input_json = {
        "version": 16,
        "customer[split]Invoice.Customer": {
            "authorizedAt": None,
            "createdAt": "2022-07-27 19:27:22",
            "zip": "3141"
        },
        "invoice[split]Invoice": {
            "externalInvoice": 0,
            "invoiceTriggeredByUser": True,
            "billingAddress": {
                "company": None
            },
            "marketPartner[split]Invoice.MarketPartner": {
                "partnerId": "123"
            }
        },
        "contracts": [
           {
               "contract[split]Invoice.Contract": {
                   "startDate": "2007-05-10",
                   "endDate": None
               }
           }
        ]
    }

    splitter = JsonSplitter(input_json)
    final_result = splitter.split_all()
    # 保存最终的主JSON文件
    with open('output/main.json', 'w') as f:
        json.dump(final_result, f, indent=4)

关键逻辑说明 #

• 递归优先级：在处理当前键之前，先对值执行递归拆分，确保子对象内的所有[split]键先完成文件生成和$ref替换
• 层级处理顺序：以invoice[split]Invoice为例，会先处理其内部的marketPartner[split]Invoice.MarketPartner，生成Invoice.MarketPartner.json并替换为$ref，之后再处理invoice[split]Invoice本身，生成的Invoice.json中已经包含处理后的子内容
• 健壮性处理：提前创建输出目录，避免文件写入时出现目录不存在的错误

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