Python之bz2模块的使用

#!/usr/bin/env python3
# encoding: utf-8

import bz2
import binascii

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

compressed = bz2.compress(original_data)
print(‘压缩后的数据源长度 : {} bytes‘.format(len(compressed)))
hex_version = binascii.hexlify(compressed)
print(hex_version)

#这里主要学习是如下，数据的折分方法
for i in range(len(hex_version) // 40 + 1):
    print(hex_version[i * 40:(i + 1) * 40])

#!/usr/bin/env python3
# encoding: utf-8

import bz2

original_data = b‘This is the original text.‘

fmt = ‘{:>15} {:>15}‘
print(fmt.format(‘len(data)‘, ‘len(compressed)‘))
print(fmt.format(‘-‘ * 15, ‘-‘ * 15))
for i in range(5):
    data = original_data * i
    compressed = bz2.compress(data)
    print(fmt.format(len(data), len(compressed)), end=‘‘)#加end表示，结尾不使用回车符
    print(‘*‘ if len(data) < len(compressed) else ‘‘) #使用三目运算

#!/usr/bin/env python3
# encoding: utf-8

import bz2
import binascii
import io

compressor = bz2.BZ2Compressor()

# 打开文件，类型为字节，读模式
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(‘刷新所有的压缩数据:{}‘.format(binascii.hexlify(remaining)))

#!/usr/bin/env python3
# encoding: utf-8

import bz2

content = open(‘content.txt‘, ‘rt‘).read().encode(‘utf-8‘)
compressed = bz2.compress(content)  # 压缩的数据
combined = compressed + content  # 压缩的数据+未压缩的数据

# 创建一个解压的对象
decompressor = bz2.BZ2Decompressor()

# 解压混合的数据
decompressed = decompressor.decompress(combined)

# 判断解压数据与未解压数据是否相等
decompressed_matches = decompressed == content
print(‘解压数据与未解压数据是否相等‘, decompressed_matches)

# 判断解压对象的unused_data（即包含所有未使用的数据）与未解压数据是否相等
unused_matche = decompressor.unused_data == content
print(‘解压对象的unused_data（即包含所有未使用的数据）与未解压数据是否相等‘, unused_matche)

#!/usr/bin/env python3
# encoding: utf-8

import bz2
import io
import os

data = "Content of the example  file go here.\n"

with bz2.BZ2File(‘example.bz2‘,‘wb‘) as output:
    with io.TextIOWrapper(output,encoding=‘utf-8‘) as enc:
        enc.write(data)
os.system(‘file example.bz2‘)

#!/usr/bin/env python3
# encoding: utf-8

import bz2
import io
import os

data = open(‘content.txt‘, ‘r‘, encoding=‘utf-8‘).read() * 1024
print(‘输入 {} 字节‘.format(len(data.encode(‘utf-8‘))))

for i in range(1, 10):
    filename = ‘compress-level-{}.bz2‘.format(i)
    with bz2.BZ2File(filename, ‘wb‘, compresslevel=i) as output:
        with io.TextIOWrapper(output, encoding=‘utf-8‘) as enc:
            enc.write(data)
    os.system(‘cksum {}‘.format(filename))

#!/usr/bin/env python3
# encoding: utf-8

import bz2
import io
import os
import itertools

data = ‘The same line,over and over.\n‘
with bz2.BZ2File(‘lines.bz2‘, ‘wb‘) as output:
    with io.TextIOWrapper(output, encoding=‘utf-8‘) as enc:
        enc.writelines(itertools.repeat(data, 10))

os.system(‘bzcat lines.bz2‘)

#!/usr/bin/env python3
# encoding: utf-8

import bz2
import io

with bz2.BZ2File(‘example.bz2‘, ‘rb‘) as input:
    with io.TextIOWrapper(input, encoding=‘utf-8‘) as enc:
        print(enc.read())

#!/usr/bin/env python3
# encoding: utf-8

import bz2

with bz2.BZ2File(‘example.bz2‘, ‘rb‘) as input:
    print(‘读取压缩文件内容‘)
    all_data = input.read()
    print(‘内容:‘, all_data)

    # 切片取值
    expected = all_data[5:15]

    # 指针指向0
    input.seek(0)

    # 指针移动5位
    input.seek(5)
    print(‘开始相对下标获取5到10个字节‘)
    partial = input.read(10)
    print(‘移动指针后获取的值‘, partial)
    print(‘一次性取值切片获取的值‘, expected)

#!/usr/bin/env python3
# encoding: utf-8

import bz2

data = ‘Character with an åccent.‘

with bz2.open(‘example.bz2‘, ‘wt‘, encoding=‘utf-8‘) as output:
    output.write(data)

with bz2.open(‘example.bz2‘, ‘rt‘, encoding=‘utf-8‘) as input:
    print(‘压缩文件所有的内容:{}‘.format(input.read()))

with bz2.open(‘example.bz2‘, ‘rt‘, encoding=‘utf-8‘) as input:
    input.seek(18)
    print(‘读取第一个字符{}‘.format(input.read(1)))

with bz2.open(‘example.bz2‘, ‘rt‘, encoding=‘utf-8‘) as input:
    input.seek(19)
    try:
        print(‘读取第一个字符{}‘, input.read(1))
    except UnicodeDecodeError:
        print(‘错误：解码失败,因为文件指针是针对字节移动，而不是按字符移动，unicode编码占2个字节‘)

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 bz2
import socketserver
import logging
import binascii

BLOCK_SIZE = 32

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

    def handle(self):
        # 创建一个压缩对象
        compressor = bz2.BZ2Compressor()
        file_name = self.request.recv(1024).decode(‘utf-8‘)
        self.logger.debug(‘接收到客户端文件名：%s‘, file_name)
        with open(file_name, ‘rb‘) as input:
            while True:
                block = input.read(BLOCK_SIZE)
                if not block:
                    break
                self.logger.debug(‘读取到数据：%r‘, block)
                # 压缩数据
                compressed = compressor.compress(block)
                if compressed:
                    self.logger.debug(‘发送中 %r‘, 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(‘FLUSHING %r‘, binascii.hexlify(to_send))
            self.request.send(to_send)
        return


if __name__ == ‘__main__‘:
    address = (‘localhost‘, 8888)
    server = socketserver.TCPServer(address, Bz2RequestHandler)
    ip, port = server.server_address
    server.serve_forever()

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

import socket
import sys
from io import StringIO
import logging
import bz2
import binascii

if __name__ == ‘__main__‘:
    BLOCK_SIZE = 64

    logging.basicConfig(level=logging.DEBUG, format=‘%(name)s : %(message)s‘)
    server_address = (‘localhost‘, 8888)
    logger = logging.getLogger(‘Client‘)
    logging.debug(‘连接服务器: %s:%s‘, server_address[0], server_address[1])
    sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sk.connect(server_address)

    # 输入文件名的参数
    requested_file = (sys.argv[0] if len(sys.argv) > 1 else "content.txt")

    sk.send(requested_file.encode(‘utf-8‘))

    buffer = StringIO()
    decompressor = bz2.BZ2Decompressor()  # 创建解压对像

    while True:
        response = sk.recv(BLOCK_SIZE)
        if not response:
            break
        logging.debug(‘读取到 %r‘, binascii.hexlify(response))

        decompressed = decompressor.decompress(response)
        print(decompressed)
        if decompressed:
            logging.debug(‘解压数据 %r‘, decompressed)
            buffer.write(decompressed.decode(‘utf-8‘))
        else:
            logging.debug(‘缓存中...‘)

    full_response = buffer.getvalue()
    content = open(requested_file, ‘rt‘).read()
    logging.debug(‘读取文件内容与服务器响应数据的匹配: %s‘, full_response == content)

    sk.close()