基于动态映射解析转换嵌套JSON数据的Python实现求助

针对动态映射嵌套JSON并支持聚合计算的需求，以下是具体的解决思路、代码实现和工具库建议：

核心思路 #

1. 依赖成熟的JSON路径库：避免手动编写递归解析嵌套结构的代码，直接用支持路径查询的工具处理嵌套字段提取和数组遍历（比如orders[*].total这类语法）。
2. 区分规则类型：将映射规则分为两类处理：
   • 普通路径提取：直接通过JSON路径定位并提取数据。
   • 聚合/计算指令：识别函数前缀（如sum()），解析内部路径后执行对应计算。
3. 添加异常处理：处理路径不存在、数据类型不匹配等情况，保证转换稳定性。

推荐工具库 #

• jmespath：原生支持JSON路径查询和多种内置聚合函数（sum、avg、count等），语法和你的映射规则高度兼容，无需大量自定义代码。
• jsonpath-ng：提供更灵活的路径解析能力，支持自定义函数扩展，适合需要复杂自定义逻辑的场景。
• pydantic：若需同时进行数据校验和转换，可结合其模型定义与自定义字段实现，但整体复杂度稍高。

代码示例（基于jmespath） #

安装依赖 #

pip install jmespath

实现脚本 #

import json
import jmespath
import re

# 自定义函数映射（可扩展）
FUNCTION_MAP = {
    'sum': lambda values: sum(values) if isinstance(values, list) and all(isinstance(v, (int, float)) for v in values) else None,
    'avg': lambda values: sum(values)/len(values) if isinstance(values, list) and values else None,
    'count': lambda values: len(values) if isinstance(values, list) else None
}

def apply_json_mapping(input_data, mapping):
    result = {}
    # 适配输入为数组或单个对象的情况
    target_data = input_data[0] if isinstance(input_data, list) else input_data
    
    for output_key, rule in mapping.items():
        # 匹配函数式规则（如sum(...)）
        func_match = re.match(r'^(\w+)\((.*)\)$', rule.strip())
        if func_match:
            func_name, path = func_match.groups()
            # 提取路径对应的数据
            raw_values = jmespath.search(path.strip(), target_data)
            # 执行自定义函数
            result[output_key] = FUNCTION_MAP.get(func_name, lambda x: None)(raw_values)
        else:
            # 普通路径提取
            try:
                result[output_key] = jmespath.search(rule, target_data)
            except:
                result[output_key] = None
    return result

# 加载输入数据与映射配置
if __name__ == '__main__':
    with open('sample_data.json', 'r', encoding='utf-8') as f:
        input_json = json.load(f)
    
    with open('sample_maping.json', 'r', encoding='utf-8') as f:
        mapping_config = json.load(f)
    
    # 生成转换结果
    output_json = apply_json_mapping(input_json, mapping_config)
    print(json.dumps(output_json, indent=2, ensure_ascii=False))

输出结果 #

运行上述脚本后，会生成如下转换后的JSON：

{
  "user_id": 1,
  "username": "Markus",
  "purchase_history": [
    {
      "order_id": 59748,
      "total": 12
    },
    {
      "order_id": 59742,
      "total": 70
    }
  ],
  "total_spent": 82
}

扩展建议 #

• 若需要支持更多自定义函数，只需在FUNCTION_MAP中添加对应的函数逻辑即可。
• 可添加日志记录，跟踪转换过程中的异常情况，方便调试。
• 对于复杂的嵌套数组处理，jmespath支持更丰富的过滤、排序语法，可直接集成到映射规则中。

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