如何在Python中用流式正则解析大型二进制文件？

我正尝试在Python中实现类似strings(1)的功能，现有实现代码如下：

import re

def strings(f, n=5):
    # TODO: support files larger than available RAM
    return re.finditer(br'[!-~\s]{%i,}' % n, f.read())

if __name__ == '__main__':
    import sys
    with open(sys.argv[1], 'rb') as f:
        for m in strings(f):
            print(m[0].decode().replace('\x0A', '\u240A'))

当前代码的问题是：当文件体积超出可用内存时直接失效。如果直接迭代文件对象f，即便处理二进制文件也会按行读取，这并不合适：

• 一是可能导致与直接对整个输入执行正则匹配的结果不一致；
• 二是若文件中存在匹配rb'[^\n]{8589934592,}'的内容，仍会引发内存问题。

请问Python的正则库是否提供简便方法，通过流式方式使用re.finditer处理二进制文件？

*注：我知晓存在部分正则表达式的执行需占用与输入长度呈指数级相关的CPU或内存，此类情况不在讨论范围内；假设机器具备足够资源处理正则表达式、输入中的最大匹配结果、获取该匹配结果及忽略非匹配内容。

此外，本问题并非以下问题的重复：

• 与《Regular expression parsing a binary file?》不同，后者聚焦于类字节对象，而本问题针对二进制文件本身；
• 与《Parse a binary file with Regular Expressions?》不同，原因同上；
• 与《Regular expression for binary files》不同，后者仅针对已知所有匹配偏移量的特殊场景；
• 与另一篇《Regular expression for binary files》不同，原因兼具上述两点。

解决方案 #

Python标准库的re模块本身没有原生支持流式正则匹配的API，但我们可以通过分块读取文件+维护未匹配的剩余字节的方式实现流式处理，同时保证匹配结果和全量读取文件时完全一致。

核心思路：

1. 每次从文件中读取固定大小的块（比如64KB或1MB）；
2. 将上一次处理后剩余的不完整匹配前缀，与当前块拼接；
3. 用正则查找所有匹配项，保留块末尾可能的不完整匹配片段（最多n-1个字节，确保跨块的连续可打印字符不会被截断）；
4. 循环处理直到文件结束，最后检查剩余字节中是否有符合要求的匹配。

针对strings(1)场景的具体实现：

import re
import sys

def strings_stream(f, n=5, block_size=1024*64):
    pattern = re.compile(br'[!-~\s]{%i,}' % n)
    leftover = b''  # 保存上一次块末尾的不完整匹配前缀

    while True:
        block = f.read(block_size)
        if not block:
            break
        
        # 拼接剩余字节与当前块
        data = leftover + block
        matches = list(pattern.finditer(data))

        if matches:
            # 处理除最后一个可能跨块的匹配外的所有结果
            for match in matches[:-1]:
                yield match.group()
            
            last_match = matches[-1]
            if last_match.end() == len(data):
                # 最后一个匹配刚好到块末尾，直接返回并清空剩余字节
                yield last_match.group()
                leftover = b''
            else:
                # 保留块末尾最多n-1个字节，避免截断跨块的连续匹配
                leftover = data[-(n-1):] if len(data) >= n-1 else data
        else:
            # 当前块无匹配，保留可能的前缀（最多n-1个字节）
            leftover = data[-(n-1):] if len(data) >= n-1 else data
    
    # 处理文件结束后剩余的字节
    if len(leftover) >= n:
        for match in pattern.finditer(leftover):
            yield match.group()

if __name__ == '__main__':
    if len(sys.argv) != 2:
        print(f"Usage: {sys.argv[0]} <file>")
        sys.exit(1)
    with open(sys.argv[1], 'rb') as f:
        for s in strings_stream(f):
            print(s.decode().replace('\x0A', '\u240A'))

关键细节说明 #

• 块大小选择：block_size可根据实际情况调整，64KB~1MB是比较均衡的选择，过大占内存、过小增加IO次数；
• 剩余字节处理：保留最后n-1个字节是因为，只有连续可打印字符长度≥n才会被匹配，这样能保证跨块的连续序列不会被截断；
• 匹配一致性：这种方式的匹配结果和全量读取文件时完全一致，不会漏掉跨块的有效字符串；
• 内存控制：每次仅处理一个块加最多n-1字节的剩余数据，内存占用始终可控，不会因文件过大崩溃。

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