如何用Python3创建多分区eMMC的压缩备份（模拟dd管道压缩）

我来帮你完善并优化这个Python脚本，实现和dd | bzip2/dd | xz完全一致的流式压缩备份效果——这种方式不需要把整个eMMC数据加载到内存里，特别适合大容量存储设备的备份场景，而且比单纯的shell命令更方便自定义进度统计这类逻辑。

一、BZ2压缩备份完整实现 #

你已经用到了bz2.BZ2Compressor这个增量压缩器，方向完全正确，下面是补全后的可运行代码：

import bz2
import time

def backup_emmc_bz2(source_path, dest_path, chunksize=1024*1024):
    """
    流式压缩备份eMMC到bz2格式文件
    :param source_path: 源设备路径，比如/dev/mmcblk0
    :param dest_path: 备份文件路径，比如/media/usb0/backup_emmc.bz2
    :param chunksize: 每次读取的块大小，默认1MB，可根据内存调整
    """
    compressor = bz2.BZ2Compressor()
    total_read = 0
    start_time = time.time()

    with open(source_path, 'rb') as src, open(dest_path, 'wb') as dest:
        while True:
            chunk = src.read(chunksize)
            if not chunk:
                # 处理最后一批未压缩的剩余数据
                final_chunk = compressor.flush()
                if final_chunk:
                    dest.write(final_chunk)
                break
            
            # 增量压缩并写入文件
            compressed_chunk = compressor.compress(chunk)
            if compressed_chunk:
                dest.write(compressed_chunk)
            
            # 可选：每5秒打印一次备份进度
            total_read += len(chunk)
            elapsed = time.time() - start_time
            if elapsed > 5:
                speed = total_read / (1024*1024) / elapsed  # 计算MB/s速度
                print(f"已读取: {total_read/(1024**3):.2f} GB | 备份速度: {speed:.2f} MB/s", end='\r')
                start_time = time.time()
                total_read = 0

    print("\n备份完成！")

# 调用示例
if __name__ == "__main__":
    backup_emmc_bz2('/dev/mmcblk0', '/media/usb0/backup_emmc.bz2')

二、XZ压缩备份实现（更高压缩率） #

如果想要更高的压缩率，可以用Python内置的lzma模块实现xz格式的流式压缩：

import lzma
import time

def backup_emmc_xz(source_path, dest_path, chunksize=1024*1024, compression_level=6):
    """
    流式压缩备份eMMC到xz格式文件
    :param source_path: 源设备路径，比如/dev/mmcblk0
    :param dest_path: 备份文件路径，比如/media/usb0/backup_emmc.xz
    :param chunksize: 每次读取的块大小，默认1MB
    :param compression_level: 压缩级别0-9，级别越高压缩率越高、速度越慢，默认6
    """
    compressor = lzma.LZMACompressor(format=lzma.FORMAT_XZ, preset=compression_level)
    total_read = 0
    start_time = time.time()

    with open(source_path, 'rb') as src, open(dest_path, 'wb') as dest:
        while True:
            chunk = src.read(chunksize)
            if not chunk:
                final_chunk = compressor.flush()
                if final_chunk:
                    dest.write(final_chunk)
                break
            
            compressed_chunk = compressor.compress(chunk)
            if compressed_chunk:
                dest.write(compressed_chunk)
            
            # 可选：进度统计
            total_read += len(chunk)
            elapsed = time.time() - start_time
            if elapsed > 5:
                speed = total_read / (1024*1024) / elapsed
                print(f"已读取: {total_read/(1024**3):.2f} GB | 备份速度: {speed:.2f} MB/s", end='\r')
                start_time = time.time()
                total_read = 0

    print("\n备份完成！")

# 调用示例
if __name__ == "__main__":
    backup_emmc_xz('/dev/mmcblk0', '/media/usb0/backup_emmc.xz')

三、关键注意事项 #

• 权限要求：读取/dev/mmcblk0这类块设备需要root权限，运行脚本时记得加sudo，比如sudo python3 backup_script.py
• 块大小调整：chunksize可以根据设备内存大小调整，比如内存充足的话可以设为4MB或8MB，能提升处理效率
• 备份验证：备份完成后可以用bzcat backup_emmc.bz2 | cmp - /dev/mmcblk0（bz2格式）或xzcat backup_emmc.xz | cmp - /dev/mmcblk0（xz格式）验证备份的完整性

内容的提问来源于stack exchange，提问作者Subhamoy S.