如何遍历键值对不规范的字典列表并提取指定字段值？

我有一个大型字典列表，键值对没有固定模式，需要提取username、hostname、host_ip等相关值。部分字典可能完全不存在目标值，比如有的包含username，有的则没有。

示例数据 #

logs = [{
        '_index': '.siem-signals-default-000015', 
        '_type': '_doc', 
        '_id': '456', 
        '_score': None, 
        '_source': {
            '@timestamp': '2023-03-22T14:05:53.855Z', 
            'ecs': {'version': '1.10.0'}, 
            'host': {
                'id': '456', 
                'name': 'SRV2', 
                'ip': ['172.30.5.115', 'fe80::5efe:ac1e:573'], 
                'mac': ['00:00:00:00:00:00:00'],
                'architecture': 'x86_64', 
                'os': {'build': '14393.5717', 'type': 'windows'}
                }, 
                'log': {'level': 'warning'}, 
                'message': 'Some message', 
                'file': {
                    'directory': 'C:', 
                    'extension': 'psm1', 
                    'path': 'C:', 
                    'name': 'Heartbeat.psm1'
                    }, 
                'agent': {
                    'name': 'SRV2', 
                    'type': 'winlogbeat', 
                    'version': '7.14.1', 
                    'hostname': 'SRV2', 
                    'ephemeral_id': '456a', 
                    'id': '4567'
                    }, 
                'winlog': {
                    'user': {'name': 'SYSTEM', 'type': 'User', 'identifier': 'S-1-5-18', 'domain': 'NT AUTHORITY'}, 
                    'event_id': '4104', 
                    'opcode': 'On create calls', 
                    'provider_name': 'Microsoft-Windows-PowerShell', 
                    'channel': 'Microsoft-Windows-PowerShell/Operational'
                    }
                }}, 

    {
        '_index': '.siem-signals-default-000015', 
        '_type': '_doc', 
        '_id': '123', 
        '_score': None, 
        '_source': {
            '@timestamp': '2023-03-22T14:05:53.854Z', 
            'winlog': {
                'user': {'identifier': 'S-1-5-18', 'domain': 'NT AUTHORITY', 'name': 'SYSTEM', 'type': 'User'}, 
                'task': 'Execute a Remote Command', 
                'computer_name': 'SRV1', 
                'provider_name': 'Microsoft-Windows-PowerShell'
                }, 
            'event': {
                'provider': 'Microsoft-Windows-PowerShell', 
                'action': 'Execute a Remote Command', 
                'created': '2023-03-22T14:02:36.498Z'
                }, 
            'log': {'level': 'warning'}, 
            'powershell': {
                'sequence': 1, 
                'total': 1, 
                'file': {
                    'script_block_id': '123', 
                    'script_block_text': "Some script"
                    }
                                }, 
            'ecs': {'version': '1.10.0'}, 
            'message': "Some message", 
            'host': {
                'mac': ['00:00:00:00:00:00:00', '00:00:00:00:00:00:00'], 
                'hostname': 'SRV1', 
                'architecture': 'x86_64', 
                'os': {
                    'build': '14393.5717', 
                    'type': 'windows'
                    }, 
                'id': '456', 
                'ip': ['172.30.5.115', 'fe80::5efe:ac1e:573']
                    }, 
            'user': {'id': 'S-1-5-18'}
                    }, 
        'sort': [1679493953854]
    }]

问题说明 #

示例中两个字典都包含hostname，但存储位置不同：第一个字典的hostname位于['_source']['host']['name']，第二个则位于['_source']['host']['hostname']。

我尝试了以下递归打印函数，但未解决问题：

def printlogs(logs):
    for items in logs:
        for k, v in items.items():
            if isinstance(v, dict):
                pritlogs(v)
            else:
                print("{0} : {1}".format(k, v))
printlogs(logs)

解决方案 #

问题分析 #

原函数存在拼写错误（pritlogs应为printlogs），且仅能打印所有键值对，无法针对性提取目标字段。需要编写递归查找逻辑，根据目标字段的可能键名遍历字典，收集所需值。

实现代码 #

def find_value(data, target_keys):
    """递归查找字典/列表中匹配目标键的值，返回第一个匹配结果"""
    if isinstance(data, dict):
        for k, v in data.items():
            if k in target_keys:
                return v
            elif isinstance(v, (dict, list)):
                result = find_value(v, target_keys)
                if result is not None:
                    return result
    elif isinstance(data, list):
        for item in data:
            result = find_value(item, target_keys)
            if result is not None:
                return result
    return None

def extract_fields(logs):
    extracted_list = []
    for log_entry in logs:
        extracted = {}
        # 提取username：匹配键名包括'name'（常见于user/winlog.user节点）
        extracted['username'] = find_value(log_entry, ['name'])
        # 提取hostname：匹配键名包括'hostname'、'name'、'computer_name'
        extracted['hostname'] = find_value(log_entry, ['hostname', 'name', 'computer_name'])
        # 提取host_ip：匹配键名'ip'，优先取IPv4地址
        ip_list = find_value(log_entry, ['ip'])
        if ip_list:
            # 筛选IPv4地址（含小数点）
            ipv4_addresses = [ip for ip in ip_list if '.' in ip]
            extracted['host_ip'] = ipv4_addresses[0] if ipv4_addresses else ip_list[0]
        else:
            extracted['host_ip'] = None
        extracted_list.append(extracted)
    return extracted_list

# 执行提取并打印结果
extracted_results = extract_fields(logs)
for res in extracted_results:
    print(res)

代码说明 #

1. find_value函数：递归遍历字典和列表，查找匹配目标键名的值，返回第一个找到的结果，适配非结构化数据的嵌套结构。
2. extract_fields函数：针对每个日志条目，调用find_value提取指定字段，同时处理IP的特殊情况（从列表中优先选取IPv4地址）。
3. 可根据实际需求扩展目标键名，比如若username还可能出现在其他键名下，只需添加到target_keys列表即可。

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