Python标准库之数据压缩和存档的应用详解

import tarfile

# 创建带xz压缩的TAR文件
with tarfile.open('archive.tar.xz', 'w:xz') as tar:
    tar.add('/data', arcname='data')  # 保留原始目录结构

# 提取时过滤危险路径
with tarfile.open('archive.tar', 'r') as tar:
    members = tar.getmembers()
    safe_members = [m for m in members if not m.name.startswith('/')]
    tar.extractall(members=safe_members)

import zipfihttp://www.chinasem.cnle

# 分块压缩大文件
with zipfile.ZipFile('large.zip', 'w', compression=zipfile.ZIP_DEFLATED) as zf:
    with open('source.bin', 'rb') as f:
        while chunk := f.read(64*1024):
            zf.writestr('data.bin', chunk, compress_type=zipfile.ZIP_DEFLATED)

# 处理稀疏文件
with zipfile.ZipFile('sparse.zip', 'w') as zf:
    zf.write_sparse('sparse.txt', [(0, 1024), (1048576, 1024)])

import gzip

with gzip.open('large.log.gz', 'wb', compresslevel=5) as f_out:
    with open('large.log', 'rb') as f_in:
        for chunk in iter(lambda: f_in.read(1024*1024), b''):
            f_out.write(chunk)

import mmap

with open('data.bin', 'r+b') as f:
    with mmap.mmap(f.fileno(), 0) as mm:
        compressed = zlib.compress(mm.read())

from multiprocessing import Pool
import zlib

def compress_chunk(chunk):
    return zlib.compress(chunk, level=9)

with Pool(4) as p:
    compressed_data = b''.join(p.imap(compress_chunk, split_into_chunks(data)))

import asyncio
import aiofiles

async def async_compress(input_pathjs, output_path):
    async with aiofiles.open(input_path, 'rb') as f_in:
        async with aiofiles.open(output_path, 'wb') as f_out:
            async for chunk in f_in.iter_chunk(1024*1024):
                await f_out.write(zlib.compress(chunk))

import os
import tarfilESswnze

def add_with_permissions(tar, name):
    info = tar.gettarinfo(name)
    info.mode = os.stat(name).China编程st_mode
    with open(name, 'rb') as f:
        tar.addfile(info, f)

with tarfile.open('archive.tar', 'w') as tar:
    add_with_permissions(tar, 'script.sh')

import tarfile

with tarfile.open('symlinks.tar', 'w') as tar:
    tar.add('/path/to/symlink', filter='data')  # 仅保存符号链接本身

import zlib

data = b'important data'
crc = zlib.crc32(data)
compressed = zlib.compress(data)
# 传输后校验
assert zlib.crc32(zlib.decompress(compressed)) == crc

import os
import shutil
from datetime import datetime

def incremental_backup(source, dest):
    today = datetime.now().strftime('%Y%m%d')
    dest_dir = os.path.join(dest, today)
    os.makedirs(dest_dir, exist_ok=True)
    
    for root, dirs, files in os.walk(source):
        for file in files:
            src_path = os.path.join(root, file)
            dest_path = os.path.join(dest_dir, os.path.relpath(src_path, source))
            if not os.path.exists(dest_path) or \
               os.path.getmtime(src_path) > os.path.getmtime(dest_path):
                shutil.copy2(src_path, dest_path)

import gzip
import shutil
import logging
from logging.handlers import RotatingFileHandler

class CompressedRotatingFileHandler(RotatingFileHandler):
    def doRollover(self):
        super().doRollover()
        with open(self.baseFilename, 'rb') as f_in:
            with gzip.open(self.baseFilename + '.gz', 'wb') as f_out:
                shutil.copyfileobj(f_in,js f_out)
        os.remove(self.baseFilename)

import io
import zlib
import requests

def compress_stream(url):
    response = requests.get(url, stream=True)
    compressor = zlib.compressobj(level=9)
    for chunk in response.iter_content(chunk_size=8192):
        yield compressor.compress(chunk)
    yield compressor.flush()