如何用Python流式逐行处理超大.7z文件（内存友好）

核心思路是借助libarchive的分块解压能力，配合缓冲区处理截断的行，实现低内存开销的逐行读取：

• 用libarchive分块读取解压后的二进制数据，避免一次性加载整个文件
• 维护一个缓冲区存储上一次分块中未完成的行（分块可能刚好截断一行）
• 每次读取新块后，拼接缓冲区与当前块，按换行符拆分出完整行进行处理，剩余不完整的行留到下一轮处理

代码实现 #

import libarchive

def process_large_7z_line_by_line(archive_path, target_file):
    leftover_buffer = b''
    # 打开7z文件并创建读取器
    with libarchive.memory_reader(open(archive_path, 'rb')) as archive_reader:
        for entry in archive_reader:
            # 筛选出要处理的目标文件（如果7z里只有一个文件可跳过此判断）
            if entry.pathname == target_file:
                # 分块读取解压后的内容
                for chunk in entry.get_blocks():
                    leftover_buffer += chunk
                    # 按换行符拆分，保留换行符以区分完整行与截断行
                    split_lines = leftover_buffer.splitlines(True)
                    # 处理所有完整行（除了最后一行，可能是截断的）
                    for line in split_lines[:-1]:
                        # 根据文件实际编码调整decode参数，比如gbk、utf-16
                        processed_line = line.decode('utf-8').rstrip('\r\n')
                        # 这里替换成你的行处理逻辑
                        print(processed_line)
                    # 更新缓冲区为最后一行（可能不完整）
                    leftover_buffer = split_lines[-1] if split_lines else b''
        # 处理最后剩余的不完整行（如果存在）
        if leftover_buffer:
            final_line = leftover_buffer.decode('utf-8').rstrip('\r\n')
            # 处理最后一行
            print(final_line)

关键细节说明 #

• 目标文件筛选：如果7z压缩包包含多个文件，必须通过entry.pathname指定要处理的文件；若只有单个文件，可直接去掉该判断
• 编码适配：根据实际文件的编码修改decode的参数，遇到编码错误时可添加errors='replace'或errors='ignore'跳过异常
• 内存控制：缓冲区仅保留未完成的行，内存占用始终保持在几KB到几十KB级别，不会随文件大小增长
• 换行符兼容：splitlines(True)会自动处理\n、\r\n等不同换行格式，确保拆分逻辑兼容各类文本文件

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