如何在Python中处理重叠IP段：合并同属性段并拆分异属性段

核心逻辑 #

先将所有IP段转换为起始/结束整数（利用IP转32位整数的辅助函数），通过排序确保优先级高的小段先参与处理，最后遍历拆分/合并段，全程用整数运算保证效率。

具体步骤 #

1. 标准化IP段
   • 把所有CIDR格式的IP段转换为(起始整数, 结束整数, ASN, 国家)的元组，统一数据格式。
2. 排序IP段
   • 按起始整数升序排序；若起始整数相同，按结束整数升序排序（确保小段优先进入处理流程）。
3. 遍历处理重叠段
   • 初始化空结果列表，逐个处理排序后的IP段：
     • 若结果列表为空，直接加入当前段。
     • 取出结果列表最后一个已处理段，分情况处理：
       • 无重叠：当前段起始大于已处理段结束，直接加入结果列表。
       • 属性相同：合并成覆盖范围更大的段（取两者起始最小值、结束最大值），替换结果列表最后一个元素。
       • 属性不同：
         1. 拆分已处理段中未被当前段覆盖的前半部分，替换结果列表最后一个元素。
         2. 加入当前段。
         3. 若当前段结束小于已处理段结束，拆分并加入已处理段未被覆盖的后半部分。
4. 优化结果（可选）
   • 遍历结果列表，合并连续且属性相同的IP段，转换回CIDR格式减少结果数量。

百万级数据优化点 #

• 数据库端排序：如果数据量过大无法全量加载到内存，直接用SQLite的ORDER BY start_int, end_int语句排序，逐行读取处理。
• 整数运算优先：全程用整数处理IP地址，避免字符串操作带来的性能损耗。
• 线性遍历处理：仅与结果列表最后一个元素对比，整体时间复杂度为O(n log n)（主要来自排序），适配百万级数据规模。

示例验证 #

输入：

• 1.0.0.0/8 (ASN1, CN)
• 1.0.0.0/24 (ASN2, US)

处理后输出：

• 1.0.0.0/24 (ASN2, US)
• 1.0.1.0/24 ~ 1.255.255.255（合并为最大CIDR段：1.0.1.0/8）(ASN1, CN)

代码框架（伪代码） #

# 假设已有辅助函数：
# cidr_to_int_range(cidr) -> (start_int, end_int)
# int_to_cidr(start_int, end_int) -> cidr_str

def process_ip_segments(segments):
    # 标准化段
    standardized = []
    for cidr, asn, country in segments:
        start, end = cidr_to_int_range(cidr)
        standardized.append( (start, end, asn, country) )
    
    # 排序
    standardized.sort(key=lambda x: (x[0], x[1]))
    
    # 遍历处理
    result = []
    for curr in standardized:
        curr_start, curr_end, curr_asn, curr_country = curr
        if not result:
            result.append(curr)
            continue
        
        last_start, last_end, last_asn, last_country = result[-1]
        
        # 无重叠
        if curr_start > last_end:
            result.append(curr)
            continue
        
        # 属性相同则合并
        if curr_asn == last_asn and curr_country == last_country:
            new_start = last_start
            new_end = max(last_end, curr_end)
            result[-1] = (new_start, new_end, curr_asn, curr_country)
            continue
        
        # 属性不同则拆分
        if last_start < curr_start:
            result[-1] = (last_start, curr_start - 1, last_asn, last_country)
        result.append(curr)
        if curr_end < last_end:
            result.append( (curr_end + 1, last_end, last_asn, last_country) )
    
    # 合并连续同属性段
    merged_result = []
    for seg in result:
        if not merged_result:
            merged_result.append(seg)
            continue
        last_seg = merged_result[-1]
        if (last_seg[2] == seg[2] and last_seg[3] == seg[3] and last_seg[1] + 1 == seg[0]):
            new_start = last_seg[0]
            new_end = seg[1]
            merged_result[-1] = (new_start, new_end, seg[2], seg[3])
        else:
            merged_result.append(seg)
    
    # 转换回CIDR格式
    final_result = []
    for start, end, asn, country in merged_result:
        cidr = int_to_cidr(start, end)
        final_result.append( (cidr, asn, country) )
    
    return final_result

内容的提问来源于stack exchange，提问作者Ξένη Γήινος