Python遍历嵌套JSON日志并提取指定字段至CSV的实现

问题分析 #

当JSON存在多层嵌套结构时，仅用json.load()加载后直接调用顶层键无法获取深层字段，需要通过层级链式访问或通用提取函数来获取目标值，再导出到CSV文件。

示例嵌套JSON日志 #

{
  "logVersion": "1.0",
  "event": {
    "incident": {
      "incidentId": "INC-2024-001",
      "timestamp": "2024-05-20T14:30:00Z"
    },
    "violator": {
      "violatorId": "USER-12345",
      "details": {
        "name": "John Doe",
        "email": "john@example.com"
      }
    },
    "metadata": {
      "source": "firewall",
      "severity": "high"
    }
  },
  "anotherEvent": {
    "incident": {
      "incidentId": "INC-2024-002"
    },
    "violator": {
      "violatorId": "USER-67890"
    }
  }
}

问题代码（仅能访问顶层） #

import json
import csv

with open('logs.json', 'r') as f:
    data = json.load(f)

with open('output.csv', 'w', newline='') as csvfile:
    fieldnames = ['violatorId', 'incidentId']
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    # 无法获取嵌套字段，返回None
    writer.writerow({
        'violatorId': data.get('violatorId'),
        'incidentId': data.get('incidentId')
    })

解决方案1：直接层级访问（结构固定场景） #

如果JSON结构固定，可直接通过链式索引获取深层字段：

import json
import csv

with open('logs.json', 'r') as f:
    data = json.load(f)

# 提取目标字段
event_data = data['event']
violator_id = event_data['violator']['violatorId']
incident_id = event_data['incident']['incidentId']

# 导出到CSV
with open('output.csv', 'w', newline='') as csvfile:
    fieldnames = ['violatorId', 'incidentId']
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerow({
        'violatorId': violator_id,
        'incidentId': incident_id
    })

解决方案2：通用递归提取函数（结构多变/批量处理场景） #

如果JSON存在多个条目或结构不固定，用递归函数可灵活提取任意深度的键：

import json
import csv

def get_nested_value(data, key_path):
    """递归提取嵌套JSON的值，key_path为键的路径列表，如['event', 'violator', 'violatorId']"""
    current = data
    for key in key_path:
        if isinstance(current, dict) and key in current:
            current = current[key]
        else:
            return None
    return current

# 加载JSON日志
with open('logs.json', 'r') as f:
    data = json.load(f)

# 定义需提取的字段及其路径
fields = {
    'violatorId': ['event', 'violator', 'violatorId'],
    'incidentId': ['event', 'incident', 'incidentId'],
    'severity': ['event', 'metadata', 'severity']
}

# 提取数据
extracted_row = {}
for field_name, path in fields.items():
    extracted_row[field_name] = get_nested_value(data, path)

# 导出到CSV
with open('output.csv', 'w', newline='') as csvfile:
    writer = csv.DictWriter(csvfile, fieldnames=fields.keys())
    writer.writeheader()
    writer.writerow(extracted_row)

# 批量处理数组格式日志示例（若logs.json为[{"event": {...}}, {"event": {...}}]）
# with open('logs.json', 'r') as f:
#     logs = json.load(f)
# with open('output.csv', 'w', newline='') as csvfile:
#     writer = csv.DictWriter(csvfile, fieldnames=fields.keys())
#     writer.writeheader()
#     for log in logs:
#         row = {}
#         for field_name, path in fields.items():
#             row[field_name] = get_nested_value(log, path)
#         writer.writerow(row)

关键点说明 #

• 直接层级访问：代码简洁，适合结构固定的场景，但JSON结构变化时需同步修改索引路径。
• 递归提取函数：扩展性强，只需修改字段路径配置即可适配不同结构，适合批量或多变的日志处理。
• 空值处理：递归函数在路径不存在时返回None，导出CSV时会显示为空，可根据需求添加默认值逻辑。

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