如何從使用7z壓縮的文本文件讀取？

Question

我想從csv（文本）文件中讀取（在Python 2.7中），它是7z壓縮的。我不想對整個（大）文件進行解壓縮，而是對這些行進行流式處理。

我試過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 

Answer 1

這將讓你迭代線。它部分來源於我在answer中發現的一些代碼。

據我所知，目前py7zlib沒有提供API，允許將檔案成員讀爲字節流或字符流 - 它的ArchiveFile類只提供了一個解壓縮的read()函數並返回包含該成員的所有未壓縮數據。考慮到這一點，你可以做的最好的事情是迭代地使用它作爲緩衝區來返回字節或行。下面這樣做，但許多沒有幫助，如果問題是檔案成員文件本身是巨大的。

我修改了下面的代碼，可以在Python 2.7和3.x中使用。

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): 
     """ Iterator of lines from an archive member. """ 
     if name not in self: 
      raise SevenZFileError('archive member %r not found in %r' % 
            (name, self.filepath)) 

     for line in io.StringIO(self.archive.getmember(name).read().decode()): 
      yield line 

使用範例：

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)) 

Answer 2

如果你使用Python 3.3+，您可能能夠做到這一點使用lzma模塊這是添加到該版本的標準庫中。

請參見：lzmaExamples