Huffman的底层编码解码实现与压缩解压文件实操

import osimport pickle 
# 可对Python的原生对象进行打包存储import structfrom paddle import inverse# 用来存取二进制数据import copy# 定义结点类class Node(object):
    def __init__(self,left=None,right=None,symbol='', weight = 0):
        """
        Args:
            left (_type_, optional): 左子树 Defaults to None.
            right (_type_, optional): 右子树 Defaults to None.
            symbol (str, optional): 字符 Defaults to ''.
            weight (int, optional): 权值 Defaults to 0.
        """
        self.left = left
        self.right = right
        self.symbol = symbol
        self.weight = weight    
    #判断是否为叶子结点
    def isLeaf(self):
        if self.left==None and self.right==None:            return True
        return False# 字符与其编码huffman_code = {}# 根节点huffman_root_node = Node()# 字符及其出现次数letter_frequency  = {}

def readFile(file_path):
    """
    Args:
        file_path (_type_): 文件路径
    在函数体内部把字符及其频率存入全局字典
    Returns:
        _type_: 总的文件字符串
    """
    with open(file_path,'r', encoding="utf-8") as f:
        text = ""
        for line in f.readlines():
            text += line            for char in line:
                letter_frequency[char] = letter_frequency.get(char,0)+1
    return text

def createHuffmanTree():
    """
    创建哈夫曼树，并赋值给全局的根节点
    """
    global huffman_root_node
    n = len(letter_frequency)
    nodes_list = []    for char,frequency in letter_frequency.items():
        nodes_list.append(Node(symbol=char,weight=frequency))        
    for _ in range(n - 1):
        nodes_list.sort(key = (lambda n: n.weight))        # 每次取出最小的元素
        left = nodes_list.pop(0)
        right = nodes_list.pop(0)        # 合并为新的节点
        parent = Node("", left.weight + right.weight)        # 新的节点作为父节点
        parent.left = left
        parent.right = right
        nodes_list.append(parent)        
    if nodes_list:
        huffman_root_node=(nodes_list[0])

def getHuffmanMap(node, code=""):
    """
    递归给哈夫曼树的节点编码
    Args:
        node (_type_): 哈夫曼树的节点
        code (str, optional): 要编码的值（左侧+'0' 右侧+'1'）. Defaults to "".
    """
    if node.isLeaf():
        huffman_code[node.symbol] = code    else:        if node.left:
            getHuffmanMap(node.left, code + "0")        if node.right:
            getHuffmanMap(node.right, code + "1")

def showTree(tree):
    """
    遍历一整颗树，并在叶子节点打印信息
    Args:
        tree (_type_): _description_
    """
    if tree:        if tree.isLeaf():            print(f'{tree.symbol}--{huffman_code[tree.symbol]}--{tree.weight}')
        showTree(tree.left)
        showTree(tree.right)def encode(all_string, file):
    """
    Args:
        all_string (_type_):文件txt中所有字符
        file (_type_):要存储的文件路径
    """
    #  编码
    coded_data = ""
    for letter in all_string:        # print(letter)
        # print(huffman_code[letter])
        # break
        coded_data += huffman_code[letter]        
    # 将全局的letter_frequency和编码后的coded_data转为二进制，并写入new_file
    # pickle只能以二进制格式存储数据到文件
    huffman_map_bytes = pickle.dumps(huffman_code)    # pickle模块实现了python的所有数据序列化和反序列化。它不是用于多种语言间的传输，
    # 它仅作为python对象的持久化或者python程序间进行互相传输对象的。
    with open(file, "wb") as f:        #  存储全局的huffman_map
        # I unsigned int
        # s char[]
        # 存入两个数值，一个是len(huffman_map_bytes)使用unsigned int存储，一个是huffman_map_bytes，使用char存储
        # print(len(huffman_map_bytes))
        f.write(struct.pack('I%ds'%(len(huffman_map_bytes),),len(huffman_map_bytes),huffman_map_bytes))        
        #  存储编码后数据coded_data
        # B unsigned char
        f.write(struct.pack('B',len(coded_data)%8))        # print(len(coded_data)%8)
        # q long long
        # print(len(coded_data)//8)
        f.write(struct.pack('q',len(coded_data)//8))        for index in range(0, len(coded_data),8):            if index + 8 < len(coded_data):                # print(coded_data[index:index+16])
                # break
                #  将8位二进制字符串（占8字节）转换为1位十进制（占1字节）整型
                f.write(struct.pack('B',int(coded_data[index:index+8],2)))            else:                #  最后若干位（<=8）转换成一个十进制整型
                f.write(struct.pack('B',int(coded_data[index:],2)))

def decompress(file):
    """
    Args:
        file (_type_): 二进制文件地址
    Returns:
        _type_:解码后的字符串
    """
    with open(file,"rb") as f:        #  读取huffman_map
        length = struct.unpack('I',f.read(4))[0] #  前四个字节存储了huffman_map
        # print(length)
        huffman_map_byte = pickle.loads(f.read(length))

        moded_bytes = struct.unpack('B',f.read(1))[0]        # print(moded_bytes)
        len_bytes = struct.unpack('q',f.read(8))[0]        # print(len_bytes)

        bin_str_all=''
        # 读取完整部分数据（可以被8整除部分
        for index in range(0, len_bytes):
            byte_str = struct.unpack('B',f.read(1))[0]            # print(byte_str)
            # 在左边填充0，知道填充至8位
            bin_str=bin(byte_str)[2:].zfill(8)
            bin_str_all+=bin_str        # 读取不完整部分数据（不可以被8整除部分
        byte_str = struct.unpack('B',f.read(1))[0]        # 主要区别体现在zfill(moded_bytes)部分，在左边填充0
        bin_str=bin(byte_str)[2:].zfill(moded_bytes)
        bin_str_all+=bin_str        # print(len(bin_str_all))
        # 将原有字典key value对调
        inverse_mappint={}        for k,v in huffman_map_byte.items():
            inverse_mappint[v]=k        
        # print(inverse_mappint)

        # 解码后的字符串
        decode_str=''
        temp_str=''
        for s in bin_str_all:
            temp_str+=s            # 因为没用公众前缀，因此只要出现在字典中，就可以成功解码
            if temp_str in inverse_mappint:
                decode_str+=inverse_mappint[temp_str]
                temp_str=''

        return decode_str

print('样例输入：如本路径下存在一个叫做en.txt的文件，只需要输入en即可')
name=input('请输入统一路径下的txt文件名[不包含.txt]：')
text=readFile(f'./{name}.txt')# 创建哈夫曼树createHuffmanTree()# 获取编码与字符映射关系getHuffmanMap(huffman_root_node)# 打印树与其编码showTree(huffman_root_node)

encode(text,f'./{name}_encode.hfm')
decode_str=decompress(f'./{name}_encode.hfm')with open(f'./{name}_decode.txt','w') as f:
    f.write(decode_str)