Pisilinux3/pisi
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

* improve zipext:

Browse Source 
- refactor to eliminate redundancy. better oo now
  - add Bzip2 *read* support only, nothing writes in APPNOTE so I just called it straight :)
This commit is contained in:
Eray Özkural
2006-04-20 13:53:36 +00:00
parent 6a033f1010
commit c3f0a3b203
1 changed files with 62 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pisi/zipfileext.py
+62 -78
View File
@@ -9,10 +9,11 @@
#
# Please read the COPYING file.
#
# Authors: Eray Ozkural <eray@pardus.org.tr>
# Faik Uygur <faik@pardus.org.tr>
# Authors: Eray Ozkural <eray at pardus.org.tr>
# Faik Uygur <faik at pardus.org.tr>
#
# Extends zipfile module with lzma and bzip2 support
"""Extends zipfile module with lzma and bzip2 support"""
# python standard library modules
import os
@@ -28,9 +29,9 @@ except ImportError:
zlib = None
try:
import bzip2
import bz2
except ImportError:
bzip2 = None
bz2 = None
try:
import pylzma
@@ -42,7 +43,8 @@ ZIP_LZMA = 255
compression_methods = [ZIP_STORED, ZIP_DEFLATED, ZIP_BZIP2, ZIP_LZMA]
class FileEntry:
class ZipFileEntry:
"""File-like object used to access entries in a ZipFile"""
def __init__(self, fp, length):
@@ -51,25 +53,18 @@ class FileEntry:
self.returnedBytes = 0
self.length = length
self.finished = 0
class LzmaFileEntry(FileEntry):
"""File-like object used to read a LZMA entry in a ZipFile"""
def __init__(self, fp, length):
FileEntry.__init__(self, fp, length)
self.decomp = pylzma.decompressobj()
self.buffer = ""
def tell(self):
return self.returnedBytes
def read(self, n=None):
if self.finished:
return ""
if n is None:
result = [self.buffer,]
result.append(self.decomp.decompress(self.fp.read(self.length - self.readBytes)))
result.append(self.decomp.flush())
compr_data = self.fp.read(self.length - self.readBytes)
result.append(self.decompress(compr_data))
self.buffer = ""
self.finished = 1
result = "".join(result)
@@ -77,30 +72,64 @@ class LzmaFileEntry(FileEntry):
return result
else:
while len(self.buffer) < n:
data = self.fp.read(min(n, 1024, self.length - self.readBytes))
self.readBytes += len(data)
compr_data = self.fp.read(min(n, 8192, self.length - self.readBytes))
self.readBytes += len(compr_data)
if not data:
result = self.buffer + self.decomp.decompress() + self.decomp.flush()
result = self.buffer # + self.flush_decompressor()
self.finished = 1
self.buffer = ""
self.returnedBytes += len(result)
return result
else:
self.buffer += self.decomp.decompress(data)
self.buffer += self.decompress(data)
result = self.buffer[:n]
self.buffer = self.buffer[n:]
self.returnedBytes += len(result)
return result
def close(self):
self.finished = 1
del self.fp
class ZipFileEntry(FileEntry):
class DeflatedZipFileEntry(ZipFileEntry):
"""File-like object used to read a deflated entry in a ZipFile"""
def __init__(self, fp, length):
ZipFileEntry.__init__(self, fp, length)
self.decomp = zlib.decompressobj(-15)
def decompress(self, compr_data):
return self.decomp.decompress(compr_data) + self.decomp.decompress("Z") + self.decomp.flush()
class Bzip2ZipFileEntry(ZipFileEntry):
"""File-like object used to read a deflated entry in a ZipFile"""
def __init__(self, fp, length):
ZipFileEntry.__init__(self, fp, length)
self.decomp = bz2.Decompressor()
def decompress(self, compr_data):
return self.decomp.decompress(compr_data)
class LzmaZipFileEntry(ZipFileEntry):
"""File-like object used to read a LZMA entry in a ZipFile"""
def __init__(self, fp, length):
ZipFileEntry.__init__(self, fp, length)
self.decompressor = pylzma.decompressobj()
def decompress(self, compr_data):
return self.decomp.decompress(compr_data) + self.decomp(flush)
class StoredZipFileEntry(ZipFileEntry):
"""File-like object used to read an uncompressed entry in a ZipFile"""
def __init__(self, fp, length):
FileEntry.__init__(self, fp, length)
ZipFileEntry.__init__(self, fp, length)
def tell(self):
return self.readBytes
@@ -122,51 +151,6 @@ class ZipFileEntry(FileEntry):
del self.fp
class DeflatedZipFileEntry(FileEntry):
"""File-like object used to read a deflated entry in a ZipFile"""
def __init__(self, fp, length):
FileEntry.__init__(self, fp, length)
self.decomp = zlib.decompressobj(-15)
self.buffer = ""
def tell(self):
return self.returnedBytes
def read(self, n=None):
if self.finished:
return ""
if n is None:
result = [self.buffer,]
result.append(self.decomp.decompress(self.fp.read(self.length - self.readBytes)))
result.append(self.decomp.decompress("Z"))
result.append(self.decomp.flush())
self.buffer = ""
self.finished = 1
result = "".join(result)
self.returnedBytes += len(result)
return result
else:
while len(self.buffer) < n:
data = self.fp.read(min(n, 1024, self.length - self.readBytes))
self.readBytes += len(data)
if not data:
result = self.buffer + self.decomp.decompress("Z") + self.decomp.flush()
self.finished = 1
self.buffer = ""
self.returnedBytes += len(result)
return result
else:
self.buffer += self.decomp.decompress(data)
result = self.buffer[:n]
self.buffer = self.buffer[n:]
self.returnedBytes += len(result)
return result
def close(self):
self.finished = 1
del self.fp
class ZipFileExt(ZipFile):
def _writecheck(self, zinfo):
@@ -185,9 +169,9 @@ class ZipFileExt(ZipFile):
if zinfo.compress_type == ZIP_DEFLATED and not zlib:
raise RuntimeError, \
"Compression requires the (missing) zlib module"
if zinfo.compress_type == ZIP_BZIP2 and not bzip2:
if zinfo.compress_type == ZIP_BZIP2 and not bz2:
raise RuntimeError, \
"Compression requires the (missing) bzip2 module"
"Compression requires the (missing) bz2 module"
if zinfo.compress_type == ZIP_LZMA and not pylzma:
raise RuntimeError, \
"Compression requires the (missing) pylzma module"
@@ -277,22 +261,22 @@ class ZipFileExt(ZipFile):
zinfo = self.getinfo(name)
self.fp.seek(zinfo.file_offset, 0)
if zinfo.compress_type == ZIP_STORED:
return ZipFileEntry(self.fp, zinfo.compress_size)
return StoredZipFileEntry(self.fp, zinfo.compress_size)
elif zinfo.compress_type == ZIP_DEFLATED:
if not zlib:
raise RuntimeError, \
"De-compression requires the (missing) zlib module"
return DeflatedZipFileEntry(self.fp, zinfo.compress_size)
## elif zinfo.compress_type == ZIP_BZIP2:
## if not bzip2:
## raise RuntimeError, \
## "De-compression requires the (missing) bzip2 module"
## return LZMAFileEntry(self.fp, zinfo.compress_size)
elif zinfo.compress_type == ZIP_BZIP2:
if not bzip2:
raise RuntimeError, \
"De-compression requires the (missing) bzip2 module"
return Bzip2ZipFileEntry(self.fp, zinfo.compress_size)
elif zinfo.compress_type == ZIP_LZMA:
if not pylzma:
raise RuntimeError, \
"De-compression requires the (missing) pylzma module"
return LzmaFileEntry(self.fp, zinfo.compress_size)
return LzmaZipFileEntry(self.fp, zinfo.compress_size)
else:
raise BadZipfile, \
"Unsupported compression method %d for file %s" % \