我想(在Python 2.7中)逐行读取一个经过7z压缩的csv(文本)文件。我不想解压缩整个(大)文件,而是流式传输行。
我尝试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
这将允许您迭代行。它部分源于我在回答另一个问题时发现的一些代码。
此时此刻(pylzma-0.5.0
),py7zlib
模块未实现API,该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 'rn' 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))
如果您使用的是Python 3.3+,您可能可以使用该版本中添加到标准库中的lzma
模块来完成此操作。
请参阅:lzma
示例
如果你可以使用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)