如何读取使用7z压缩的文本文件？

Question

我想逐行读取一个7z压缩的csv（文本）文件（在Python 2.7中）。 我不想解压整个（大）文件，而是要流式传输行。

我尝试过pylzma.decompressobj()但失败了。我收到了数据错误的提示。请注意，此代码尚未逐行读取：

input_filename = r"testing.csv.7z"
with open(input_filename, 'rb') as infile:
    obj = pylzma.decompressobj()
    o = open('decompressed.raw', 'wb')
    obj = pylzma.decompressobj()
    while True:
        tmp = infile.read(1)
        if not tmp: break
        o.write(obj.decompress(tmp))
    o.close()

输出：

    o.write(obj.decompress(tmp))
ValueError: data error during decompression

- Yariv

---

你为什么不把你的代码和一个示例文件发布出来，这样我们就可以重现你的错误，并且看看我们如何帮助你呢？ - Martijn Pieters

.7z文件是可以包含多个文件的容器（存档），那么你想读取testing.7z内部的哪个文件呢？ - martineau

@martineau，testing.csv - Yariv

Answer 1

这将允许您迭代行。它部分源自我在另一个问题的答案中找到的一些代码。

目前（pylzma-0.5.0），py7zlib模块没有实现API，允许以字节或字符流的形式读取存档成员 - 其 ArchiveFile 类仅提供了一个read()函数，一次性解压缩并返回成员中的未压缩数据。鉴于此，最好的做法是使用Python生成器通过缓冲区逐个返回字节或行。

以下代码执行后者，但如果问题是存档成员文件本身太大，则可能无济于事。

以下代码应该适用于Python 3.x和2.7。

import io
import os
import py7zlib


class SevenZFileError(py7zlib.ArchiveError):
    pass

class SevenZFile(object):
    @classmethod
    def is_7zfile(cls, filepath):
        """ Determine if filepath points to a valid 7z archive. """
        is7z = False
        fp = None
        try:
            fp = open(filepath, 'rb')
            archive = py7zlib.Archive7z(fp)
            _ = len(archive.getnames())
            is7z = True
        finally:
            if fp: fp.close()
        return is7z

    def __init__(self, filepath):
        fp = open(filepath, 'rb')
        self.filepath = filepath
        self.archive = py7zlib.Archive7z(fp)

    def __contains__(self, name):
        return name in self.archive.getnames()

    def readlines(self, name, newline=''):
        r""" Iterator of lines from named archive member.

        `newline` controls how line endings are handled.

        It can be None, '', '\n', '\r', and '\r\n' and works the same way as it does
        in StringIO. Note however that the default value is different and is to enable
        universal newlines mode, but line endings are returned untranslated.
        """
        archivefile = self.archive.getmember(name)
        if not archivefile:
            raise SevenZFileError('archive member %r not found in %r' %
                                  (name, self.filepath))

        # Decompress entire member and return its contents iteratively.
        data = archivefile.read().decode()
        for line in io.StringIO(data, newline=newline):
            yield line


if __name__ == '__main__':

    import csv

    if SevenZFile.is_7zfile('testing.csv.7z'):
        sevenZfile = SevenZFile('testing.csv.7z')

        if 'testing.csv' not in sevenZfile:
            print('testing.csv is not a member of testing.csv.7z')
        else:
            reader = csv.reader(sevenZfile.readlines('testing.csv'))
            for row in reader:
                print(', '.join(row))

- martineau

Answer 2

如果您正在使用Python 3.3+，您可能可以使用lzma模块来完成此操作，该模块在该版本的标准库中添加。

请参见：lzma示例。

- blakev

---

问题被标记为python-2.7，因此我们可以假设这里不是使用的Python 3。 - Martijn Pieters

此外，您应该提到Python 3.3（来自文档链接），而不仅仅是3。 - shad0w_wa1k3r

当我发表评论时，@MartijnPieters没有那个标签。 - blakev

即使OP使用的是Python 3.3+，lzma模块仅提供使用LZMA压缩算法压缩和解压缩原始数据的功能，这与处理可能包含多个文件/成员的7zip格式归档文件不同，而PyLZMA第三方模块可以实现。 - martineau

Answer 3

如果您使用Python 3，有一个有用的库py7zr，支持以下部分7zip解压缩：

import py7zr
import re
filter_pattern = re.compile(r'<your/target/file_and_directories/regex/expression>')
with SevenZipFile('archive.7z', 'r') as archive:
    allfiles = archive.getnames()
    selective_files = [f if filter_pattern.match(f) for f in allfiles]
    archive.extract(targets=selective_files)

- Zhou Hongbo

---

这并没有实现OP所要求的，即流式传输单个输出文件。 - Jonathon Reinhart