高效识别并归类文本中重复区块的算法优化教程-图灵python

本文介绍了一种内存友好、时间高效的算法，用于在超大文本文件(600万行)中识别特定标记(如”aaaaaaa一开始的重复块，避免内存溢出，并按规范格式输出所有重复实例及其原始行号。

本文介绍了一种内存友好、时间高效的算法，用于在超大文本文件(600万行)中识别特定标记(如”aaaaaaa一开始的重复块，避免内存溢出，并按规范格式输出所有重复实例及其原始行号。

处理大规模日志、固件镜像或序列化数据(如问题所述) .vmp 当驱动标签文件)被提取以固定头部(如 "AAAAAAA"）分隔的逻辑区块，并对内容完全相同的区块进行聚类和追溯。原方案有三个瓶颈:

双重遍历+随机访问效率低下：扫描所有起始行号，然后重复 seek() + islice() 读取区块，I/O 开销巨大；
将整个块加载到内存：将整个块（可能包含数百行）转换为内存 tuple 作为字典键，导致 160 万候选块容易触发 MemoryError；
没有哈希化的字符串比较：直接使用 list/tuple 做字典键，Python 底层需要逐行比较，时间复杂 O(N×M)，N 为块数，M 平均块长。

✅ 应遵循正确的解法单流式扫描 + 内容哈希 + 行号延迟记录原则：

核心优化策略

扫描一次，边读边切块:从第一次开始 86 行开始逐行读取，遇见 "AAAAAAA" 也就是说，它被视为新块的起点，前一块结束；
用哈希代替原始内容存储:每个块的内容(不包括第一行) "AAAAAAA"）计算 xxhash（比内置 hash() 更稳定，抗碰撞)；
哈希只存在于字典中→[行号列表] 映射，不缓存原始文本；
二次扫描只提取需要输出的块：最后只出现哈希值 ≥2 第二块，重新定位并写出完整的内容(包括第一行)和所有行号。

代码的完整实现

import xxhash  # pip install xxhash（比 hashlib 更快，适合大量短文)
from pathlib import Path

INPUT_PATH = r"C:\Users\Azerty\Downloads\MyHypervisorDriver.vmp\Copuies.tag"
OUTPUT_PATH = r"C:\Users\Azerty\Downloads\MyHypervisorDriver.vmp\Result.txt"
HEADER = "AAAAAAA"

# 第一阶段：单次扫描，构建 {hash → [start_line_numbers]} 映射
block_hashes = {}
current_block_lines = []
current_start_line = 86  # 第一块开始行为第86行(1)-indexed）

with open(INPUT_PATH, 'r', encoding='utf-8') as f:
    # 跳过前85行
    for _ in range(85):
        next(f, None)

    line_num = 86
    for line in f:
        stripped = line.rstrip('\n\r')
        if stripped == HEADER:
            # 遇到新块头:保存最后一块(如果存在)
            if current_block_lines:
                # 哈希计算块体内容(不包括HEADER)
                block_body = '\n'.join(current_block_lines)
                h = xxhash.xxh64(block_body).intdigest()
                if h not in block_hashes:
                    block_hashes[h] = []
                block_hashes[h].append(current_start_line)

            # 重置:新块从现在开始
            current_block_lines = []
            current_start_line = line_num
        else:
            current_block_lines.append(stripped)
        line_num += 1

    # 处理文件末尾最后一块
    if current_block_lines:
        block_body = '\n'.join(current_block_lines)
        h = xxhash.xxh64(block_body).intdigest()
        if h not in block_hashes:
            block_hashes[h] = []
        block_hashes[h].append(current_start_line)

# 第二阶段:只对重复块(出现)≥2)精确输出执行
repeated_blocks = {h: lines for h, lines in block_hashes.items() if len(lines) >= 2}

if not repeated_blocks:
    print("未发现重复块。")
else:
    with open(OUTPUT_PATH, 'w', encoding='utf-8') as out_f:
        block_id = 1
        for h, line_numbers in repeated_blocks.items():
            # 重读完整的第一个例子(包括HEADER)
            with open(INPUT_PATH, 'r', encoding='utf-8') as f:
                # 跳到这一块开始行(1-indexed → 0-indexed）
                for _ in range(line_numbers[0] - 1):
                    next(f, None)
                # 读完整块:HEADER + 后续行，直到下一个HEADER或EOF
                block_lines = []
                for line in f:
                    stripped = line.rstrip('\n\r')
                    block_lines.append(line.rstrip('\n\r'))
                    if stripped == HEADER and len(block_lines) > 1:
                        block_lines.pop()  # 排除下一个HEADER
                        break

            # 写入结果
            out_f.write(f"block {block_id}:\n")
            out_f.write(HEADER + "\n")
            out_f.writelines(line + "\n" for line in block_lines[1:])  # 跳过HEADER
            out_f.write(f"Line numbers where this block occurs: {line_numbers}\n\n")
            block_id += 1

关键注意事项

编码安全：显式指定 encoding='utf-8'，避免 Windows 默认编码(例如 cp1252）导致乱码或读取中断；
哈希选择：xxhash 比 hashlib.md5 快 5–10 倍，且 intdigest() 输出整数，字典搜索更快；如果不能安装第三方库，可以使用 hashlib.sha256(block_body.encode()).hexdigest() 替代(性能稍低但更通用)；
内存控制：没有全块内容缓存，峰值内存仅取决于最长块的长度，而不是总块数；
行号准确性：严格按压 1-indexed 符合主题要求的行号记录和输出；
增强健壮性:实际部署可以增加 try/except 包裹 I/O 操作并添加进度日志(如每次处理) 10 万行打印 .）。

此方案在 600 万行、160 在典型的万块场景下，哈希构建可以在几秒钟内完成，内存占用稳定 MemoryError，是工业级文本区块分析的推荐实践。