评估处理300MB大JSON的Python递归代码健壮性与容错能力

现有代码的核心问题 #

• 参数设计冗余且易出错：函数强依赖调用方传入firstArray/firstObj/firstUseful三个初始值，没有做默认值兜底，只要调用方传参类型不对（比如传None、传非空的历史数据），会直接抛出异常或者返回混入脏数据的结果。
• 硬编码笔误：提取IP时写入的键名是ip_add，和预期输出要求的ip_addr不一致，会直接导致返回字段不符合业务要求。
• IP提取逻辑低效且易误判：IP提取逻辑写在字典遍历循环内，每遍历一个键就会重复执行一次IP查询，浪费性能；同时没有判断ip_addr对应值的类型，如果该字段值为None（比如示例中robottxt: null的同类场景），直接调用.get("value")会抛出属性错误；如果嵌套结构中其他位置存在同名ip_addr字段，还会提取到错误的IP值覆盖正确结果。
• 对象引用污染：递归过程中全程复用tempValue字典的引用，没有为每个节点创建独立的存储对象，上一个未达标节点（比如示例中tags为空的status节点）写入的value值会残留到后续节点的结果中，出现数据串扰。
• 类型覆盖不全：仅对字典类型做递归处理，完全跳过列表类型，但目标JSON最外层就是数组，一旦后续结构调整、目标节点出现在数组元素中，会直接漏提取；同时没有做空值容错，传入非字典类型时直接调用.items()会抛出异常。
• 递归实现不规范：递归调用时没有接收返回值，完全依赖列表的可变引用特性收集结果，代码可读性极差，后续维护很容易引入bug；且递归深度受Python默认递归栈限制，遇到嵌套极深的结构会抛出递归深度超限错误。
• 大文件适配缺失：默认依赖全量JSON加载到内存后的字典对象，300MB的JSON解析为Python对象后内存占用会达到900MB~1.2GB，内存不足时会直接触发OOM崩溃。
• 边界判断缺失：没有处理tags字段为非列表类型（None、字符串、字典等）的场景，也没有处理节点存在非空tags但缺失value字段的场景，要么漏判要么返回结构不完整的结果。

优化方向 #

• 简化函数参数，内置初始值逻辑，不需要调用方手动传入临时变量，降低调用出错概率。
• 修正字段名笔误，调整IP提取逻辑到循环外，每次进入新的字典节点时只做一次IP检查，同时加类型判断避免空值报错。
• 每次处理新节点时创建独立的临时字典，确认节点符合提取要求（非空tags、存在value）时再组装完整结果追加到列表，彻底避免引用串扰问题。
• 补全类型判断逻辑，同时支持字典、列表两种嵌套结构的递归遍历，遇到非容器类型直接跳过，兼容各种JSON结构场景。
• 大文件场景下替换全量JSON加载逻辑，改用ijson等流式解析库边读边处理，内存占用可以降到几十MB级别；如果需要兼容极深嵌套结构，可以把递归实现改为栈实现的广度优先遍历，避免递归栈溢出。
• 补充边界值判断：对tags、value字段的类型做校验，异常值按空规则跳过，避免类型错误。

参考实现代码 #

import json

def extract_target_nodes(data, current_ip=None):
    result = []
    # 处理列表结构：遍历所有元素递归
    if isinstance(data, list):
        for item in data:
            result.extend(extract_target_nodes(item, current_ip))
        return result
    # 非字典结构直接返回，避免类型错误
    if not isinstance(data, dict):
        return result
    # 提取当前层级的IP，更新上下文IP值
    ip_node = data.get("ip_addr")
    if isinstance(ip_node, dict) and "value" in ip_node:
        current_ip = ip_node["value"]
    # 遍历当前层所有节点
    for key, value in data.items():
        # 遇到容器类型递归处理
        if isinstance(value, (dict, list)):
            # 检查当前字典节点是否符合提取规则
            if isinstance(value, dict):
                tags = value.get("tags")
                node_val = value.get("value")
                # 仅保留tags为非空列表、存在value的节点
                if isinstance(tags, list) and len(tags) > 0 and node_val is not None:
                    result.append({
                        "Key": key,
                        "value": node_val,
                        "useful": {"ip_addr": current_ip} if current_ip else {},
                        "tags": tags
                    })
            # 追加子节点的提取结果
            result.extend(extract_target_nodes(value, current_ip))
    return result

# 调用示例
if __name__ == "__main__":
    # 300MB大文件建议替换为ijson流式解析，避免内存溢出
    with open("target.json", "r", encoding="utf-8") as f:
        raw_data = json.load(f)
    res = extract_target_nodes(raw_data)
    print(json.dumps(res, indent=2, ensure_ascii=False))

这段代码不需要额外传初始参数，自动兼容列表、字典嵌套结构，空值、类型异常场景不会抛出错误，也不会出现数据串扰问题，针对给出的示例JSON可以直接返回符合预期的结果。

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