Python 压缩与解压缩模块详解

import zipfile
import os

# ---------- 方法 1：write() 添加已有文件 ----------
with zipfile.ZipFile('example.zip', 'w', zipfile.ZIP_DEFLATED) as zf:
    zf.write('file1.txt')                          # 添加文件，归档名与原文件名相同
    zf.write('file2.txt', arcname='docs/file2.txt') # 指定归档内的路径名
    # ZIP_DEFLATED = 使用 zlib 压缩（推荐默认值）
    # ZIP_STORED   = 仅打包不压缩（适合已压缩文件如 .jpg/.mp4）

# ---------- 方法 2：writestr() 直接写入字符串/字节 ----------
with zipfile.ZipFile('data.zip', 'w') as zf:
    zf.writestr('readme.txt', 'This is a readme file')        # 写入文本
    zf.writestr('config.json', '{"key": "value"}')            # 写入 JSON
    zf.writestr('binary.bin', b'\x00\x01\x02\x03')            # 写入二进制

# ---------- 方法 3：追加文件到已有 ZIP ----------
with zipfile.ZipFile('example.zip', 'a') as zf:    # 'a' = append 模式
    zf.write('new_file.txt')

import zipfile

# ---------- 安全检查 ----------
if not zipfile.is_zipfile('example.zip'):           # 验证是否为合法 ZIP
    raise ValueError("Not a valid ZIP file")

with zipfile.ZipFile('example.zip', 'r') as zf:
    # 列出所有文件名
    print(zf.namelist())                            # ['file1.txt', 'docs/file2.txt', ...]

    # 获取文件元数据
    for info in zf.infolist():                      # 每个元素是 ZipInfo 对象
        print(f"文件: {info.filename}")             # 归档内文件名
        print(f"  原始大小: {info.file_size} bytes")
        print(f"  压缩大小: {info.compress_size} bytes")
        print(f"  压缩率: {1 - info.compress_size / info.file_size:.1%}")
        print(f"  修改时间: {info.date_time}")      # (年, 月, 日, 时, 分, 秒)

    # 读取单个文件内容（不解压到磁盘，返回 bytes）
    content = zf.read('file1.txt')
    print(content.decode('utf-8')

    # 提取单个文件
    zf.extract('file1.txt', 'output_folder')        # 解压到 output_folder/file1.txt

    # 提取全部文件
    zf.extractall('extracted_folder')               # 解压所有文件到指定目录

    # 测试 ZIP 完整性（返回第一个损坏文件的文件名，或 None）
    bad = zf.testzip()
    if bad:
        print(f"损坏的文件: {bad}")

import zipfile

# 四种压缩方法对比
methods = {
    zipfile.ZIP_STORED:   "不压缩，仅打包（最快，体积最大）",
    zipfile.ZIP_DEFLATED: "zlib DEFLATE 压缩（默认，速度与压缩率平衡）",
    zipfile.ZIP_BZIP2:    "bzip2 压缩（压缩率更高，速度更慢）",
    zipfile.ZIP_LZMA:     "LZMA 压缩（压缩率最高，速度最慢）",
}

# compresslevel 参数（仅 ZIP_DEFLATED / ZIP_BZIP2 / ZIP_LZMA 有效）
# DEFLATED: 0-9，默认 6
# BZIP2:    1-9，默认 9
# LZMA:     无 compresslevel，使用 preset（0-9，默认 6）

with zipfile.ZipFile('compressed.zip', 'w', zipfile.ZIP_DEFLATED,
                     compresslevel=9) as zf:        # 9 = 最高压缩率
    zf.write('large_file.txt')

import zipfile

# ⚠️ 重要：Python 标准库的 ZIP 密码仅支持旧式 ZIP 加密（弱加密），
#          不支持 AES-256 等强加密。生产环境请使用 pyzipper 第三方库。

# 创建带密码的 ZIP
with zipfile.ZipFile('secure.zip', 'w') as zf:
    zf.setpassword(b'my_password')                  # 设置全局密码（bytes 类型）
    zf.write('secret.txt', compress_type=zipfile.ZIP_DEFLATED)

# 读取带密码的 ZIP
with zipfile.ZipFile('secure.zip', 'r') as zf:
    zf.setpassword(b'my_password')                  # 提供密码
    zf.extractall('extracted')

# 对单个文件设置不同密码（Python 3.7+）
with zipfile.ZipFile('multi_pass.zip', 'w') as zf:
    zf.write('public.txt')                          # 无密码
    zf.write('private.txt')                         # 将使用全局密码
    zf.setpassword(b'global_pwd')

import tarfile

# ---------- 未压缩的 TAR ----------
with tarfile.open('archive.tar', 'w') as tar:
    tar.add('file1.txt')                            # 添加单个文件
    tar.add('directory', arcname='backup/directory') # 添加目录，指定归档内名称

# ---------- gzip 压缩的 TAR ----------
with tarfile.open('archive.tar.gz', 'w:gz') as tar:
    tar.add('file1.txt')

# ---------- bzip2 压缩的 TAR ----------
with tarfile.open('archive.tar.bz2', 'w:bz2') as tar:
    tar.add('file1.txt')

# ---------- lzma/xz 压缩的 TAR ----------
with tarfile.open('archive.tar.xz', 'w:xz') as tar:
    tar.add('file1.txt')

import tarfile

# 'r:*' 自动检测压缩格式（推荐）
with tarfile.open('archive.tar.gz', 'r:*') as tar:
    # 列出所有文件名
    print(tar.getnames())                           # ['file1.txt', 'backup/directory/...']

    # 获取成员元数据
    for member in tar.getmembers():                 # 每个元素是 TarInfo 对象
        print(f"文件: {member.name}")
        print(f"  大小: {member.size} bytes")
        print(f"  权限: {oct(member.mode)}")        # 如 0o100644
        print(f"  UID: {member.uid}, GID: {member.gid}")
        print(f"  类型: {member.type}")              # b'0'=普通文件, b'5'=目录, b'2'=符号链接
        print(f"  修改时间: {member.mtime}")

    # 读取文件内容（不提取到磁盘）
    file_obj = tar.extractfile('file1.txt')         # 返回类文件对象（仅普通文件）
    if file_obj:
        content = file_obj.read()
        print(content.decode('utf-8')

    # 提取单个文件
    tar.extract('file1.txt', 'output_folder')

    # 提取全部文件
    tar.extractall('extracted_folder')

    # 安全提取（Python 3.12+，推荐）
    # tar.extractall('extracted_folder', filter='data')
    # filter='data' 会阻止提取符号链接、设备文件等危险内容

import tarfile

# 写入时过滤文件
def filter_func(tarinfo):
    """自定义过滤器：排除 .pyc 文件并统一权限"""
    if tarinfo.name.endswith('.pyc'):                # 排除编译缓存
        return None                                  # 返回 None 表示跳过
    tarinfo.mode = 0o644                             # 统一设置文件权限
    tarinfo.uid = 0                                  # 统一设置 UID
    tarinfo.gid = 0                                  # 统一设置 GID
    tarinfo.uname = 'root'                           # 用户名
    tarinfo.gname = 'root'                           # 组名
    return tarinfo                                   # 返回修改后的 TarInfo

with tarfile.open('archive.tar.gz', 'w:gz') as tar:
    tar.add('project', filter=filter_func)           # filter 参数过滤

# 读取时选择性提取
with tarfile.open('archive.tar.gz', 'r:*') as tar:
    # 只提取 .py 文件
    py_members = [m for m in tar.getmembers()
                  if m.name.endswith('.py') and m.isfile()]
    tar.extractall('python_files', members=py_members)

import tarfile

# ⚠️ 追加模式仅支持未压缩的 TAR
# 压缩的 TAR（.tar.gz 等）不支持追加，需要重建归档
with tarfile.open('archive.tar', 'a') as tar:
    tar.add('new_file.txt')

# 对压缩 TAR 的"追加"——重建归档
def append_to_compressed_tar(tar_path, new_file, compression='gz'):
    """向压缩 TAR 追加文件的通用方法（重建归档）"""
    import tempfile, os
    temp_path = tempfile.mktemp(suffix='.tar')
    # 1. 解压到临时 TAR
    with tarfile.open(tar_path, f'r:{compression}') as tar_in:
        with tarfile.open(temp_path, 'w') as tar_out:
            for member in tar_in.getmembers():
                tar_out.addfile(member, tar_in.extractfile(member)
            tar_out.add(new_file)                    # 2. 添加新文件
    # 3. 重新压缩
    with tarfile.open(temp_path, 'r') as tar_in:
        with tarfile.open(tar_path, f'w:{compression}') as tar_out:
            for member in tar_in.getmembers():
                tar_out.addfile(member, tar_in.extractfile(member)
    os.remove(temp_path)

import gzip
import shutil

# ---------- 压缩文件（推荐：流式复制，内存安全） ----------
with open('file.txt', 'rb') as f_in:
    with gzip.open('file.txt.gz', 'wb', compresslevel=6) as f_out:
        shutil.copyfileobj(f_in, f_out)             # 分块复制，适合大文件

# ---------- 直接写入压缩数据 ----------
with gzip.open('data.gz', 'wb') as f:
    f.write(b'This is compressed data')
    f.write('中文内容'.encode('utf-8')

# ---------- 解压文件 ----------
with gzip.open('file.txt.gz', 'rb') as f:
    content = f.read()
    print(content.decode('utf-8')

# ---------- 文本模式（自动编解码） ----------
with gzip.open('text.gz', 'wt', encoding='utf-8') as f:
    f.write('压缩的文本内容\n')
    f.write('This is compressed text\n')

with gzip.open('text.gz', 'rt', encoding='utf-8') as f:
    for line in f:                                  # 逐行读取，内存友好
        print(line, end='')

# ---------- compress/decompress（内存中操作） ----------
data = b'This is some data to compress'
compressed = gzip.compress(data, compresslevel=9)   # 返回压缩后的 bytes
decompressed = gzip.decompress(compressed)          # 返回原始 bytes
assert data == decompressed

# ---------- 流式处理大文件 ----------
def compress_large_file(input_path, output_path, chunk_size=65536):
    """流式压缩大文件，内存占用恒定"""
    with open(input_path, 'rb') as f_in:
        with gzip.open(output_path, 'wb') as f_out:
            while True:
                chunk = f_in.read(chunk_size)       # 每次读 64KB
                if not chunk:
                    break
                f_out.write(chunk)

import bz2

# ---------- 压缩文件 ----------
with open('file.txt', 'rb') as f_in:
    with bz2.open('file.txt.bz2', 'wb', compresslevel=9) as f_out:
        f_out.write(f_in.read()

# ---------- 解压文件 ----------
with bz2.open('file.txt.bz2', 'rb') as f:
    content = f.read()
    print(content.decode('utf-8')

# ---------- 文本模式 ----------
with bz2.open('text.bz2', 'wt', encoding='utf-8') as f:
    f.write('压缩的文本内容\n')

with bz2.open('text.bz2', 'rt', encoding='utf-8') as f:
    print(f.read()

# ---------- compress/decompress ----------
data = b'This is some data'
compressed = bz2.compress(data, compresslevel=9)    # 压缩
decompressed = bz2.decompress(compressed)           # 解压
assert data == decompressed

import lzma

# ---------- 压缩文件（默认 XZ 格式） ----------
with open('file.txt', 'rb') as f_in:
    with lzma.open('file.txt.xz', 'wb', preset=6) as f_out:
        f_out.write(f_in.read()

# ---------- 解压文件 ----------
with lzma.open('file.txt.xz', 'rb') as f:
    content = f.read()
    print(content.decode('utf-8')

# ---------- 文本模式 ----------
with lzma.open('text.xz', 'wt', encoding='utf-8') as f:
    f.write('压缩的文本内容\n')

with lzma.open('text.xz', 'rt', encoding='utf-8') as f:
    print(f.read()

# ---------- compress/decompress ----------
data = b'This is some data'
compressed = lzma.compress(data, preset=9)          # preset 0-9，默认 6
decompressed = lzma.decompress(compressed)
assert data == decompressed

# ---------- 格式选择 ----------
# FORMAT_XZ    = 默认，带校验和和索引，推荐（.xz 后缀）
# FORMAT_ALONE = 旧版 LZMA 格式，兼容性差（.lzma 后缀）
# FORMAT_RAW   = 裸数据流，需手动指定 filters

with lzma.open('file.xz', 'wb', format=lzma.FORMAT_XZ) as f:
    f.write(b'data')

with lzma.open('file.lzma', 'wb', format=lzma.FORMAT_ALONE) as f:
    f.write(b'data')

import zlib

# ---------- 一次性压缩/解压 ----------
data = b'Hello, ' * 1000                            # 原始数据

# compress：压缩数据
compressed = zlib.compress(data, level=6)            # level 0-9，默认 6
print(f"原始: {len(data)} bytes -> 压缩后: {len(compressed)} bytes")

# decompress：解压数据
decompressed = zlib.decompress(compressed)
assert data == decompressed

# ---------- 流式压缩（Compress 对象） ----------
comp_obj = zlib.compressobj(level=6,                 # 压缩级别
                            method=zlib.DEFLATED,    # 压缩方法
                            wbits=15)                # zlib 格式（15=zlib头, -15=raw, 31=gzip头）
result = comp_obj.compress(data)                     # 输入数据
result += comp_obj.flush()                           # 刷新缓冲区

# ---------- 流式解压（Decompress 对象） ----------
decomp_obj = zlib.decompressobj(wbits=15)
result = decomp_obj.decompress(compressed)
result += decomp_obj.flush()

# ---------- wbits 参数详解 ----------
# wbits=15:  zlib 格式（RFC 1950），带 zlib 头尾，最常用
# wbits=31:  gzip 格式（RFC 1952），带 gzip 头尾
# wbits=-15: raw DEFLATE（RFC 1951），无头尾，适合自定义协议

# ---------- 计算校验和 ----------
crc32_val = zlib.crc32(data)                        # CRC32 校验和
adler32_val = zlib.adler32(data)                    # Adler-32 校验和（更快）

# ---------- 实战：自定义协议中的压缩 ----------
def pack_compressed_message(payload: bytes) -> bytes:
    """自定义消息格式：4字节长度 + zlib压缩数据 + 4字节CRC32"""
    compressed = zlib.compress(payload, level=6)
    length = len(compressed).to_bytes(4, 'big')
    checksum = zlib.crc32(compressed).to_bytes(4, 'big')
    return length + compressed + checksum

def unpack_compressed_message(packet: bytes) -> bytes:
    """解析自定义压缩消息"""
    length = int.from_bytes(packet[:4], 'big')
    compressed = packet[4:4 + length]
    checksum = int.from_bytes(packet[4 + length:8 + length], 'big')
    if zlib.crc32(compressed) != checksum:
        raise ValueError("CRC32 checksum mismatch")
    return zlib.decompress(compressed)

import shutil

# ---------- 创建归档 ----------
shutil.make_archive('backup', 'zip', 'source_directory')        # backup.zip
shutil.make_archive('backup', 'gztar', 'source_directory')      # backup.tar.gz
shutil.make_archive('backup', 'bztar', 'source_directory')      # backup.tar.bz2
shutil.make_archive('backup', 'xztar', 'source_directory')      # backup.tar.xz
shutil.make_archive('backup', 'tar', 'source_directory')        # backup.tar

# 指定根目录和子目录
shutil.make_archive('backup', 'zip',
                    root_dir='/path/to/parent',                   # 归档根目录
                    base_dir='child')                             # 从哪个子目录开始

# ---------- 解压归档 ----------
shutil.unpack_archive('backup.zip', 'extract_to_directory')
shutil.unpack_archive('backup.tar.gz', 'extract_to_directory')

# 查看支持的格式
print(shutil.get_archive_formats()
# [('bztar', "bzip2'ed tar-file"), ('gztar', "gzip'ed tar-file"),
#  ('tar', 'uncompressed tar file'), ('xztar', "xz'ed tar-file"),
#  ('zip', 'ZIP file')]

import zipfile
import os
from pathlib import Path
from datetime import datetime

def compress_logs(log_dir: str, output_zip: str, pattern: str = '*.log',
                  compresslevel: int = 6) -> dict:
    """
    批量压缩日志文件到 ZIP 归档。

    Args:
        log_dir:     日志目录路径
        output_zip:  输出 ZIP 文件路径
        pattern:     文件匹配模式（默认 *.log）
        compresslevel: 压缩级别 0-9

    Returns:
        包含统计信息的字典
    """
    stats = {'count': 0, 'original_size': 0, 'compressed_size': 0}

    with zipfile.ZipFile(output_zip, 'w', zipfile.ZIP_DEFLATED,
                         compresslevel=compresslevel) as zf:
        for log_file in sorted(Path(log_dir).glob(pattern)):
            original_size = log_file.stat().st_size
            zf.write(log_file, arcname=log_file.name)          # 仅保留文件名
            stats['count'] += 1
            stats['original_size'] += original_size

    # 统计压缩后大小
    stats['compressed_size'] = os.path.getsize(output_zip)
    stats['ratio'] = 1 - stats['compressed_size'] / max(stats['original_size'], 1)

    print(f"压缩完成: {stats['count']} 个文件")
    print(f"原始大小: {stats['original_size'] / 1024:.1f} KB")
    print(f"压缩大小: {stats['compressed_size'] / 1024:.1f} KB")
    print(f"压缩率:   {stats['ratio']:.1%}")
    return stats

compress_logs('logs', 'logs_backup.zip')

import tarfile
import os
import json
from pathlib import Path
from datetime import datetime

def incremental_backup(source_dir: str, backup_dir: str,
                       manifest_path: str = None) -> str:
    """
    增量备份：仅备份自上次备份以来修改过的文件。

    Args:
        source_dir:    要备份的源目录
        backup_dir:    备份文件存放目录
        manifest_path: 备份清单文件路径（记录上次备份的文件时间戳）

    Returns:
        创建的备份文件路径
    """
    source_path = Path(source_dir)
    manifest_path = manifest_path or os.path.join(backup_dir, '.backup_manifest.json')

    # 1. 加载上次备份的清单（文件路径 -> 修改时间）
    old_manifest = {}
    if os.path.exists(manifest_path):
        with open(manifest_path, 'r') as f:
            old_manifest = json.load(f)

    # 2. 扫描当前文件，找出变更
    new_manifest = {}
    changed_files = []
    for filepath in source_path.rglob('*'):
        if not filepath.is_file():
            continue
        rel_path = str(filepath.relative_to(source_path)
        mtime = filepath.stat().st_mtime
        new_manifest[rel_path] = mtime
        if rel_path not in old_manifest or old_manifest[rel_path] < mtime:
            changed_files.append((filepath, rel_path))         # 记录变更文件

    if not changed_files:
        print("没有变更文件，跳过备份")
        return None

    # 3. 创建增量备份（仅包含变更文件）
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_file = os.path.join(backup_dir, f'incremental_{timestamp}.tar.gz')

    with tarfile.open(backup_file, 'w:gz') as tar:
        for filepath, rel_path in changed_files:
            tar.add(filepath, arcname=rel_path)                # 使用相对路径
            print(f"  备份: {rel_path}")

    # 4. 更新清单
    os.makedirs(os.path.dirname(manifest_path), exist_ok=True)
    with open(manifest_path, 'w') as f:
        json.dump(new_manifest, f, indent=2)

    print(f"增量备份完成: {len(changed_files)} 个文件 -> {backup_file}")
    return backup_file

incremental_backup('my_project', 'backups')

import gzip
import json
import urllib.request
from io import BytesIO

def read_remote_gzip(url: str, encoding: str = 'utf-8') -> str:
    """
    直接从 URL 读取 gzip 压缩文件，无需保存到磁盘。

    Args:
        url:      远程 .gz 文件的 URL
        encoding: 文本编码

    Returns:
        解压后的文本内容
    """
    # 1. 下载压缩数据到内存
    with urllib.request.urlopen(url) as response:
        compressed_data = response.read()                       # 读取全部数据到内存

    # 2. 在内存中解压
    decompressed = gzip.decompress(compressed_data)
    return decompressed.decode(encoding)


def stream_remote_gzip(url: str, chunk_size: int = 65536) -> None:
    """
    流式读取远程 gzip 文件，逐行处理，内存占用恒定。

    Args:
        url:       远程 .gz 文件的 URL
        chunk_size: 读取块大小
    """
    # 1. 打开远程连接
    with urllib.request.urlopen(url) as response:
        # 2. 将响应体包装为 GzipFile 进行流式解压
        with gzip.GzipFile(fileobj=response) as gz:
            # 3. 逐行读取
            for line_number, line in enumerate(gz, 1):
                text = line.decode('utf-8').rstrip()
                # 在这里处理每一行数据
                print(f"Line {line_number}: {text[:80]}...")
                if line_number >= 100:                          # 限制演示行数
                    break


def read_zip_from_url(url: str) -> dict:
    """
    从 URL 读取 ZIP 文件，直接在内存中解析。

    Returns:
        {文件名: 文件内容} 的字典
    """
    import zipfile

    with urllib.request.urlopen(url) as response:
        zip_data = BytesIO(response.read())                     # 包装为类文件对象

    result = {}
    with zipfile.ZipFile(zip_data, 'r') as zf:
        for name in zf.namelist():
            if not name.endswith('/'):                          # 跳过目录
                result[name] = zf.read(name).decode('utf-8')
                print(f"读取: {name} ({len(result[name])} chars)")

    return result

import zipfile
import tarfile

def check_zip_integrity(zip_path: str) -> bool:
    """检查 ZIP 文件完整性"""
    try:
        with zipfile.ZipFile(zip_path, 'r') as zf:
            bad_file = zf.testzip()                             # 返回第一个损坏文件名
            if bad_file:
                print(f"损坏的文件: {bad_file}")
                return False
            print("ZIP 文件完整")
            return True
    except zipfile.BadZipFile as e:
        print(f"无效的 ZIP 文件: {e}")
        return False

def check_tar_integrity(tar_path: str) -> bool:
    """检查 TAR 文件完整性"""
    try:
        with tarfile.open(tar_path, 'r:*') as tar:
            tar.getmembers()                                    # 读取所有成员，触发错误
            print("TAR 文件完整")
            return True
    except tarfile.TarError as e:
        print(f"TAR 文件错误: {e}")
        return False

import zipfile
import re

def extract_by_pattern(zip_path: str, pattern: str, extract_to: str) -> list:
    """
    按正则模式选择性提取 ZIP 中的文件。

    Args:
        zip_path:  ZIP 文件路径
        pattern:   正则表达式
        extract_to: 提取目标目录

    Returns:
        提取的文件名列表
    """
    regex = re.compile(pattern)
    extracted = []
    with zipfile.ZipFile(zip_path, 'r') as zf:
        for name in zf.namelist():
            if regex.search(name) and not name.endswith('/'):
                zf.extract(name, extract_to)
                extracted.append(name)
                print(f"提取: {name}")
    return extracted

# 只提取 .py 文件
extract_by_pattern('project.zip', r'\.py$', 'python_files')

import zipfile

# ---------- 问题原因 ----------
# ZIP 规范中文件名编码为 CP437（DOS 时代），中文文件名在不同系统上
# 创建的 ZIP 可能使用不同编码（CP437 / GBK / UTF-8），导致乱码。
# Python 3.7+ 默认使用 CP437 解码文件名，UTF-8 标志位存在时才用 UTF-8。

# ---------- 解决方案 ----------
with zipfile.ZipFile('chinese.zip', 'r') as zf:
    for info in zf.infolist():
        try:
            filename = info.filename
            # 尝试解码为 UTF-8
            filename.encode('cp437').decode('utf-8')
        except (UnicodeDecodeError, UnicodeEncodeError):
            try:
                # 回退到 GBK（中文 Windows 常见）
                filename = info.filename.encode('cp437').decode('gbk')
            except (UnicodeDecodeError, UnicodeEncodeError):
                pass  # 使用原始文件名
        print(filename)

# ---------- 写入时使用 UTF-8（推荐） ----------
# Python 3.4+ 创建 ZIP 时默认使用 UTF-8 标志位
with zipfile.ZipFile('chinese_safe.zip', 'w') as zf:
    zf.write('中文文件.txt')                        # 自动设置 UTF-8 标志位

import zipfile
import shutil

# ❌ 错误：大文件一次性加载到内存
with zipfile.ZipFile('large.zip', 'r') as zf:
    data = zf.read('huge_file.bin')                 # 可能 OOM

# ✅ 正确：流式读取
with zipfile.ZipFile('large.zip', 'r') as zf:
    with zf.open('huge_file.bin') as f_in:          # zf.open() 返回类文件对象
        with open('output.bin', 'wb') as f_out:
            shutil.copyfileobj(f_in, f_out)         # 分块复制

# ✅ 正确：tarfile 的流式读取
import tarfile
with tarfile.open('large.tar.gz', 'r:gz') as tar:
    member = tar.getmember('huge_file.bin')
    f_in = tar.extractfile(member)
    if f_in:
        with open('output.bin', 'wb') as f_out:
            shutil.copyfileobj(f_in, f_out)

# ✅ 正确：gzip 的流式读取
import gzip
with gzip.open('large.gz', 'rb') as f:
    with open('output', 'wb') as f_out:
        for chunk in iter(lambda: f.read(65536), b''):
            f_out.write(chunk)

import zipfile

# ⚠️ Python 标准库的 ZIP 密码使用的是 ZIPCrypto 加密（弱加密，易被破解）
# 不支持 AES-256 等强加密算法

# ✅ 推荐方案 1：使用 pyzipper 第三方库（支持 AES-256）
# pip install pyzipper
# import pyzipper
# with pyzipper.AESZipFile('secure.zip', 'w',
#                          compression=pyzipper.ZIP_DEFLATED,
#                          encryption=pyzipper.WZ_AES) as zf:
#     zf.setpassword(b'strong_password')
#     zf.writestr('secret.txt', 'sensitive data')

# ✅ 推荐方案 2：先加密数据，再压缩
from cryptography.fernet import Fernet
import gzip

key = Fernet.generate_key()                         # 生成加密密钥
cipher = Fernet(key)

data = b'sensitive data'
encrypted = cipher.encrypt(data)                    # 加密
compressed = gzip.compress(encrypted)               # 压缩加密后的数据

import zipfile
import tarfile
import os
import stat

# ---------- TAR 自动保留权限 ----------
with tarfile.open('archive.tar.gz', 'r:gz') as tar:
    for member in tar.getmembers():
        tar.extract(member, 'output')               # 自动恢复权限
        print(f"{member.name}: {oct(member.mode)}")

# ---------- ZIP 需要手动恢复权限 ----------
with zipfile.ZipFile('archive.zip', 'r') as zf:
    for info in zf.infolist():
        zf.extract(info, 'output')
        # 从 external_attr 中提取 Unix 权限（高 16 位）
        if info.external_attr:
            unix_mode = info.external_attr >> 16
            if unix_mode:                           # 非 0 表示有 Unix 权限信息
                file_path = os.path.join('output', info.filename)
                os.chmod(file_path, unix_mode & 0o7777)
                print(f"{info.filename}: {oct(unix_mode & 0o7777)}")

# ---------- 写入 ZIP 时保存权限 ----------
def write_zip_with_permissions(zip_path, files):
    """创建保留 Unix 权限的 ZIP 文件"""
    with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
        for filepath in files:
            stat_info = os.stat(filepath)
            info = zipfile.ZipInfo(filepath)
            # 将 Unix 权限写入 external_attr 的高 16 位
            info.external_attr = (stat_info.st_mode & 0xFFFF) << 16
            with open(filepath, 'rb') as f:
                zf.writestr(info, f.read()

import zipfile
import tarfile
import os

# ---------- zipfile 安全提取 ----------
def safe_extract_zip(zip_path: str, extract_to: str) -> None:
    """安全提取 ZIP 文件，防范路径遍历攻击"""
    extract_to = os.path.abspath(extract_to)
    with zipfile.ZipFile(zip_path, 'r') as zf:
        for member in zf.infolist():
            member_path = os.path.abspath(os.path.join(extract_to, member.filename)
            if not member_path.startswith(extract_to + os.sep):
                raise ValueError(f"危险路径: {member.filename}")
            zf.extract(member, extract_to)

# Python 3.12+ 内置安全过滤
# with zipfile.ZipFile('archive.zip', 'r') as zf:
#     zf.extractall('output', filter='data')  # 自动阻止路径遍历

# ---------- tarfile 安全提取 ----------
def safe_extract_tar(tar_path: str, extract_to: str) -> None:
    """安全提取 TAR 文件，防范路径遍历攻击"""
    extract_to = os.path.abspath(extract_to)
    with tarfile.open(tar_path, 'r:*') as tar:
        for member in tar.getmembers():
            member_path = os.path.abspath(os.path.join(extract_to, member.name)
            if not member_path.startswith(extract_to + os.sep):
                raise ValueError(f"危险路径: {member.name}")
            # 阻止符号链接指向归档外部
            if member.issym() or member.islnk():
                link_path = os.path.abspath(os.path.join(extract_to, member.linkname)
                if not link_path.startswith(extract_to + os.sep):
                    raise ValueError(f"危险链接: {member.name} -> {member.linkname}")
            tar.extract(member, extract_to)

# Python 3.12+ 内置安全过滤
# with tarfile.open('archive.tar.gz', 'r:*') as tar:
#     tar.extractall('output', filter='data')

import gzip

# ⚠️ 对已经压缩的文件（.jpg, .mp4, .zip, .gz 等）再压缩，体积可能反而增大
# 原因：压缩算法添加了头部/尾部元数据，而数据本身已无法进一步压缩

# ✅ 最佳实践：先判断文件类型，跳过已压缩格式
SKIP_EXTENSIONS = {
    '.zip', '.gz', '.bz2', '.xz', '.lzma',           # 压缩文件
    '.jpg', '.jpeg', '.png', '.gif', '.webp',         # 图片
    '.mp3', '.mp4', '.avi', '.mkv',                    # 音视频
    '.pdf', '.docx', '.xlsx', '.pptx',                 # 已压缩文档
    '.7z', '.rar',                                      # 其他压缩格式
}

def should_compress(filepath: str) -> bool:
    """判断文件是否值得压缩"""
    from pathlib import Path
    return Path(filepath).suffix.lower() not in SKIP_EXTENSIONS