如何高效过滤嵌套字典中符合多字段条件的 NetFlow 数据

本文介绍一种灵活、可扩展的方法，用于根据用户传入的字典型过滤条件（如 `{'dstport': '443', 'srcaddr': '192.168.10.10'}`），精准提取嵌套 netflow 数据字典中匹配的 packet 及其子结构（如 flowset 和 flow），并保持原始嵌套结构。

在处理网络流量分析数据（如 Cisco NetFlow v9 解析后的结构化字典）时，常需按协议字段（如 srcaddr、dstport、protocol）进行细粒度筛选。但原始数据结构高度嵌套且键名冗长（如 "cflow.FlowSet 14 [id=10000] (1 flows).Flow 1.cflow.dstport"），直接使用 dict.get() 或逐层遍历易出错、难维护。下面提供一个健壮、可读性强、支持多条件 AND 逻辑的过滤方案。

✅ 核心思路：键名模糊匹配 + 值精确校验

由于目标字段（如 dstport）可能出现在任意层级的键中（如 cflow.dstport、...Flow 1.cflow.dstport），我们不依赖固定路径，而是：

• 遍历每个 packet（如 "packet27"）下的所有键；
• 检查键名是否包含待过滤字段名（如 "dstport" 是 "cflow.FlowSet 14 [...] .Flow 1.cflow.dstport" 的子串）；
• 若匹配，再比对对应值是否等于过滤条件中的期望值；
• 所有条件同时满足的 packet 才被保留，并仅保留其包含匹配字段的完整 FlowSet/Flow 子树（非整个 packet）。

? 实现代码（推荐版本）

def filter_nested_netflow(data: dict, filter_criteria: dict) -> dict:
    """
    过滤嵌套 NetFlow 字典，返回仅含匹配 FlowSet/Flow 的精简结构

    Args:
        data: 原始嵌套字典（key 为 packet 名，value 为该 packet 的全部字段）
        filter_criteria: 过滤条件字典，如 {'srcaddr': '192.168.10.10', 'dstport': '443'}

    Returns:
        过滤后字典，结构同输入，但每个 packet 下仅保留满足所有条件的 FlowSet/Flow 相关键值对
    """
    result = {}

    for packet_name, packet_dict in data.items():
        if not isinstance(packet_dict, dict):
            continue

        # 存储当前 packet 中匹配的所有键值对
        matched_entries = {}

        # 对每个过滤条件，查找所有匹配的键值
        for field, expected_value in filter_criteria.items():
            for key, value in packet_dict.items():
                # 关键：判断字段名是否作为子串出现在 key 中（忽略大小写和前缀）
                if field.lower() in key.lower():
                    if str(value) == str(expected_value):  # 统一转字符串比较，避免类型差异
                        matched_entries[key] = value

        # ⚠️ 注意：此处需确保 *同一个 FlowSet/Flow 下所有条件均被满足*
        # 简单策略：只保留那些 key 能“覆盖”所有条件字段的子树（见下方增强版）
        # 基础版：若至少有一个匹配，则暂存（适合快速原型）
        if matched_entries:
            result[packet_name] = matched_entries

    return result

# 使用示例
filter_criteria = {'srcaddr': '192.168.10.10', 'dstport': '443'}
filtered = filter_nested_netflow(netflow_data, filter_criteria)

? 增强版：确保同一 FlowSet/Flow 内部全条件命中

基础版可能将不同 FlowSet 中的 srcaddr 和 dstport 拼凑在一起（误报）。更严谨的做法是按 FlowSet 分组，再检查组内是否同时存在所有条件字段：

import re

def filter_by_flowset(data: dict, filter_criteria: dict) -> dict:
    """增强版：按 FlowSet 分组，确保所有条件在同一 FlowSet/Flow 内满足"""
    result = {}

    for packet_name, packet_dict in data.items():
        if not isinstance(packet_dict, dict):
            continue

        # 提取所有 FlowSet 相关键（如 "FlowSet 14 [id=10000] (1 flows)"）
        flowset_keys = [k for k in packet_dict.keys() 
                       if re.match(r'FlowSet \d+ \[id=\d+\] \(\d+ flows\)', k)]

        packet_matches = {}

        # 遍历每个 FlowSet
        for fs_key in flowset_keys:
            # 收集该 FlowSet 下所有键值（包括子项，如 ".Flow 1.cflow.srcaddr"）
            fs_entries = {k: v for k, v in packet_dict.items() 
                         if k == fs_key or k.startswith(fs_key + '.') or k.startswith('cflow.' + fs_key + '.')}

            # 检查该 FlowSet 是否满足全部条件
            all_matched = True
            for field, expected in filter_criteria.items():
                found = False
                for k, v in fs_entries.items():
                    if field.lower() in k.lower() and str(v) == str(expected):
                        found = True
                        break
                if not found:
                    all_matched = False
                    break

            if all_matched:
                # 保留整个 FlowSet 及其所有相关键（含 padding、template 等）
                for k, v in packet_dict.items():
                    if k == fs_key or k.startswith(fs_key + '.') or k.startswith('cflow.' + fs_key + '.'):
                        packet_matches[k] = v

        if packet_matches:
            result[packet_name] = packet_matches

    return result

⚠️ 注意事项与最佳实践

• 字符串化比较：NetFlow 字段值可能为 int、str 或 float，统一用 str(value) == str(expected) 避免类型不匹配；
• 键名模糊性：srcaddr 可能出现在 cflow.srcaddr、cflow.Flow 1.cflow.srcaddr 等位置，正则或 in 判断更鲁棒；
• 性能优化：对超大数据集，可预编译正则、使用生成器或 filter() 函数减少内存占用；
• 扩展性：支持添加 operator 参数（如 {'dstport': ('>=', 443)}）实现范围查询；
• 输出验证：建议在生产环境添加日志，记录匹配的 FlowSet ID 和 packet 名，便于审计。

通过以上方法，您可精准、可维护地从复杂嵌套结构中提取所需流量片段，为后续分析（如异常检测、会话还原）奠定坚实基础。