mirror of
https://github.com/google/brotli.git
synced 2024-11-21 19:20:09 +00:00
Add file opener brotli.open
This commit is contained in:
Jeong YunWon
parent
d9f60101c8
commit
3462440035
@ -1,30 +1,25 @@
|
||||
"""Functions that read and write gzipped files.
|
||||
"""Functions that read and write brotli files.
|
||||
|
||||
The user of the file doesn't have to worry about the compression,
|
||||
but random access is not allowed."""
|
||||
|
||||
# forked from CPython 3.8.1 gzip.py which is
|
||||
# based on Andrew Kuchling's minigzip.py distributed with the zlib module
|
||||
|
||||
import struct, sys, time, os
|
||||
import zlib
|
||||
import os
|
||||
import builtins
|
||||
import io
|
||||
import _compression
|
||||
import brotli
|
||||
|
||||
__all__ = ["BadGzipFile", "GzipFile", "open", "compress", "decompress"]
|
||||
|
||||
FTEXT, FHCRC, FEXTRA, FNAME, FCOMMENT = 1, 2, 4, 8, 16
|
||||
__all__ = ["BrotliFile", "open"]
|
||||
|
||||
READ, WRITE = 1, 2
|
||||
|
||||
_COMPRESS_LEVEL_FAST = 1
|
||||
_COMPRESS_LEVEL_TRADEOFF = 6
|
||||
_COMPRESS_LEVEL_BEST = 9
|
||||
|
||||
|
||||
def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_BEST,
|
||||
def open(filename, mode="rb", quality=11, lgwin=22, lgblock=0,
|
||||
encoding=None, errors=None, newline=None):
|
||||
"""Open a gzip-compressed file in binary or text mode.
|
||||
"""Open a brotli-compressed file in binary or text mode.
|
||||
|
||||
The filename argument can be an actual filename (a str or bytes object), or
|
||||
an existing file object to read from or write to.
|
||||
@ -33,14 +28,13 @@ def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_BEST,
|
||||
binary mode, or "rt", "wt", "xt" or "at" for text mode. The default mode is
|
||||
"rb", and the default compresslevel is 9.
|
||||
|
||||
For binary mode, this function is equivalent to the GzipFile constructor:
|
||||
GzipFile(filename, mode, compresslevel). In this case, the encoding, errors
|
||||
and newline arguments must not be provided.
|
||||
For binary mode, this function is equivalent to the BrotliFile constructor:
|
||||
BrotliFile(filename, mode, compresslevel). In this case, the encoding,
|
||||
errors and newline arguments must not be provided.
|
||||
|
||||
For text mode, a GzipFile object is created, and wrapped in an
|
||||
For text mode, a BrotliFile object is created, and wrapped in an
|
||||
io.TextIOWrapper instance with the specified encoding, error handling
|
||||
behavior, and line ending(s).
|
||||
|
||||
"""
|
||||
if "t" in mode:
|
||||
if "b" in mode:
|
||||
@ -55,9 +49,10 @@ def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_BEST,
|
||||
|
||||
gz_mode = mode.replace("t", "")
|
||||
if isinstance(filename, (str, bytes, os.PathLike)):
|
||||
binary_file = GzipFile(filename, gz_mode, compresslevel)
|
||||
binary_file = BrotliFile(filename, gz_mode, quality, lgwin, lgblock)
|
||||
elif hasattr(filename, "read") or hasattr(filename, "write"):
|
||||
binary_file = GzipFile(None, gz_mode, compresslevel, filename)
|
||||
binary_file = BrotliFile(
|
||||
None, gz_mode, quality, lgwin, lgblock, filename)
|
||||
else:
|
||||
raise TypeError("filename must be a str or bytes object, or a file")
|
||||
|
||||
@ -66,64 +61,13 @@ def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_BEST,
|
||||
else:
|
||||
return binary_file
|
||||
|
||||
def write32u(output, value):
|
||||
# The L format writes the bit pattern correctly whether signed
|
||||
# or unsigned.
|
||||
output.write(struct.pack("<L", value))
|
||||
|
||||
class _PaddedFile:
|
||||
"""Minimal read-only file object that prepends a string to the contents
|
||||
of an actual file. Shouldn't be used outside of gzip.py, as it lacks
|
||||
essential functionality."""
|
||||
|
||||
def __init__(self, f, prepend=b''):
|
||||
self._buffer = prepend
|
||||
self._length = len(prepend)
|
||||
self.file = f
|
||||
self._read = 0
|
||||
|
||||
def read(self, size):
|
||||
if self._read is None:
|
||||
return self.file.read(size)
|
||||
if self._read + size <= self._length:
|
||||
read = self._read
|
||||
self._read += size
|
||||
return self._buffer[read:self._read]
|
||||
else:
|
||||
read = self._read
|
||||
self._read = None
|
||||
return self._buffer[read:] + \
|
||||
self.file.read(size-self._length+read)
|
||||
|
||||
def prepend(self, prepend=b''):
|
||||
if self._read is None:
|
||||
self._buffer = prepend
|
||||
else: # Assume data was read since the last prepend() call
|
||||
self._read -= len(prepend)
|
||||
return
|
||||
self._length = len(self._buffer)
|
||||
self._read = 0
|
||||
|
||||
def seek(self, off):
|
||||
self._read = None
|
||||
self._buffer = None
|
||||
return self.file.seek(off)
|
||||
|
||||
def seekable(self):
|
||||
return True # Allows fast-forwarding even in unseekable streams
|
||||
|
||||
|
||||
class BadGzipFile(OSError):
|
||||
"""Exception raised in some cases for invalid gzip files."""
|
||||
|
||||
|
||||
class GzipFile(_compression.BaseStream):
|
||||
"""The GzipFile class simulates most of the methods of a file object with
|
||||
class BrotliFile(_compression.BaseStream):
|
||||
"""The BrotliFile class simulates most of the methods of a file object with
|
||||
the exception of the truncate() method.
|
||||
|
||||
This class only supports opening files in binary mode. If you need to open a
|
||||
compressed file in text mode, use the gzip.open() function.
|
||||
|
||||
This class only supports opening files in binary mode. If you need to open
|
||||
a compressed file in text mode, use the brotli.open() function.
|
||||
"""
|
||||
|
||||
# Overridden with internal file object to be closed, if only a filename
|
||||
@ -131,8 +75,9 @@ class GzipFile(_compression.BaseStream):
|
||||
myfileobj = None
|
||||
|
||||
def __init__(self, filename=None, mode=None,
|
||||
compresslevel=_COMPRESS_LEVEL_BEST, fileobj=None, mtime=None):
|
||||
"""Constructor for the GzipFile class.
|
||||
quality=11, lgwin=22, lgblock=0,
|
||||
fileobj=None):
|
||||
"""Constructor for the BrotliFile class.
|
||||
|
||||
At least one of fileobj and filename must be given a
|
||||
non-trivial value.
|
||||
@ -142,27 +87,11 @@ class GzipFile(_compression.BaseStream):
|
||||
It defaults to None, in which case filename is opened to provide
|
||||
a file object.
|
||||
|
||||
When fileobj is not None, the filename argument is only used to be
|
||||
included in the gzip file header, which may include the original
|
||||
filename of the uncompressed file. It defaults to the filename of
|
||||
fileobj, if discernible; otherwise, it defaults to the empty string,
|
||||
and in this case the original filename is not included in the header.
|
||||
|
||||
The mode argument can be any of 'r', 'rb', 'a', 'ab', 'w', 'wb', 'x', or
|
||||
'xb' depending on whether the file will be read or written. The default
|
||||
is the mode of fileobj if discernible; otherwise, the default is 'rb'.
|
||||
A mode of 'r' is equivalent to one of 'rb', and similarly for 'w' and
|
||||
'wb', 'a' and 'ab', and 'x' and 'xb'.
|
||||
|
||||
The compresslevel argument is an integer from 0 to 9 controlling the
|
||||
level of compression; 1 is fastest and produces the least compression,
|
||||
and 9 is slowest and produces the most compression. 0 is no compression
|
||||
at all. The default is 9.
|
||||
|
||||
The mtime argument is an optional numeric timestamp to be written
|
||||
to the last modification time field in the stream when compressing.
|
||||
If omitted or None, the current time is used.
|
||||
|
||||
The mode argument can be any of 'r', 'rb', 'a', 'ab', 'w', 'wb', 'x',
|
||||
or 'xb' depending on whether the file will be read or written. The
|
||||
default is the mode of fileobj if discernible; otherwise, the default
|
||||
is 'rb'. A mode of 'r' is equivalent to one of 'rb', and similarly for
|
||||
'w' and 'wb', 'a' and 'ab', and 'x' and 'xb'.
|
||||
"""
|
||||
|
||||
if mode and ('t' in mode or 'U' in mode):
|
||||
@ -182,35 +111,22 @@ class GzipFile(_compression.BaseStream):
|
||||
|
||||
if mode.startswith('r'):
|
||||
self.mode = READ
|
||||
raw = _GzipReader(fileobj)
|
||||
raw = _BrotliReader(fileobj, _BrotliDecompressor)
|
||||
self._buffer = io.BufferedReader(raw)
|
||||
self.name = filename
|
||||
|
||||
elif mode.startswith(('w', 'a', 'x')):
|
||||
self.mode = WRITE
|
||||
self._init_write(filename)
|
||||
self.compress = zlib.compressobj(compresslevel,
|
||||
zlib.DEFLATED,
|
||||
-zlib.MAX_WBITS,
|
||||
zlib.DEF_MEM_LEVEL,
|
||||
0)
|
||||
self._write_mtime = mtime
|
||||
self.size = 0
|
||||
self.offset = 0
|
||||
self.name = filename
|
||||
self.compress = brotli.Compressor(
|
||||
quality=quality, lgwin=lgwin, lgblock=lgblock)
|
||||
else:
|
||||
raise ValueError("Invalid mode: {!r}".format(mode))
|
||||
|
||||
self.fileobj = fileobj
|
||||
|
||||
if self.mode == WRITE:
|
||||
self._write_gzip_header(compresslevel)
|
||||
|
||||
@property
|
||||
def filename(self):
|
||||
import warnings
|
||||
warnings.warn("use the name attribute", DeprecationWarning, 2)
|
||||
if self.mode == WRITE and self.name[-3:] != ".gz":
|
||||
return self.name + ".gz"
|
||||
return self.name
|
||||
|
||||
@property
|
||||
def mtime(self):
|
||||
"""Last modification time read from stream, or None"""
|
||||
@ -218,56 +134,16 @@ class GzipFile(_compression.BaseStream):
|
||||
|
||||
def __repr__(self):
|
||||
s = repr(self.fileobj)
|
||||
return '<gzip ' + s[1:-1] + ' ' + hex(id(self)) + '>'
|
||||
return '<brotli ' + s[1:-1] + ' ' + hex(id(self)) + '>'
|
||||
|
||||
def _init_write(self, filename):
|
||||
self.name = filename
|
||||
self.crc = zlib.crc32(b"")
|
||||
self.size = 0
|
||||
self.writebuf = []
|
||||
self.bufsize = 0
|
||||
self.offset = 0 # Current file offset for seek(), tell(), etc
|
||||
|
||||
def _write_gzip_header(self, compresslevel):
|
||||
self.fileobj.write(b'\037\213') # magic header
|
||||
self.fileobj.write(b'\010') # compression method
|
||||
try:
|
||||
# RFC 1952 requires the FNAME field to be Latin-1. Do not
|
||||
# include filenames that cannot be represented that way.
|
||||
fname = os.path.basename(self.name)
|
||||
if not isinstance(fname, bytes):
|
||||
fname = fname.encode('latin-1')
|
||||
if fname.endswith(b'.gz'):
|
||||
fname = fname[:-3]
|
||||
except UnicodeEncodeError:
|
||||
fname = b''
|
||||
flags = 0
|
||||
if fname:
|
||||
flags = FNAME
|
||||
self.fileobj.write(chr(flags).encode('latin-1'))
|
||||
mtime = self._write_mtime
|
||||
if mtime is None:
|
||||
mtime = time.time()
|
||||
write32u(self.fileobj, int(mtime))
|
||||
if compresslevel == _COMPRESS_LEVEL_BEST:
|
||||
xfl = b'\002'
|
||||
elif compresslevel == _COMPRESS_LEVEL_FAST:
|
||||
xfl = b'\004'
|
||||
else:
|
||||
xfl = b'\000'
|
||||
self.fileobj.write(xfl)
|
||||
self.fileobj.write(b'\377')
|
||||
if fname:
|
||||
self.fileobj.write(fname + b'\000')
|
||||
|
||||
def write(self,data):
|
||||
def write(self, data):
|
||||
self._check_not_closed()
|
||||
if self.mode != WRITE:
|
||||
import errno
|
||||
raise OSError(errno.EBADF, "write() on read-only GzipFile object")
|
||||
raise OSError(errno.EBADF, "write() on read-only BrotliFile object")
|
||||
|
||||
if self.fileobj is None:
|
||||
raise ValueError("write() on closed GzipFile object")
|
||||
raise ValueError("write() on closed BrotliFile object")
|
||||
|
||||
if isinstance(data, bytes):
|
||||
length = len(data)
|
||||
@ -277,9 +153,8 @@ class GzipFile(_compression.BaseStream):
|
||||
length = data.nbytes
|
||||
|
||||
if length > 0:
|
||||
self.fileobj.write(self.compress.compress(data))
|
||||
self.fileobj.write(self.compress.process(data))
|
||||
self.size += length
|
||||
self.crc = zlib.crc32(data, self.crc)
|
||||
self.offset += length
|
||||
|
||||
return length
|
||||
@ -288,7 +163,7 @@ class GzipFile(_compression.BaseStream):
|
||||
self._check_not_closed()
|
||||
if self.mode != READ:
|
||||
import errno
|
||||
raise OSError(errno.EBADF, "read() on write-only GzipFile object")
|
||||
raise OSError(errno.EBADF, "read() on write-only BrotliFile object")
|
||||
return self._buffer.read(size)
|
||||
|
||||
def read1(self, size=-1):
|
||||
@ -298,7 +173,7 @@ class GzipFile(_compression.BaseStream):
|
||||
self._check_not_closed()
|
||||
if self.mode != READ:
|
||||
import errno
|
||||
raise OSError(errno.EBADF, "read1() on write-only GzipFile object")
|
||||
raise OSError(errno.EBADF, "read1() on write-only BrotliFile object")
|
||||
|
||||
if size < 0:
|
||||
size = io.DEFAULT_BUFFER_SIZE
|
||||
@ -308,7 +183,7 @@ class GzipFile(_compression.BaseStream):
|
||||
self._check_not_closed()
|
||||
if self.mode != READ:
|
||||
import errno
|
||||
raise OSError(errno.EBADF, "peek() on write-only GzipFile object")
|
||||
raise OSError(errno.EBADF, "peek() on write-only BrotliFile object")
|
||||
return self._buffer.peek(n)
|
||||
|
||||
@property
|
||||
@ -323,9 +198,7 @@ class GzipFile(_compression.BaseStream):
|
||||
try:
|
||||
if self.mode == WRITE:
|
||||
fileobj.write(self.compress.flush())
|
||||
write32u(fileobj, self.crc)
|
||||
# self.size may exceed 2 GiB, or even 4 GiB
|
||||
write32u(fileobj, self.size & 0xffffffff)
|
||||
fileobj.write(self.compress.finish())
|
||||
elif self.mode == READ:
|
||||
self._buffer.close()
|
||||
finally:
|
||||
@ -334,11 +207,11 @@ class GzipFile(_compression.BaseStream):
|
||||
self.myfileobj = None
|
||||
myfileobj.close()
|
||||
|
||||
def flush(self,zlib_mode=zlib.Z_SYNC_FLUSH):
|
||||
def flush(self):
|
||||
self._check_not_closed()
|
||||
if self.mode == WRITE:
|
||||
# Ensure the compressor's buffer is flushed
|
||||
self.fileobj.write(self.compress.flush(zlib_mode))
|
||||
self.fileobj.write(self.compress.flush())
|
||||
self.fileobj.flush()
|
||||
|
||||
def fileno(self):
|
||||
@ -390,212 +263,60 @@ class GzipFile(_compression.BaseStream):
|
||||
return self._buffer.readline(size)
|
||||
|
||||
|
||||
class _GzipReader(_compression.DecompressReader):
|
||||
def __init__(self, fp):
|
||||
super().__init__(_PaddedFile(fp), zlib.decompressobj,
|
||||
wbits=-zlib.MAX_WBITS)
|
||||
# Set flag indicating start of a new member
|
||||
self._new_member = True
|
||||
self._last_mtime = None
|
||||
class _BrotliDecompressor:
|
||||
eof = False
|
||||
|
||||
def _init_read(self):
|
||||
self._crc = zlib.crc32(b"")
|
||||
self._stream_size = 0 # Decompressed size of unconcatenated stream
|
||||
def __init__(self):
|
||||
self.decompressor = brotli.Decompressor()
|
||||
self.needs_input = True
|
||||
self._buffer = bytearray(1)
|
||||
self._bufview = memoryview(self._buffer)
|
||||
self._buflen = len(self._buffer)
|
||||
self._pos = 0
|
||||
|
||||
def _read_exact(self, n):
|
||||
'''Read exactly *n* bytes from `self._fp`
|
||||
def _check_buffer(self, new_len):
|
||||
if self._buflen < new_len:
|
||||
new_len = max(self._buflen, new_len)
|
||||
del self._bufview
|
||||
self._buffer.extend(b'\0' * (new_len * 2))
|
||||
self._bufview = memoryview(self._buffer)
|
||||
self._buflen = len(self._buffer)
|
||||
|
||||
This method is required because self._fp may be unbuffered,
|
||||
i.e. return short reads.
|
||||
'''
|
||||
|
||||
data = self._fp.read(n)
|
||||
while len(data) < n:
|
||||
b = self._fp.read(n - len(data))
|
||||
if not b:
|
||||
raise EOFError("Compressed file ended before the "
|
||||
"end-of-stream marker was reached")
|
||||
data += b
|
||||
return data
|
||||
|
||||
def _read_gzip_header(self):
|
||||
magic = self._fp.read(2)
|
||||
if magic == b'':
|
||||
return False
|
||||
|
||||
if magic != b'\037\213':
|
||||
raise BadGzipFile('Not a gzipped file (%r)' % magic)
|
||||
|
||||
(method, flag,
|
||||
self._last_mtime) = struct.unpack("<BBIxx", self._read_exact(8))
|
||||
if method != 8:
|
||||
raise BadGzipFile('Unknown compression method')
|
||||
|
||||
if flag & FEXTRA:
|
||||
# Read & discard the extra field, if present
|
||||
extra_len, = struct.unpack("<H", self._read_exact(2))
|
||||
self._read_exact(extra_len)
|
||||
if flag & FNAME:
|
||||
# Read and discard a null-terminated string containing the filename
|
||||
while True:
|
||||
s = self._fp.read(1)
|
||||
if not s or s==b'\000':
|
||||
break
|
||||
if flag & FCOMMENT:
|
||||
# Read and discard a null-terminated string containing a comment
|
||||
while True:
|
||||
s = self._fp.read(1)
|
||||
if not s or s==b'\000':
|
||||
break
|
||||
if flag & FHCRC:
|
||||
self._read_exact(2) # Read & discard the 16-bit header CRC
|
||||
return True
|
||||
|
||||
def read(self, size=-1):
|
||||
if size < 0:
|
||||
return self.readall()
|
||||
# size=0 is special because decompress(max_length=0) is not supported
|
||||
if not size:
|
||||
return b""
|
||||
|
||||
# For certain input data, a single
|
||||
# call to decompress() may not return
|
||||
# any data. In this case, retry until we get some data or reach EOF.
|
||||
while True:
|
||||
if self._decompressor.eof:
|
||||
# Ending case: we've come to the end of a member in the file,
|
||||
# so finish up this member, and read a new gzip header.
|
||||
# Check the CRC and file size, and set the flag so we read
|
||||
# a new member
|
||||
self._read_eof()
|
||||
self._new_member = True
|
||||
self._decompressor = self._decomp_factory(
|
||||
**self._decomp_args)
|
||||
|
||||
if self._new_member:
|
||||
# If the _new_member flag is set, we have to
|
||||
# jump to the next member, if there is one.
|
||||
self._init_read()
|
||||
if not self._read_gzip_header():
|
||||
self._size = self._pos
|
||||
return b""
|
||||
self._new_member = False
|
||||
|
||||
# Read a chunk of data from the file
|
||||
buf = self._fp.read(io.DEFAULT_BUFFER_SIZE)
|
||||
|
||||
uncompress = self._decompressor.decompress(buf, size)
|
||||
if self._decompressor.unconsumed_tail != b"":
|
||||
self._fp.prepend(self._decompressor.unconsumed_tail)
|
||||
elif self._decompressor.unused_data != b"":
|
||||
# Prepend the already read bytes to the fileobj so they can
|
||||
# be seen by _read_eof() and _read_gzip_header()
|
||||
self._fp.prepend(self._decompressor.unused_data)
|
||||
|
||||
if uncompress != b"":
|
||||
break
|
||||
if buf == b"":
|
||||
raise EOFError("Compressed file ended before the "
|
||||
"end-of-stream marker was reached")
|
||||
|
||||
self._add_read_data( uncompress )
|
||||
self._pos += len(uncompress)
|
||||
return uncompress
|
||||
|
||||
def _add_read_data(self, data):
|
||||
self._crc = zlib.crc32(data, self._crc)
|
||||
self._stream_size = self._stream_size + len(data)
|
||||
|
||||
def _read_eof(self):
|
||||
# We've read to the end of the file
|
||||
# We check the that the computed CRC and size of the
|
||||
# uncompressed data matches the stored values. Note that the size
|
||||
# stored is the true file size mod 2**32.
|
||||
crc32, isize = struct.unpack("<II", self._read_exact(8))
|
||||
if crc32 != self._crc:
|
||||
raise BadGzipFile("CRC check failed %s != %s" % (hex(crc32),
|
||||
hex(self._crc)))
|
||||
elif isize != (self._stream_size & 0xffffffff):
|
||||
raise BadGzipFile("Incorrect length of data produced")
|
||||
|
||||
# Gzip files can be padded with zeroes and still have archives.
|
||||
# Consume all zero bytes and set the file position to the first
|
||||
# non-zero byte. See http://www.gzip.org/#faq8
|
||||
c = b"\x00"
|
||||
while c == b"\x00":
|
||||
c = self._fp.read(1)
|
||||
if c:
|
||||
self._fp.prepend(c)
|
||||
|
||||
def _rewind(self):
|
||||
super()._rewind()
|
||||
self._new_member = True
|
||||
|
||||
def compress(data, compresslevel=_COMPRESS_LEVEL_BEST, *, mtime=None):
|
||||
"""Compress data in one shot and return the compressed string.
|
||||
Optional argument is the compression level, in range of 0-9.
|
||||
"""
|
||||
buf = io.BytesIO()
|
||||
with GzipFile(fileobj=buf, mode='wb', compresslevel=compresslevel, mtime=mtime) as f:
|
||||
f.write(data)
|
||||
return buf.getvalue()
|
||||
|
||||
def decompress(data):
|
||||
"""Decompress a gzip compressed string in one shot.
|
||||
Return the decompressed string.
|
||||
"""
|
||||
with GzipFile(fileobj=io.BytesIO(data)) as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
def main():
|
||||
from argparse import ArgumentParser
|
||||
parser = ArgumentParser(description=
|
||||
"A simple command line interface for the gzip module: act like gzip, "
|
||||
"but do not delete the input file.")
|
||||
group = parser.add_mutually_exclusive_group()
|
||||
group.add_argument('--fast', action='store_true', help='compress faster')
|
||||
group.add_argument('--best', action='store_true', help='compress better')
|
||||
group.add_argument("-d", "--decompress", action="store_true",
|
||||
help="act like gunzip instead of gzip")
|
||||
|
||||
parser.add_argument("args", nargs="*", default=["-"], metavar='file')
|
||||
args = parser.parse_args()
|
||||
|
||||
compresslevel = _COMPRESS_LEVEL_TRADEOFF
|
||||
if args.fast:
|
||||
compresslevel = _COMPRESS_LEVEL_FAST
|
||||
elif args.best:
|
||||
compresslevel = _COMPRESS_LEVEL_BEST
|
||||
|
||||
for arg in args.args:
|
||||
if args.decompress:
|
||||
if arg == "-":
|
||||
f = GzipFile(filename="", mode="rb", fileobj=sys.stdin.buffer)
|
||||
g = sys.stdout.buffer
|
||||
else:
|
||||
if arg[-3:] != ".gz":
|
||||
print("filename doesn't end in .gz:", repr(arg))
|
||||
continue
|
||||
f = open(arg, "rb")
|
||||
g = builtins.open(arg[:-3], "wb")
|
||||
def decompress(self, raw, size):
|
||||
if raw:
|
||||
uncompress = self.decompressor.process(raw)
|
||||
new_len = len(uncompress)
|
||||
self.needs_input = False
|
||||
else:
|
||||
if arg == "-":
|
||||
f = sys.stdin.buffer
|
||||
g = GzipFile(filename="", mode="wb", fileobj=sys.stdout.buffer,
|
||||
compresslevel=compresslevel)
|
||||
else:
|
||||
f = builtins.open(arg, "rb")
|
||||
g = open(arg + ".gz", "wb")
|
||||
while True:
|
||||
chunk = f.read(1024)
|
||||
if not chunk:
|
||||
break
|
||||
g.write(chunk)
|
||||
if g is not sys.stdout.buffer:
|
||||
g.close()
|
||||
if f is not sys.stdin.buffer:
|
||||
f.close()
|
||||
uncompress = b''
|
||||
new_len = 0
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
if self._pos >= size:
|
||||
r = bytes(self._bufview[:size])
|
||||
pos = self._pos - size
|
||||
|
||||
self._check_buffer(pos + new_len)
|
||||
self._bufview[:pos] = self._bufview[size:self._pos]
|
||||
self._bufview[pos:pos + new_len] = uncompress
|
||||
self._pos = pos + new_len
|
||||
elif self._pos + new_len >= size:
|
||||
used_len = size - self._pos
|
||||
r = bytes(self._bufview[:self._pos]) + uncompress[:used_len]
|
||||
|
||||
rem_len = new_len - used_len
|
||||
self._check_buffer(rem_len)
|
||||
self._bufview[:rem_len] = uncompress[used_len:]
|
||||
self._pos = rem_len
|
||||
else:
|
||||
r = bytes(self._bufview[:self._pos]) + uncompress
|
||||
self._pos = 0
|
||||
self.needs_input = True
|
||||
return r
|
||||
|
||||
|
||||
class _BrotliReader(_compression.DecompressReader):
|
||||
def read(self, size=-1):
|
||||
try:
|
||||
return super(_BrotliReader, self).read(size)
|
||||
except EOFError:
|
||||
return b''
|
||||
|
||||
68
python/tests/file_test.py
Normal file
68
python/tests/file_test.py
Normal file
@ -0,0 +1,68 @@
|
||||
# Copyright 2016 The Brotli Authors. All rights reserved.
|
||||
#
|
||||
# Distributed under MIT license.
|
||||
# See file LICENSE for detail or copy at https://opensource.org/licenses/MIT
|
||||
|
||||
import unittest
|
||||
|
||||
from . import _test_utils
|
||||
import brotli
|
||||
import brotli_file
|
||||
|
||||
|
||||
class TestCompress(_test_utils.TestCase):
|
||||
|
||||
VARIANTS = {'quality': (1, 6, 9, 11), 'lgwin': (10, 15, 20, 24)}
|
||||
|
||||
def _check_decompression(self, test_data, **kwargs):
|
||||
kwargs = {}
|
||||
# Write decompression to temp file and verify it matches the original.
|
||||
temp_uncompressed = _test_utils.get_temp_uncompressed_name(test_data)
|
||||
temp_compressed = _test_utils.get_temp_compressed_name(test_data)
|
||||
original = test_data
|
||||
with open(temp_uncompressed, 'wb') as out_file:
|
||||
with open(temp_compressed, 'rb') as in_file:
|
||||
out_file.write(brotli.decompress(in_file.read(), **kwargs))
|
||||
self.assertFilesMatch(temp_uncompressed, original)
|
||||
|
||||
def _compress(self, test_data, **kwargs):
|
||||
temp_compressed = _test_utils.get_temp_compressed_name(test_data)
|
||||
with brotli_file.open(temp_compressed, 'w', **kwargs) as out_file:
|
||||
with open(test_data, 'rb') as in_file:
|
||||
out_file.write(in_file.read())
|
||||
|
||||
def _test_compress(self, test_data, **kwargs):
|
||||
self._compress(test_data, **kwargs)
|
||||
self._check_decompression(test_data, **kwargs)
|
||||
|
||||
|
||||
_test_utils.generate_test_methods(TestCompress, variants=TestCompress.VARIANTS)
|
||||
|
||||
|
||||
def _get_original_name(test_data):
|
||||
return test_data.split('.compressed')[0]
|
||||
|
||||
|
||||
class TestDecompress(_test_utils.TestCase):
|
||||
|
||||
def _check_decompression(self, test_data):
|
||||
# Verify decompression matches the original.
|
||||
temp_uncompressed = _test_utils.get_temp_uncompressed_name(test_data)
|
||||
original = _get_original_name(test_data)
|
||||
self.assertFilesMatch(temp_uncompressed, original)
|
||||
|
||||
def _decompress(self, test_data):
|
||||
temp_uncompressed = _test_utils.get_temp_uncompressed_name(test_data)
|
||||
with open(temp_uncompressed, 'wb') as out_file:
|
||||
with brotli_file.open(test_data) as in_file:
|
||||
out_file.write(in_file.read())
|
||||
|
||||
def _test_decompress(self, test_data):
|
||||
self._decompress(test_data)
|
||||
self._check_decompression(test_data)
|
||||
|
||||
|
||||
_test_utils.generate_test_methods(TestDecompress, for_decompression=True)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user