3. 差分升级包生成算法实现

差分升级包生成算法实现 学习卡片

任务目标

实现一个差分升级包生成算法:给定旧版本固件与新版本固件,生成仅包含差异数据的升级包,并能够从旧固件与差分包还原出新固件。本案例聚焦于bsdiff风格的差分算法核心逻辑,使用Python实现,不依赖第三方库。

核心实现

import hashlib
import struct

def split_chunks(data, size=32):
    """将数据切分为固定大小的块"""
    return [data[i:i+size] for i in range(0, len(data), size)]

def build_index(old_chunks):
    """构建旧固件块的哈希索引,用于快速匹配"""
    index = {}
    for i, chunk in enumerate(old_chunks):
        h = hashlib.md5(chunk).digest()
        if h not in index:
            index[h] = []
        index[h].append(i)
    return index

def generate_diff(old_data, new_data, block_size=32):
    """生成差分包:格式为 [header][diff_entries]"""
    old_chunks = split_chunks(old_data, block_size)
    new_chunks = split_chunks(new_data, block_size)
    index = build_index(old_chunks)
    
    diff_entries = []
    old_pos = 0
    
    for new_chunk in new_chunks:
        h = hashlib.md5(new_chunk).digest()
        matched = False
        
        if h in index:
            for candidate in index[h]:
                if candidate >= old_pos:
                    # 找到匹配块,记录跳过的旧数据长度
                    if candidate > old_pos:
                        diff_entries.append(('skip', candidate - old_pos))
                    diff_entries.append(('copy', candidate))
                    old_pos = candidate + 1
                    matched = True
                    break
        
        if not matched:
            # 未匹配,记录新增数据
            diff_entries.append(('add', new_chunk))
    
    # 序列化差分包
    header = struct.pack('<II', block_size, len(diff_entries))
    payload = b''
    for entry in diff_entries:
        if entry[0] == 'skip':
            payload += struct.pack('<B I', 0, entry[1])
        elif entry[0] == 'copy':
            payload += struct.pack('<B I', 1, entry[1])
        else:  # add
            payload += struct.pack('<B I', 2, len(entry[1])) + entry[1]
    
    return header + payload

def apply_diff(old_data, diff_data):
    """应用差分包,从旧固件还原新固件"""
    block_size, entry_count = struct.unpack('<II', diff_data[:8])
    pos = 8
    result = bytearray()
    old_pos = 0
    
    for _ in range(entry_count):
        op, val = struct.unpack('<B I', diff_data[pos:pos+5])
        pos += 5
        
        if op == 0:  # skip
            old_pos += val
        elif op == 1:  # copy
            result.extend(old_data[val*block_size:(val+1)*block_size])
        else:  # add
            result.extend(diff_data[pos:pos+val])
            pos += val
    
    return bytes(result)

def verify(old_data, new_data, diff_data):
    """验证差分包正确性"""
    restored = apply_diff(old_data, diff_data)
    return restored == new_data, hashlib.md5(restored).hexdigest()

# 最小可运行示例
if __name__ == '__main__':
    old_fw = bytes(range(256)) * 4  # 模拟旧固件:1024字节
    new_fw = old_fw[:512] + bytes([0xFF]*64) + old_fw[576:]  # 修改64字节
    
    diff = generate_diff(old_fw, new_fw)
    ok, digest = verify(old_fw, new_fw, diff)
    
    print(f"差分包大小: {len(diff)} 字节 (原始新固件: {len(new_fw)} 字节)")
    print(f"验证通过: {ok}")
    print(f"还原固件MD5: {digest}")

运行与验证

运行方式:将代码保存为 diff_patch.py,执行 python3 diff_patch.py。

验证方法:程序会输出差分包大小、验证是否通过以及还原固件的MD5值。判定通过条件:验证通过: True,且还原固件MD5与原始新固件一致。

预期输出:差分包大小约为200字节左右(远小于1024字节的新固件),验证通过为True。由于差分包包含64字节新增数据及少量头部信息,实际大小取决于匹配效率。

算法说明

本实现采用块匹配策略:将新旧固件按32字节分块,对旧固件块建立MD5索引。遍历新固件块时,若能在旧固件中找到匹配块,则记录复制操作;否则记录新增数据。差分包格式包含块大小、条目数及操作序列,应用时按序执行即可还原。该算法适用于固件局部修改的场景,能显著减少升级包体积。