@@ -36,7 +36,8 @@ class DecompressReader(io.RawIOBase):
3636 def readable (self ):
3737 return True
3838
39- def __init__ (self , fp , decomp_factory , trailing_error = (), ** decomp_args ):
39+ def __init__ (self , fp , decomp_factory , trailing_error = (), * ,
40+ stream_magic = (), ** decomp_args ):
4041 self ._fp = fp
4142 self ._eof = False
4243 self ._pos = 0 # Current offset in decompressed stream
@@ -56,6 +57,11 @@ def __init__(self, fp, decomp_factory, trailing_error=(), **decomp_args):
5657 # trailing data to ignore
5758 self ._trailing_error = trailing_error
5859
60+ # Prefix (or tuple of prefixes) that every compressed stream starts
61+ # with, if the format has one. Data starting with it is a damaged
62+ # stream rather than trailing data, so its error is not ignored.
63+ self ._stream_magic = stream_magic
64+
5965 def close (self ):
6066 self ._decompressor = None
6167 return super ().close ()
@@ -90,6 +96,9 @@ def read(self, size=-1):
9096 try :
9197 data = self ._decompressor .decompress (rawblock , size )
9298 except self ._trailing_error :
99+ if (self ._stream_magic
100+ and rawblock .startswith (self ._stream_magic )):
101+ raise # Damaged stream, not trailing data.
93102 # Trailing data isn't a valid compressed stream; ignore it.
94103 break
95104 else :
@@ -160,3 +169,60 @@ def seek(self, offset, whence=io.SEEK_SET):
160169 def tell (self ):
161170 """Return the current file position."""
162171 return self ._pos
172+
173+
174+ def _decompress_all (data , decomp_factory , trailing_error = (), * ,
175+ stream_magic = (), truncated_error , ** decomp_args ):
176+ """Decompress all of the concatenated streams in *data*.
177+
178+ Each stream is decompressed by a new decompressor, made by calling
179+ *decomp_factory* with *decomp_args*. As in DecompressReader, an error
180+ of class *trailing_error* in a stream after the first one ends the
181+ data, unless that stream starts with *stream_magic*. A stream without
182+ an end-of-stream marker raises *truncated_error*.
183+ """
184+ results = []
185+ # The views are released on exit, so that the caller can resize a
186+ # bytearray argument again, also while handling an exception.
187+ with memoryview (data ) as view , view .cast ("B" ) as data :
188+ pos = 0
189+ # There is usually a single stream: the first chunk is all of the
190+ # data, to decompress it in one go.
191+ chunk_size = len (data )
192+ while True :
193+ decomp = decomp_factory (** decomp_args )
194+ start = pos
195+ res = []
196+ try :
197+ # Feed the stream in chunks of increasing size, so that what
198+ # is read past its end (and copied into unused_data) is
199+ # bounded by the size of the stream, not by the size of all
200+ # the streams that follow it.
201+ while not decomp .eof and pos < len (data ):
202+ with data [pos :pos + chunk_size ] as chunk :
203+ res .append (decomp .decompress (chunk ))
204+ pos += len (chunk )
205+ chunk_size *= 2
206+ except trailing_error :
207+ if not results :
208+ raise # Error in the first stream; bail out.
209+ if stream_magic :
210+ if isinstance (stream_magic , bytes ):
211+ stream_magic = (stream_magic ,)
212+ size = max (map (len , stream_magic ))
213+ with data [start :start + size ] as head :
214+ if head .tobytes ().startswith (stream_magic ):
215+ raise # Damaged stream, not trailing data.
216+ # Trailing data isn't a valid compressed stream; ignore it.
217+ break
218+ results .extend (res )
219+ if not decomp .eof :
220+ raise truncated_error ("Compressed data ended before the "
221+ "end-of-stream marker was reached" )
222+ pos -= len (decomp .unused_data )
223+ if pos == len (data ):
224+ break
225+ # Later streams start small: a larger first chunk reads further
226+ # past the end of a short stream, a smaller one needs more calls.
227+ chunk_size = 1024
228+ return b"" .join (results )
0 commit comments