Python之zlib模块的使用

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib
import binascii

original_data = b‘This is the original text.‘
print(‘源始数据:长度 : {},内容 : {}‘.format(len(original_data), original_data))

#压缩数据
compressed_data = zlib.compress(original_data)
print(‘压缩的数据:长度 : {},内容 : {}‘.format(len(compressed_data), binascii.hexlify(compressed_data))) #binascii.hexlify主要作用是将字节类转为16进制显示

#解压数据
decompress_data = zlib.decompress(compressed_data)
print(‘压缩的数据:长度 : {},内容 : {}‘.format(len(decompress_data), decompress_data))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib
import binascii

original_data = b‘This is the original text.‘

template = ‘{:>15}  {:>15}‘
print(template.format(‘原始长度‘, ‘压缩长度‘))
print(template.format(‘-‘ * 25, ‘-‘ * 25))

for i in range(5):
    data = original_data * i #数据倍增
    compressed = zlib.compress(data) #压缩数据
    highlight = ‘*‘ if len(data) < len(compressed) else ‘‘ #三目运算法，如果原始数据长度小于压缩的长度就显示*
    print(template.format(len(data), len(compressed)), highlight)

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib
import binascii

original_data = b‘This is the original text.‘ * 1024

template = ‘{:>15}  {:>15}‘
print(template.format(‘压缩级别‘, ‘压缩大小‘))
print(template.format(‘-‘ * 25, ‘-‘ * 25))

for i in range(0, 10):
    data = zlib.compress(original_data, i)  # 设置压缩级别进行压缩
    print(template.format(i, len(data)))

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib
import binascii

compressor = zlib.compressobj(1)

with open(‘content.txt‘, ‘rb‘) as input:
    while True:
        block = input.read(64)  # 每次读取64个字节
        if not block:
            break
        compressed = compressor.compress(block)
        if compressed:
            print(‘压缩数据: {}‘.format(
                binascii.hexlify(compressed)))
        else:
            print(‘数据缓存中...‘)
    remaining = compressor.flush()  # 刷新返回压缩的数据
    print(‘Flushed: {}‘.format(binascii.hexlify(remaining)))

#一次性解压数据，需要注意的是增量压缩，默认会把zlib压缩的头部信息去除，所以解压时需要带上789c
zlib_head = binascii.unhexlify(‘789c‘)
decompress_data = zlib.decompress(zlib_head + remaining)
print(decompress_data)

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib

lorem = open(‘zlib_mixed.py‘, ‘rb‘).read()
compressed = zlib.compress(lorem)

# 压缩数据和没有压缩拼接在一起
combined = compressed + lorem

# 创建一个压缩对象
decompressor = zlib.decompressobj()
decompressed = decompressor.decompress(combined)  # 这里只解压压缩的数据

decompressed_matches = decompressed == lorem
print(‘解压数据的匹配:‘, decompressed_matches)

unused_matches = decompressor.unused_data == lorem
print(‘使用不解压数据的匹配 :‘, unused_matches)

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib

data = open(‘test.py‘, ‘rb‘).read()

cksum = zlib.adler32(data)
print(‘Adler32: {:12d}‘.format(cksum))
print(‘       : {:12d}‘.format(zlib.adler32(data, cksum)))

cksum = zlib.crc32(data)
print(‘CRC-32 : {:12d}‘.format(cksum))
print(‘       : {:12d}‘.format(zlib.crc32(data, cksum)))

Lorem ipsum dolor sit amet, consectetuer adipiscing elit. Donec
egestas, enim et consectetuer ullamcorper, lectus ligula rutrum leo, a
elementum elit tortor eu quam. Duis tincidunt nisi ut ante. Nulla
facilisi. Sed tristique eros eu libero. Pellentesque vel arcu. Vivamus
purus orci, iaculis ac, suscipit sit amet, pulvinar eu,
lacus.

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import socket
import logging
from io import BytesIO
import binascii

# 每次读取的块大小
import zlib

BLOCK_SIZE = 64

if __name__ == ‘__main__‘:
    logging.basicConfig(
        level=logging.DEBUG,
        format=‘%(name)s : %(message)s‘
    )

    logger = logging.getLogger(‘Client‘)

    ip_port = (‘127.0.0.1‘, 8000)
    logging.info(‘开始连接服务器:{}‘.format(ip_port[0] + ‘:‘ + str(ip_port[1])))
    # 创建socket对象
    sk = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)

    # 连接服务器
    sk.connect(ip_port)

    # 服务端需要读取的文件名
    request_file = ‘content.txt‘
    logging.debug(‘发送文件名:{}‘.format(request_file))
    sk.send(request_file.encode(‘utf-8‘))

    # 接收服务端数据
    buffer = BytesIO()

    # 创建一个解压对象
    decompressor = zlib.decompressobj()

    while True:
        response = sk.recv(BLOCK_SIZE)
        if not response:
            break
        logger.debug(‘从服务端读取数据:{}‘.format(binascii.hexlify(response)))
        to_decompress = decompressor.unconsumed_tail + response
        while to_decompress:
            decompressed = decompressor.decompress(to_decompress)
            if decompressed:
                logger.debug(‘解压数据:{}‘.format(decompressed))
                buffer.write(decompressed)
                to_decompress = decompressor.unconsumed_tail
            else:
                logger.debug(‘缓存中...‘)
                to_decompress = None

    remainder = decompressor.flush()
    if remainder:
        logger.debug(‘刷新数据 {}‘.format(remainder))
        buffer.write(remainder)

    # 获取所有的解压数据
    full_reponse = buffer.getvalue()
    read_file = open(request_file, ‘rb‘).read()
    logger.debug(‘服务器传过来的文件与客户端读取的文件是否相等 : {}‘.format(full_reponse == read_file))
    sk.close()

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import zlib
import socketserver
import logging
import binascii

# 每次读取的块大小
BLOCK_SIZE = 64

class ZlibRquestHandler(socketserver.BaseRequestHandler):
    logger = logging.getLogger(‘Server‘)

    def handle(self):
        # 创建一个压缩的对象
        compressor = zlib.compressobj(1)
        # 接收客户端传来的文件名
        filename = self.request.recv(1024)

        self.logger.debug(‘接收客户端数据，文件名 {}‘.format(filename))

        with open(filename, ‘rb‘) as rf:
            while True:
                block = rf.read(BLOCK_SIZE)
                if not block:
                    break
                self.logger.debug(‘读取文件内容:{}‘.format(block))

                # 压缩数据
                compressed = compressor.compress(block)
                if compressed:
                    self.logger.debug(‘发送的十六进制:{}‘.format(binascii.hexlify(compressed)))
                    self.request.send(compressed)
                else:
                    self.logger.debug(‘缓存中...‘)

            # 获取压缩缓存剩下的数据
            remaining = compressor.flush()

            while remaining:  # 循环结束条件，就是刷新压缩缓存的数据，直到空为止
                to_send = remaining[:BLOCK_SIZE]
                remaining = remaining[BLOCK_SIZE:]
                self.logger.debug(‘刷新缓存数据:{}‘.format(binascii.hexlify(to_send)))
                self.request.send(to_send)
            return

if __name__ == ‘__main__‘:
    logging.basicConfig(
        level=logging.DEBUG,
        format=‘%(name)s : %(message)s‘
    )
    ip_port = (‘127.0.0.1‘, 8000)
    socketserver.TCPServer.allow_reuse_address = True
    server = socketserver.TCPServer(ip_port, ZlibRquestHandler)
    server.serve_forever()