def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
python类decompressobj()的实例源码
def __init__(self, stream):
"""
Encapsulate the given stream in an unzipper.
"""
# Keep the stream
self.stream = stream
# Make a decompressor with the accept a gzip header flag.
# See <http://stackoverflow.com/a/22311297/402891>.
# Will get None'd out after we're done decompressing and it is flushed.
self.decompressor = zlib.decompressobj(zlib.MAX_WBITS + 16)
# Are we really compressed? This gets set to True if we successfully
# read the first block, and False if we failed to read the first block.
# If it's False, there's no need to try to decompress any more blocks.
self.header_read = None
# We need to do lines ourselves, so we need to keep a buffer
self.buffer = ""
# Track out throughput
self.compressed_bytes = 0
self.uncompressed_bytes = 0
def getDanmu(cid):
if not cid:
return "?????"
try:
cid_url = "http://comment.bilibili.com/%s.xml" % cid
danmu_xml = urllib.request.urlopen(cid_url).read()
xml = zlib.decompressobj(-zlib.MAX_WBITS).decompress(danmu_xml).decode() # ????????
return xml # ?????
except Exception:
pass
# ??xml??????????????
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def content(self):
"""Raw content of response (i.e. bytes).
:returns: Body of HTTP response
:rtype: :class:`str`
"""
if not self._content:
# Decompress gzipped content
if self._gzipped:
decoder = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._content = decoder.decompress(self.raw.read())
else:
self._content = self.raw.read()
self._content_loaded = True
return self._content
def content(self):
"""Raw content of response (i.e. bytes).
:returns: Body of HTTP response
:rtype: :class:`str`
"""
if not self._content:
# Decompress gzipped content
if self._gzipped:
decoder = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._content = decoder.decompress(self.raw.read())
else:
self._content = self.raw.read()
self._content_loaded = True
return self._content
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def content(self):
"""Raw content of response (i.e. bytes).
:returns: Body of HTTP response
:rtype: str
"""
if not self._content:
# Decompress gzipped content
if self._gzipped:
decoder = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._content = decoder.decompress(self.raw.read())
else:
self._content = self.raw.read()
self._content_loaded = True
return self._content
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def content(self):
"""Raw content of response (i.e. bytes).
:returns: Body of HTTP response
:rtype: :class:`str`
"""
if not self._content:
# Decompress gzipped content
if self._gzipped:
decoder = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._content = decoder.decompress(self.raw.read())
else:
self._content = self.raw.read()
self._content_loaded = True
return self._content
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def content(self):
"""Raw content of response (i.e. bytes).
:returns: Body of HTTP response
:rtype: :class:`str`
"""
if not self._content:
# Decompress gzipped content
if self._gzipped:
decoder = zlib.decompressobj(16 + zlib.MAX_WBITS)
self._content = decoder.decompress(self.raw.read())
else:
self._content = self.raw.read()
self._content_loaded = True
return self._content
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
def decompress(self, data):
if not data:
return data
if not self._first_try:
return self._obj.decompress(data)
self._data += data
try:
return self._obj.decompress(data)
except zlib.error:
self._first_try = False
self._obj = zlib.decompressobj(-zlib.MAX_WBITS)
try:
return self.decompress(self._data)
finally:
self._data = None
