PNG IDAT 数据解压缩指南：解决zlib流不完整错误

import zlib

# 模拟从PNG文件中提取出的IDAT块数据
# 实际应用中，这些数据会从PNG文件中读取
# 为了演示，我们使用一个简单的压缩数据流并分割它
original_data = b"This is some sample data that will be compressed and then split into multiple IDAT chunks."
compressed_stream = zlib.compress(original_data)

# 模拟将一个完整的压缩流分割成多个IDAT块
# 实际的PNG文件可能会有不同大小的IDAT块
idat_chunks_data = [
    compressed_stream[0:len(compressed_stream)//3],
    compressed_stream[len(compressed_stream)//3 : 2*len(compressed_stream)//3],
    compressed_stream[2*len(compressed_stream)//3:]
]

print(f"模拟IDAT块数量: {len(idat_chunks_data)}")
for i, chunk in enumerate(idat_chunks_data):
    print(f"IDAT块 {i+1} 大小: {len(chunk)} 字节")

# 1. 拼接所有IDAT块的数据
concatenated_idat_data = b''
for chunk_data in idat_chunks_data:
    concatenated_idat_data += chunk_data

print(f"\n拼接后的IDAT数据总大小: {len(concatenated_idat_data)} 字节")

# 2. 使用zlib.decompress()一次性解压
try:
    decompressed_data = zlib.decompress(concatenated_idat_data)
    print(f"解压成功！原始数据: {decompressed_data.decode('utf-8')}")
    assert decompressed_data == original_data
except zlib.error as e:
    print(f"解压失败: {e}")

import zlib

# 模拟从PNG文件中提取出的IDAT块数据 (同上)
original_data = b"This is some sample data that will be compressed and then split into multiple IDAT chunks."
compressed_stream = zlib.compress(original_data)

idat_chunks_data = [
    compressed_stream[0:len(compressed_stream)//3],
    compressed_stream[len(compressed_stream)//3 : 2*len(compressed_stream)//3],
    compressed_stream[2*len(compressed_stream)//3:]
]

print(f"模拟IDAT块数量: {len(idat_chunks_data)}")

# 1. 创建一个decompressobj实例
decompressor = zlib.decompressobj()
all_decompressed_parts = []

# 2. 遍历每个IDAT块数据并增量解压
for i, chunk_data in enumerate(idat_chunks_data):
    try:
        # decompress() 方法返回当前已解压的数据
        decompressed_part = decompressor.decompress(chunk_data)
        all_decompressed_parts.append(decompressed_part)
        print(f"处理IDAT块 {i+1}，解压出 {len(decompressed_part)} 字节")
    except zlib.error as e:
        print(f"处理IDAT块 {i+1} 时发生错误: {e}")
        break

# 3. 处理完所有IDAT块后，调用flush()获取剩余数据
try:
    remaining_decompressed = decompressor.flush()
    all_decompressed_parts.append(remaining_decompressed)
    print(f"flush() 操作解压出 {len(remaining_decompressed)} 字节")

    final_decompressed_data = b''.join(all_decompressed_parts)
    print(f"\n增量解压成功！原始数据: {final_decompressed_data.decode('utf-8')}")
    assert final_decompressed_data == original_data
except zlib.error as e:
    print(f"flush() 操作失败: {e}")