Decompress a gzip compressed string in one shot. Return the decompressed string.
(data)
| 592 | |
| 593 | |
| 594 | def decompress(data): |
| 595 | """Decompress a gzip compressed string in one shot. |
| 596 | Return the decompressed string. |
| 597 | """ |
| 598 | decompressed_members = [] |
| 599 | while True: |
| 600 | fp = io.BytesIO(data) |
| 601 | if _read_gzip_header(fp) is None: |
| 602 | return b"".join(decompressed_members) |
| 603 | # Use a zlib raw deflate compressor |
| 604 | do = zlib.decompressobj(wbits=-zlib.MAX_WBITS) |
| 605 | # Read all the data except the header |
| 606 | decompressed = do.decompress(data[fp.tell():]) |
| 607 | if not do.eof or len(do.unused_data) < 8: |
| 608 | raise EOFError("Compressed file ended before the end-of-stream " |
| 609 | "marker was reached") |
| 610 | crc, length = struct.unpack("<II", do.unused_data[:8]) |
| 611 | if crc != zlib.crc32(decompressed): |
| 612 | raise BadGzipFile("CRC check failed") |
| 613 | if length != (len(decompressed) & 0xffffffff): |
| 614 | raise BadGzipFile("Incorrect length of data produced") |
| 615 | decompressed_members.append(decompressed) |
| 616 | data = do.unused_data[8:].lstrip(b"\x00") |
| 617 | |
| 618 | |
| 619 | def main(): |
nothing calls this directly
no test coverage detected