如何用Python将嵌套JSON数组转换为可搜索的扁平JSON？

Absolutely get where you're coming from—nested JSON arrays like your ports list become unsearchable blobs in Elasticsearch, which kills your core use case. Let's fix this with a Python script that flattens the nested structure, either turning each port into its own Elasticsearch document (the most flexible option for searching) or expanding ports into indexed fields in a single host document.

方案1：将每个端口转换为独立的扁平文档（推荐） #

这种方式把每个端口作为Elasticsearch的一条独立记录，同时保留主机的基础信息，让你可以轻松搜索特定端口、服务状态等条件。

import json

def flatten_dict(d, parent_key='', sep='_'):
    """递归扁平化嵌套字典，移除键开头的下划线，用下划线连接嵌套层级"""
    items = []
    for k, v in d.items():
        # 处理键名：去掉开头的下划线，拼接父键
        cleaned_key = k.lstrip('_')
        new_key = f"{parent_key}{sep}{cleaned_key}" if parent_key else cleaned_key
        
        if isinstance(v, dict):
            # 递归处理嵌套字典
            items.extend(flatten_dict(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)

def process_nmap_hosts(input_json):
    """处理Nmap生成的JSON，为每个端口生成独立的扁平文档"""
    nmap_data = json.loads(input_json)
    host_data = nmap_data.get('host', {})
    
    # 扁平化主机基础信息（排除ports数组）
    flattened_host = flatten_dict(host_data)
    # 移除原始的端口相关字段（我们会单独处理每个port）
    keys_to_remove = [k for k in flattened_host if k.startswith('ports_port') or k.startswith('ports_extraports')]
    for key in keys_to_remove:
        del flattened_host[key]
    
    # 遍历每个端口，合并主机信息和端口信息
    port_list = host_data.get('ports', {}).get('port', [])
    final_documents = []
    for port in port_list:
        flattened_port = flatten_dict(port)
        # 合并主机和端口的扁平数据
        combined_doc = {**flattened_host, **flattened_port}
        final_documents.append(combined_doc)
    
    return final_documents

# 示例使用
if __name__ == "__main__":
    # 读取你的Nmap JSON文件
    with open('nmap_scan_result.json', 'r') as input_file:
        raw_json_content = input_file.read()
    
    # 处理生成扁平文档
    processed_docs = process_nmap_hosts(raw_json_content)
    
    # 将结果写入每行一个文档的JSON文件（适合Logstash批量导入）
    with open('flattened_ports_docs.json', 'w') as output_file:
        for doc in processed_docs:
            json.dump(doc, output_file)
            output_file.write('\n')

效果说明 #

处理后，每个端口会变成类似这样的独立文档：

{
  "host_status_state": "up",
  "host_status_reason": "echo-reply",
  "host_address_addr": "xxx.xxx.xxx.xxx",
  "protocol": "tcp",
  "portid": "22",
  "state_state": "open",
  "service_name": "ssh",
  "service_product": "OpenSSH",
  ...
}

现在你可以直接在Elasticsearch中搜索portid:22或者service_name:ssh这类条件了。

方案2：将所有端口展开到单个主机文档中 #

如果你希望把整个主机的所有信息放在一个Elasticsearch文档里，可以把端口信息展开为带索引编号的字段（比如port_1_portid、port_2_service_name）：

def flatten_to_single_host_doc(input_json):
    nmap_data = json.loads(input_json)
    host_data = nmap_data.get('host', {})
    flattened_host = flatten_dict(host_data)
    
    # 处理每个端口，添加索引前缀
    port_list = host_data.get('ports', {}).get('port', [])
    for port_index, port in enumerate(port_list, 1):
        flattened_port = flatten_dict(port)
        for field_name, field_value in flattened_port.items():
            flattened_host[f"port_{port_index}_{field_name}"] = field_value
    
    # 移除原始的端口数组字段
    if 'ports_port' in flattened_host:
        del flattened_host['ports_port']
    return flattened_host

效果说明 #

处理后，主机文档会包含类似port_1_portid:20、port_2_service_product:vsftpd这样的字段，你可以通过port_2_service_name:ftp来搜索第二个端口的服务名。

后续导入Elasticsearch #

处理后的JSON文件可以直接用Logstash导入：

• 对于方案1的每行一个文档格式，Logstash的file输入插件可以直接读取并批量索引到Elasticsearch。
• 确保Elasticsearch的索引映射自动识别这些扁平字段（或者手动创建映射以优化搜索性能）。

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