Python 怎么压缩单个文件或数据流？gzip、bz2、lzma 该怎么选？
简化版
gzip、bz2、lzma 是标准库的三个「单流压缩」模块——它们只负责「把一段数据变小」，不负责「把多个文件打成一个包」（那是 zipfile/tarfile 的活）。三者的 API 高度一致：gzip.open(p, "wt", encoding="utf-8") 像 open() 一样用，gzip.compress(data) / gzip.decompress(data) 处理内存中的 bytes。选型只看一句话：压一次解很多次用 lzma（xz，压缩率最高但压缩极慢），日常通用用 gzip（快、生态最广），bz2 基本被淘汰。 另外三个常被问到的点：① zlib 和 gzip 的区别——zlib 是原始 DEFLATE 数据（没有文件头），gzip 是「DEFLATE + 10 字节文件头（含 mtime、原始文件名）+ 8 字节尾部（CRC32 + 长度）」，所以 HTTP 的 Content-Encoding: gzip 和 deflate 是两种东西；② 流式压缩要用 zlib.compressobj()/decompressobj() 或直接用 gzip.open() 分块读写，别把大文件一次 read() 进内存再压；③ 压缩过的数据（jpg、mp4、已有的 zip）再压不但没用还会变大。还有一个易被忽略的实践点：gzip 会把「原始文件名和修改时间」写进头部，导致同样的输入在不同时刻压出的字节不同——做可复现构建时要传 mtime=0。核心记忆：单流压缩不是归档；gzip 通用、lzma 压得最小；大数据必须流式；已压缩的别再压。
详细版
三个模块对比（100MB 文本日志的量级感受）：
| 模块 | 算法 | 压缩后 | 压缩耗时 | 解压耗时 | 定位 |
|---|---|---|---|---|---|
gzip | DEFLATE | ~12 MB | ~3 s | ~0.7 s | 通用默认，生态最广 |
bz2 | BWT | ~9 MB | ~12 s | ~4 s | 基本淘汰（慢） |
lzma（xz） | LZMA2 | ~7 MB | ~40 s | ~1.5 s | 发行包（压一次解多次） |
zlib | DEFLATE（无头） | ~12 MB | ~3 s | ~0.7 s | 协议内嵌、需要精细控制 |
| zstd（3.14 进标准库） | Zstandard | ~11 MB | ~1 s | ~0.3 s | 现代首选 |
import gzip, bz2, lzma, zlib, shutil, io
# ① 像 open 一样用（三个模块 API 一致）
with gzip.open("app.log.gz", "wt", encoding="utf-8") as f: # ★"t" 才是文本模式★
f.write("第一行\n")
with gzip.open("app.log.gz", "rt", encoding="utf-8") as f:
for line in f: # ★可以直接逐行迭代，惰性解压★
print(line, end="")
# ② 内存中的 bytes
data = b"hello " * 1000
packed = gzip.compress(data, compresslevel=6) # 1(快) ~ 9(小)，默认 9（★注意★）
print(len(data), len(packed)) # 6000 39
assert gzip.decompress(packed) == data
# ③ ★大文件必须流式：别 read() 全进内存★
with open("big.log", "rb") as src, gzip.open("big.log.gz", "wb") as dst:
shutil.copyfileobj(src, dst, length=1 << 20) # ★1MB 一块，内存恒定★
# ④ 解压同理
with gzip.open("big.log.gz", "rb") as src, open("big.log", "wb") as dst:
shutil.copyfileobj(src, dst, length=1 << 20)
# ⑤ zlib：增量压缩（数据不是一次到齐时，如网络流）
co = zlib.compressobj(level=6, wbits=zlib.MAX_WBITS | 16) # ★|16 表示输出 gzip 格式★
out = b"".join(co.compress(chunk) for chunk in chunks) + co.flush()
do = zlib.decompressobj(wbits=zlib.MAX_WBITS | 16)
raw = b"".join(do.decompress(c) for c in in_chunks) + do.flush()
# wbits 速查：15=zlib 格式；-15=★原始 deflate（无头）★；15|16=gzip；15|32=自动识别 zlib/gzip
# ⑥ ★可复现构建：gzip 头里有 mtime，同样输入压出的字节会不同★
a = gzip.compress(b"x") # 含当前时间戳
b = gzip.compress(b"x", mtime=0) # ★固定为 0，字节可复现★
print(a == b) # False
# ⑦ 防解压炸弹：★限制解压后的大小★
def safe_decompress(blob: bytes, limit=100 << 20) -> bytes:
do = zlib.decompressobj(wbits=zlib.MAX_WBITS | 16)
out = do.decompress(blob, limit) # ★max_length 参数，超出部分留在缓冲里★
if do.unconsumed_tail: # 还有没解完的 → 说明超限了
raise ValueError("解压后数据过大")
return out
# ⑧ 选算法：用同一套接口
opener = {"gz": gzip.open, "bz2": bz2.open, "xz": lzma.open}[ext]
with opener(path, "rt", encoding="utf-8") as f:
...⚠️ 三个必须记住的点：①
gzip.open()的默认模式是"rb"（二进制），想按文本读写必须写"rt"/"wt"并给encoding=——只写"r"拿到的是bytes，直接喂给csv.reader或做字符串操作会报错。②gzip.compress()的默认压缩级别是 9（最高），而gzip.open()写文件时默认也是 9——这比gzip命令行工具的默认值 6 慢不少，对大数据来说级别 6 通常是性价比最优点（体积只大 2~3%，速度快近一倍）。③zlib和gzip不是同一种格式：zlib.compress()产出的是带 2 字节 zlib 头的 DEFLATE 流，gzip.compress()产出的是带 10 字节 gzip 头（含魔数1f 8b、mtime、原始文件名）和 8 字节尾部（CRC32 + 原始长度）的格式，两者不能互相解压；要在zlib模块里产出/解析 gzip 格式，得用wbits=zlib.MAX_WBITS | 16（| 32表示自动识别）。HTTP 的Content-Encoding: gzip和deflate的混乱正源于此。
完整版教学
一、压缩 ≠ 归档：三个模块的定位
两件不同的事（★先分清才不会选错工具★）：
归档（archive）：把 N 个文件 + 目录结构打包成 1 个文件 → tarfile / zipfile
压缩（compress）：把 1 段字节流变小 → ★gzip / bz2 / lzma / zlib★
gzip 只能压★一个流★，它不知道"文件"和"目录"的概念
→ 所以 Unix 的组合拳是：tar 打包成一个流 → gzip 压缩这个流 → x.tar.gz
三个模块的统一接口（记一套就够）：
模块.open(path, mode, ...) 像内置 open()，返回文件对象
模块.compress(data, ...) bytes → bytes
模块.decompress(data) bytes → bytes
模块.压缩文件类(fileobj=...) 包装任意文件对象（★关键能力★）
gzip.open / gzip.compress / gzip.GzipFile
bz2.open / bz2.compress / bz2.BZ2File
lzma.open / lzma.compress / lzma.LZMAFile
→ ★三者可以用同一段代码切换★：opener = {"gz": gzip.open, "xz": lzma.open}[ext]
zlib 是"底层"，不是第四种算法：
zlib 和 gzip ★用的是同一个 DEFLATE 算法★，区别只在★外包装★：
zlib.compress(data) → 2 字节 zlib 头 + DEFLATE 数据 + 4 字节 Adler-32
gzip.compress(data) → ★10 字节 gzip 头★ + DEFLATE 数据 + ★CRC32 + 原始长度★
原始 deflate → 只有 DEFLATE 数据，什么头都没有
→ 三种"包装"互不兼容，用错了就是 "Error -3 while decompressing: incorrect header check"
★ 这正是 HTTP 里 Content-Encoding 混乱的根源：
"gzip" → gzip 格式（有头有尾）
"deflate" → 规范说是 zlib 格式，但★很多老服务器发的是原始 deflate★
→ 客户端要用 wbits=-15 兜底重试（requests/urllib3 内部就这么干）第一步是分清归档和压缩：tarfile/zipfile 负责「把多个文件打成一个包」，而 gzip/bz2/lzma 只负责把一段字节流变小、完全不知道「文件」和「目录」的概念——这就是为什么 Unix 要用 tar 先打包成一个流、再用 gzip 压缩它，产出 .tar.gz。三个模块的接口高度一致（open/compress/decompress 加一个文件类），所以可以用同一段代码在算法之间切换。要特别理解 zlib 不是第四种算法：它和 gzip 用的是同一个 DEFLATE 算法，区别只在外包装——zlib 是 2 字节头 + Adler-32 校验，gzip 是 10 字节头（魔数 1f 8b、mtime、原始文件名）+ CRC32 + 原始长度，还有一种是什么头都没有的原始 deflate。三种包装互不兼容，用错就报 incorrect header check——HTTP 里 Content-Encoding: deflate 的著名混乱正源于此（规范说是 zlib 格式，但很多老服务器发的是原始 deflate，客户端只能用 wbits=-15 兜底重试）。
二、流式压缩：别把大文件读进内存
✗ 反面写法（内存直接吃满）：
data = open("10GB.log", "rb").read() # ★10GB 进内存★
blob = gzip.compress(data) # 再来一份
✓ 写法一：gzip.open + shutil.copyfileobj（★最简洁★）
with open("big.log", "rb") as src, gzip.open("big.log.gz", "wb") as dst:
shutil.copyfileobj(src, dst, length=1 << 20) # ★内存恒定 ≈ 1MB★
✓ 写法二：手动分块（需要在中途做别的事时）
with gzip.open("out.gz", "wb") as dst:
for chunk in iter(lambda: src.read(1 << 20), b""):
dst.write(chunk)
✓ 写法三：zlib 增量对象（★数据不是一次到齐时★，如边收网络包边压）
co = zlib.compressobj(6, zlib.DEFLATED, zlib.MAX_WBITS | 16)
for chunk in incoming:
out.write(co.compress(chunk)) # ★可能返回 b""（数据还在内部缓冲里）★
out.write(co.flush()) # ★必须 flush，否则尾部数据丢失★
解压端：
do = zlib.decompressobj(zlib.MAX_WBITS | 16)
for chunk in incoming:
data = do.decompress(chunk) # 同样可能返回 b""
if data: handle(data)
handle(do.flush())
★ 三个流式压缩的关键细节：
① compress() 的返回值可能是空 bytes —— 压缩器在攒够一个块之前不输出，
★千万别用 "if not out: 结束了" 来判断★
② ★必须调 flush()★ —— 不调的话最后一块数据和尾部校验都不会写出，
产出的文件解压时报 "unexpected end of file"
③ decompressobj 有 unused_data / unconsumed_tail 两个属性：
unused_data —— 压缩流结束后剩下的字节（★多成员 gzip 或后面还有别的数据★）
unconsumed_tail —— 因 max_length 限制而没解完的输入（★防炸弹时用★）
包装任意文件对象（不落地）：
buf = io.BytesIO()
with gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as f:
f.write(b"data")
blob = buf.getvalue() # ★内存里造一个 .gz，直接返回给前端★
# 反过来：解压 HTTP 响应流，不落地
with gzip.GzipFile(fileobj=resp.raw) as f:
for line in io.TextIOWrapper(f, encoding="utf-8"): # ★套一层才有文本迭代★
handle(line)流式处理是压缩模块的核心用法。最简洁的写法是 gzip.open() 配 shutil.copyfileobj(src, dst, length=1<<20)——内存占用恒定在一块的大小，10GB 文件也能轻松处理。当数据不是一次到齐（边收网络包边压）时要用 zlib.compressobj()/decompressobj()，这里有三个关键细节：① compress() 的返回值可能是空 bytes（压缩器攒够一块才输出），千万别用它判断是否结束；② 必须调 flush()，否则最后一块数据和尾部校验都不会写出，产出的文件解压时报 unexpected end of file；③ decompressobj 有 unused_data 和 unconsumed_tail 两个属性——前者是压缩流结束后剩下的字节（多成员 gzip 或后面接着别的数据），后者是因 max_length 限制没解完的输入（防解压炸弹时用）。另一个高价值能力是包装任意文件对象：gzip.GzipFile(fileobj=io.BytesIO()) 能在内存里造一个 .gz 直接返回给前端，GzipFile(fileobj=resp.raw) 则能边下载边解压、不落地。
三、选算法：压缩率、速度与场景
三个维度的权衡（100MB 文本日志的量级）：
┌──────────┬────────┬────────┬────────┬──────────────────────────┐
│ 算法 │ 压缩后 │ 压缩 │ 解压 │ 什么时候选 │
├──────────┼────────┼────────┼────────┼──────────────────────────┤
│ 不压缩 │ 100 MB │ — │ — │ ★已压缩的数据（jpg/mp4）★ │
│ gzip -1 │ ~16 MB │ ~1 s │ ~0.7 s │ 实时流、CPU 紧张 │
│ gzip -6 │ ~12 MB │ ~3 s │ ~0.7 s │ ★通用默认（性价比最优）★ │
│ gzip -9 │ ~11 MB │ ~6 s │ ~0.7 s │ 收益很小，通常不值 │
│ bz2 │ ~9 MB │ ~12 s │ ~4 s │ 几乎没有场景（被 xz 取代）│
│ xz/lzma │ ~7 MB │ ~40 s │ ~1.5 s │ ★发行包：压一次解无数次★ │
│ zstd -3 │ ~11 MB │ ~1 s │ ~0.3 s │ ★现代首选（3.14 进标准库）│
└──────────┴────────┴────────┴────────┴──────────────────────────┘
★ 注意 gzip 的级别 6→9：体积只小 ~8%，时间翻倍 → ★大多数场景 6 就够★
而 Python 的 gzip.compress() ★默认是 9★（命令行 gzip 默认是 6）→ 记得显式指定
按场景决策：
① 日志归档（写一次、偶尔查） → gzip -6（或 zstd）
② 软件发行包（压一次、下载千万次） → ★xz★（多花的压缩时间被下载节省摊平）
③ HTTP 响应压缩（每个请求都要压） → ★gzip -1~4 或 br/zstd★（延迟敏感）
④ 备份到冷存储（存储费按 GB 算） → xz（体积就是钱）
⑤ 进程间/网络传输的中间数据 → ★zstd 或 lz4★（速度优先）
⑥ 已经压缩的数据（jpg/mp4/zip/parquet）→ ★别压★，会变大且浪费 CPU
★ 一个反直觉的算例：压缩不总是省时间
传输 100MB 数据，带宽 100 MB/s：
不压缩：传输 1.0 s，总计 1.0 s
gzip-6：压缩 3.0 s + 传 0.12 s + 解压 0.7 s = ★3.8 s（更慢！）★
gzip-1：压缩 1.0 s + 传 0.16 s + 解压 0.7 s = 1.9 s（仍更慢）
带宽 10 MB/s 时：
不压缩：10 s
gzip-6：3.0 + 1.2 + 0.7 = ★4.9 s（值了）★
→ ★压缩是"用 CPU 换带宽/存储"，带宽越差、数据越重复，越划算★
内存占用（容易忽略）：
gzip/zlib：解压约 32KB 窗口 + 缓冲，很轻
★lzma/xz：解压内存取决于压缩时的字典大小，最高可达 ★700MB+★
→ 在内存受限的环境（容器、嵌入式）解压 xz 可能 OOM
→ lzma.LZMADecompressor(memlimit=...) 可以设上限选算法的核心是**「用 CPU 换带宽/存储」这笔账划不划算**。上面那个反直觉的算例很关键：带宽 100MB/s 时，压缩后再传反而更慢（3.8 秒 vs 1.0 秒），而带宽降到 10MB/s 时压缩就值了（4.9 秒 vs 10 秒）——带宽越差、数据越重复，压缩越划算。场景决策也很清晰：日志归档用 gzip -6、软件发行包用 xz（多花的压缩时间被千万次下载摊平）、HTTP 响应压缩用低级别 gzip 或 br/zstd（延迟敏感）、已经压缩过的数据（jpg/mp4/zip/parquet）根本别压。两个容易忽略的细节：Python 的 gzip.compress() 默认级别是 9 而命令行 gzip 默认是 6——从 6 提到 9 体积只小 8% 但时间翻倍，通常应该显式指定 6；以及 lzma 解压的内存占用可能高达数百 MB（取决于压缩时的字典大小），在容器等内存受限环境下可能 OOM，可以用 LZMADecompressor(memlimit=...) 设上限。
四、gzip 格式的细节与可复现构建
gzip 文件的结构：
┌──────────────────────────────────────────────────────────┐
│ 10 字节固定头 │
│ 1f 8b 魔数（★判断是不是 gzip 就看这两字节★） │
│ 08 压缩方法（DEFLATE） │
│ FLG 标志位（是否含文件名/注释/额外字段） │
│ MTIME ★4 字节修改时间（Unix 时间戳）★ │
│ XFL,OS 额外标志、操作系统 │
├──────────────────────────────────────────────────────────┤
│ 可选：★原始文件名★（以 \0 结尾）、注释、额外字段 │
├──────────────────────────────────────────────────────────┤
│ DEFLATE 压缩数据 │
├──────────────────────────────────────────────────────────┤
│ CRC32（4 字节）+ ISIZE（原始长度 mod 2^32，4 字节） │
└──────────────────────────────────────────────────────────┘
由此产生三个实用推论：
① ★可复现构建的坑★：头部含 MTIME 和原始文件名
gzip.compress(b"x") 两次调用产出的字节★不同★（时间戳变了）
→ 构建产物的哈希每次都变，缓存全部失效
✓ gzip.compress(data, mtime=0)
✓ gzip.GzipFile(fileobj=buf, mode="wb", mtime=0, filename="")
（命令行等价物：gzip -n）
② ★ISIZE 只有 4 字节 = 原始长度 mod 2^32★
→ 超过 4GB 的文件，这个字段会回绕
→ 不能拿它当"解压后大小"的可靠依据（防炸弹时尤其要注意）
③ ★gzip 支持"多成员"★：多个 gzip 流直接拼接仍是合法的 gzip
cat a.gz b.gz > c.gz # ✓ 解压出来是 a+b 的内容
→ Python 的 gzip 模块解压时会自动处理多成员
→ 但用 zlib.decompressobj 手动解时，第一个成员结束后
剩余数据在 ★unused_data★ 里，要自己循环处理下一个成员
判断文件是不是压缩过（魔数）：
gzip: 1f 8b bz2: 42 5a 68 ('BZh')
xz: fd 37 7a 58 5a 00 zip: 50 4b 03 04 ('PK')
zstd: 28 b5 2f fd
✓ with open(p,"rb") as f: magic = f.read(6)
其他实用参数：
gzip.open(p, "wb", compresslevel=6)
gzip.GzipFile(filename="原始名", mtime=..., fileobj=..., mode=...)
f.rewind() / f.seek() ★GzipFile 支持 seek，但向后 seek 要从头重新解压★（很慢）
lzma.open(p, format=lzma.FORMAT_XZ / FORMAT_ALONE, preset=6, filters=[...])
bz2.open(p, compresslevel=9)理解 gzip 的文件格式能解释三个实际问题。① 可复现构建：gzip 头里存着 MTIME 和原始文件名，所以 gzip.compress(b"x") 两次调用产出的字节不同——构建产物的哈希每次都变、缓存全部失效；解法是 mtime=0（命令行是 gzip -n）。② ISIZE 字段只有 4 字节（原始长度 mod 2³²），超过 4GB 会回绕，不能拿它当「解压后大小」的可靠依据（防炸弹时尤其要注意）。③ gzip 支持「多成员」——多个 gzip 流直接 cat 拼接仍是合法的 gzip 文件，Python 的 gzip 模块解压时会自动处理，但用 zlib.decompressobj 手动解时第一个成员结束后剩余数据会留在 unused_data 里，需要自己循环处理。另外记住几个魔数用于判断文件类型：gzip 是 1f 8b、bz2 是 BZh、xz 是 fd 37 7a 58 5a 00、zip 是 PK。
五、安全：解压炸弹与资源限制
解压炸弹的原理：极高重复的数据压缩比可以达到 ★1000:1 甚至更高★
1GB 的全零文件 → gzip 后约 1MB（1000:1）
精心构造的嵌套炸弹 → 42KB 解压出 4.5PB
★ 流压缩场景的风险点：
- 接收用户上传的 .gz 并解压
- HTTP 客户端自动解压 Content-Encoding: gzip 的响应（★恶意服务器★）
- 解析别人给的压缩日志
★ 防御手段（三层）：
① 限制解压输出量（★最有效★）
do = zlib.decompressobj(zlib.MAX_WBITS | 16)
out = do.decompress(blob, MAX_OUT) # ★max_length 参数★
if do.unconsumed_tail: # 还有没吃完的输入
raise ValueError("解压后超过限制")
流式版本（边解边计数）：
total = 0
with gzip.open(p, "rb") as f:
while chunk := f.read(1 << 20):
total += len(chunk)
if total > MAX_OUT:
raise ValueError("解压炸弹")
out.write(chunk)
② 限制压缩比
ratio = 已解压字节 / 已读取的压缩字节
if ratio > 100: 拒绝 # 正常文本 3~10 倍，>100 高度可疑
③ 系统层兜底
- 磁盘配额 / ulimit -f（限制单个文件大小）
- 容器内存和磁盘限制
- 解压到临时目录，校验通过再移动
★ 不能依赖的两个"看起来靠谱"的字段：
✗ gzip 尾部的 ISIZE（原始长度）：只有 4 字节会回绕，★而且可以被伪造★
✗ 文件本身的大小：炸弹的特征恰恰是"输入很小"
lzma 的额外风险：★解压内存★
xz 的解压内存取决于压缩时用的字典大小（最高 1.5GB）
→ 恶意 .xz 可以让你的进程直接 OOM
✓ lzma.LZMADecompressor(memlimit=64 << 20) # ★超限抛 LZMAError★
✓ lzma.open(p, format=..., filters=...) 也可以限制
处理外部数据的完整模板：
def safe_gunzip(src: Path, dst: Path, max_out=1 << 30, max_ratio=100):
total_in = src.stat().st_size
total_out = 0
with gzip.open(src, "rb") as fin, open(dst, "wb") as fout:
while chunk := fin.read(1 << 20):
total_out += len(chunk)
if total_out > max_out or total_out / max(total_in, 1) > max_ratio:
dst.unlink(missing_ok=True) # ★清理已写的部分★
raise ValueError("疑似解压炸弹")
fout.write(chunk)解压炸弹在流压缩场景同样存在——极高重复的数据压缩比能达到 1000:1（1GB 全零文件 gzip 后只有约 1MB），风险点包括接收用户上传的 .gz、以及 HTTP 客户端自动解压恶意服务器返回的 Content-Encoding: gzip 响应。防御要分三层：① 限制解压输出量（最有效，用 decompressobj.decompress(data, max_length) 或边解边计数，超限立刻中断）；② 限制压缩比（正常文本 3~10 倍，超过 100 倍高度可疑）；③ 系统层兜底（磁盘配额、容器内存限制、先解压到临时目录）。有两个看起来靠谱但不能依赖的字段：gzip 尾部的 ISIZE（只有 4 字节会回绕，而且可以被伪造）和输入文件本身的大小（炸弹的特征恰恰是输入很小）。另外 lzma 有额外的内存风险——解压所需内存取决于压缩时的字典大小（最高 1.5GB），恶意 .xz 能让进程直接 OOM，要用 LZMADecompressor(memlimit=...) 设上限。
六、实战组合与常见任务
① 滚动日志自动压缩（logging 的 rotator 钩子）
import logging.handlers, gzip, shutil, os
def gz_rotator(source, dest):
with open(source, "rb") as f_in, gzip.open(f"{dest}.gz", "wb") as f_out:
shutil.copyfileobj(f_in, f_out)
os.remove(source)
h = logging.handlers.RotatingFileHandler("app.log", maxBytes=10<<20, backupCount=5)
h.rotator = gz_rotator # ★轮转时自动压缩旧文件★
h.namer = lambda name: name # 配合 rotator 使用
② 直接读压缩的 CSV / JSON（★不解压到磁盘★）
import csv, io, json
with gzip.open("data.csv.gz", "rt", encoding="utf-8", newline="") as f:
for row in csv.DictReader(f): # ★"rt" + newline="" 是 csv 的正确姿势★
...
with gzip.open("data.jsonl.gz", "rt", encoding="utf-8") as f:
for line in f:
obj = json.loads(line) # ★JSON Lines + gzip 是日志分析的标配★
③ 内存中生成 .gz 返回给前端（不落临时文件）
buf = io.BytesIO()
with gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as gz:
gz.write(csv_text.encode("utf-8"))
return Response(buf.getvalue(), headers={"Content-Encoding": "gzip"})
④ 边下载边解压（★内存恒定★）
with requests.get(url, stream=True) as r:
with gzip.GzipFile(fileobj=r.raw) as gz:
for line in io.TextIOWrapper(gz, encoding="utf-8"):
handle(line)
⑤ 压缩多个文件 → 用 tar.gz（★不要循环 gzip 每个文件★）
with tarfile.open("all.tar.gz", "w:gz") as t:
t.add("logs/")
★ 原因：整体压缩能跨文件复用字典，1000 个相似小文件可能差 5~10 倍
⑥ 判断并自动选择解压方式
MAGIC = {b"\x1f\x8b": gzip.open, b"BZh": bz2.open, b"\xfd7zXZ\x00": lzma.open}
def smart_open(p, mode="rt", **kw):
with open(p, "rb") as f:
head = f.read(6)
for magic, opener in MAGIC.items():
if head.startswith(magic):
return opener(p, mode, **kw)
return open(p, mode, **kw) # ★没匹配上就是普通文件★
⑦ 校验压缩文件是否完整
try:
with gzip.open(p, "rb") as f:
while f.read(1 << 20): pass # ★读到底会校验 CRC32★
except (EOFError, gzip.BadGzipFile, zlib.error) as e:
print("文件损坏:", e)实战中最有价值的是几个「不落地」的组合：直接读压缩的 CSV/JSON（gzip.open(p, "rt", newline="") 喂给 csv.DictReader，JSON Lines + gzip 是日志分析的标配）、在内存里生成 .gz 返回给前端（GzipFile(fileobj=io.BytesIO())）、以及边下载边解压（GzipFile(fileobj=r.raw) 套 TextIOWrapper，内存恒定）。日志场景可以给 RotatingFileHandler 挂一个 rotator 钩子实现轮转时自动压缩。有一条选型提醒要记牢：压缩多个文件应该用 tar.gz 而不是循环 gzip 每一个——整体压缩能跨文件复用字典，1000 个相似小文件可能差 5~10 倍。最后两个小工具：用魔数自动判断该用哪个 opener，以及完整读一遍来校验 CRC32（gzip 的校验只在读到流末尾时才做，中途读一半是发现不了损坏的）。
记忆钩子：「★先分清压缩和归档★：gzip/bz2/lzma 只把『一段字节流』变小、不认识文件和目录，所以要配 tar 才有 .tar.gz。三者接口一致（open/compress/decompress），★选型只看一句：通用用 gzip、压一次解无数次用 xz(lzma)、bz2 已淘汰、zstd 是现代首选（3.14 进标准库）★；而且『压缩不总是省时间』——带宽 100MB/s 时压完再传反而更慢，★带宽越差、数据越重复才越划算★，已压缩的 jpg/mp4/zip ★再压会变大★。★zlib 不是第四种算法★，它和 gzip 用同一个 DEFLATE，区别只在外包装：zlib=2 字节头+Adler32、gzip=10 字节头（含 mtime 和原始文件名）+CRC32+长度、原始 deflate 什么都没有——三者互不兼容（
incorrect header check就是这么来的），在 zlib 里产出 gzip 要 ★wbits=MAX_WBITS|16★（|32 自动识别），HTTP 的 Content-Encoding: deflate 混乱也源于此。★大数据必须流式★：gzip.open + shutil.copyfileobj(length=1MB)，或用 compressobj/decompressobj 增量处理——注意 ★compress() 可能返回空 bytes（别拿它判断结束）、必须 flush()（否则尾部丢失报 unexpected end of file）、unused_data 装着多成员 gzip 的剩余部分★。★gzip 头里有 mtime 和文件名，所以同样输入压出的字节每次都不同★，可复现构建要传 mtime=0（等价 gzip -n）；尾部 ISIZE 只有 4 字节会回绕★且能被伪造★，不能当解压后大小的依据。安全上要防解压炸弹：★用 decompress(data, max_length) 或边解边计数 + 限制压缩比（>100 可疑）★，lzma 还有额外的解压内存风险（最高 1.5GB，用 memlimit）。最后：Python 的 gzip 默认级别是 ★9★（命令行 gzip 是 6），6→9 体积只小 8% 但耗时翻倍。」
七、常见误区与追问
- 误区：
gzip可以像 zip 一样打包多个文件。 不能——gzip只是「把一段字节流变小」的压缩器，它的格式里根本没有「文件列表」「目录结构」这类概念，一个.gz文件对应且只对应一份数据流。要把多个文件放进一个压缩文件有两条路：tar先打包成一个流、再用 gzip 压缩它（得到.tar.gz，Unix 的标准做法），或者直接用zipfile（它本身就是「归档 + 压缩」一体的容器格式）。顺带记住一个性能事实：压缩多个文件应该整体打包再压，而不是逐个 gzip——整体压缩能跨文件复用字典（相似内容的重复被一起消除），1000 个内容相似的小日志文件，tar.gz可能只有逐个 gzip 的五分之一到十分之一大。 - 误区：
zlib.decompress()能解开.gz文件。 会报zlib.error: Error -3 while decompressing data: incorrect header check。因为它们是同一个 DEFLATE 算法的三种不同外包装：zlib格式是 2 字节头 + DEFLATE + 4 字节 Adler-32；gzip格式是 10 字节头（魔数1f 8b、mtime、可选的原始文件名）+ DEFLATE + CRC32 + 原始长度；还有一种是裸 DEFLATE（什么头都没有）。要在zlib模块里处理 gzip 格式，必须用wbits=zlib.MAX_WBITS | 16（| 32表示自动识别 zlib 还是 gzip，-15表示裸 deflate）。这套区别也正是 HTTP 里Content-Encoding: deflate长期混乱的根源——规范说它是 zlib 格式，但很多老服务器发的是裸 deflate，所以成熟的客户端库都会先按 zlib 解、失败再用wbits=-15重试。 - 误区：
gzip.open(p, "r")读出来的是字符串。 默认是二进制模式（等价于"rb"），读到的是bytes——这和内置open()默认文本模式的习惯正好相反，是很容易踩的一个坑（拿去做字符串操作或喂给csv.reader就会报错）。要按文本读写必须显式写"rt"/"wt"并指定encoding=（还要注意：给csv用时同样需要newline=""）。同样的规则适用于bz2.open和lzma.open。另一个相关的默认值差异是压缩级别：Python 的gzip.compress()和gzip.open()默认用 9（最高），而命令行gzip默认是 6——从 6 提到 9 通常只让体积小 2~8%，耗时却接近翻倍，处理大数据时应该显式传compresslevel=6。 - 误区：解压前先检查一下 gzip 尾部记录的原始大小，就能防住解压炸弹。 两个原因不能依赖它：① ISIZE 字段只有 4 字节，存的是「原始长度 mod 2³²」，超过 4GB 就会回绕（一个 5GB 的文件会显示成 1GB）；② 这个字段完全可以被伪造——攻击者构造炸弹时随手把它填成一个很小的值，你的检查就形同虚设。正确的防御是在解压过程中实时限制输出量：用
decompressobj.decompress(data, max_length)限制单次产出（检查unconsumed_tail是否非空），或者流式读取时累计已解压字节数、超过阈值立刻中断并删除已写出的部分；同时监控压缩比（正常文本 3~10 倍，超过 100 倍高度可疑）。再叠加系统层的磁盘配额和容器内存限制作为兜底。 - 误区：用
zlib.compressobj()压缩完，把每次compress()的返回值拼起来就是完整的压缩数据。 少了flush()的那一段。压缩器内部有缓冲，compress()在攒够一个块之前会返回空bytes，而最后一块数据和格式尾部（gzip 的 CRC32 + ISIZE）只有在调用flush()时才输出。漏掉它的后果是产出的文件看似有内容、解压时却报unexpected end of file或 CRC 校验失败，而且小数据量时可能碰巧没问题（一次就 flush 完了），到大数据量才暴露。同理，解压端的decompressobj也要在最后调flush()。还有一个连带的坑：不能用「compress()返回空」来判断数据是否处理完——那只说明压缩器还在攒数据。 - 追问：什么时候该用
lzma（xz），什么时候用gzip？ 核心看**「压缩一次，解压多少次」这个比值。lzma的压缩率明显更高（同样数据比 gzip 小 30~40%），但压缩速度慢一个数量级**（上面的算例：40 秒 vs 3 秒），解压则只比 gzip 慢一点。所以：软件发行包、系统镜像、长期归档这类「压一次、下载/解压千万次」的场景用 xz——多花的几十秒压缩时间被节省的带宽和存储成本彻底摊平；而日志滚动压缩、HTTP 响应压缩、进程间传输这类「压缩频繁、解压次数不多、延迟敏感」的场景必须用 gzip（甚至低级别的 gzip -1）或 zstd。还有两个 xz 的隐藏成本要注意：解压所需内存取决于压缩时的字典大小，最高可达 1.5GB（容器里可能直接 OOM，要用LZMADecompressor(memlimit=...)限制）；以及它的多线程支持在标准库里不可用（命令行xz -T0才有）。 - 追问：为什么同样的输入，
gzip.compress()每次产出的字节都不一样？ 因为 gzip 的文件头里包含 MTIME（4 字节 Unix 时间戳），gzip.compress()默认把当前时间写进去；用gzip.open()压缩文件时，头部还可能带上原始文件名。这在可复现构建（reproducible build） 场景是个真问题：同一份源码在不同时刻打包出的产物字节不同 → 产物哈希不同 → 下游的构建缓存、制品去重、签名校验全部失效，也无法证明「这个二进制确实由这份源码构建」。解法是显式固定这些字段：gzip.compress(data, mtime=0)，或者gzip.GzipFile(fileobj=buf, mode="wb", mtime=0, filename="")（命令行的等价物是gzip -n，tar则要配合--mtime、--owner、--sort=name等一整套参数）。zlib和lzma没有这个问题（它们的头部不含时间戳）。 - 追问：怎么处理「边下载边解压」这类流式场景？ 关键是把「网络流」直接作为
fileobj交给压缩模块，而不是先下载到内存或磁盘。典型写法是with requests.get(url, stream=True) as r: with gzip.GzipFile(fileobj=r.raw) as gz:——GzipFile会按需从r.raw读取压缩数据、解压出一块给你一块，内存占用恒定（约几十 KB 的窗口 + 缓冲），10GB 的远程日志也能处理。如果要按文本行处理，再套一层io.TextIOWrapper(gz, encoding="utf-8")就能for line in ...。反方向（边生成边压缩上传）同理：把一个可写的流对象交给GzipFile(fileobj=...)。两个注意点：①requests会自动解压Content-Encoding: gzip的响应，所以如果 URL 本身就是.gz文件（Content-Type而非Content-Encoding）才需要自己解，两者别搞混；② 流式场景更要防解压炸弹——边解边累计输出字节数并设上限，因为你无法预先知道远程数据的真实大小。
八、加强记忆
先分清压缩和归档：gzip/bz2/lzma 只把「一段字节流」变小、不认识文件和目录，所以要配 tar 才有 .tar.gz；三者接口高度一致（open/compress/decompress + 文件类），可以用同一段代码切换。选型一句话：通用用 gzip、「压一次解无数次」用 xz(lzma)、bz2 已淘汰、zstd 是现代首选（3.14 进标准库）；而且要记住**「压缩不总是省时间」——带宽 100MB/s 时压完再传反而更慢（3.8s vs 1.0s），带宽越差、数据重复度越高才越划算，而 jpg/mp4/zip 这类已压缩数据再压会变大**。zlib 不是第四种算法：它和 gzip 用同一个 DEFLATE，区别只在外包装——zlib 是 2 字节头 + Adler-32，gzip 是 10 字节头（含 mtime 和原始文件名）+ CRC32 + 原始长度，还有什么都没有的裸 deflate；三者互不兼容（incorrect header check 就是这么来的），要在 zlib 里处理 gzip 得用 wbits=zlib.MAX_WBITS | 16（|32 自动识别、-15 裸 deflate），HTTP 的 Content-Encoding: deflate 混乱也源于此。大数据必须流式：gzip.open + shutil.copyfileobj(length=1MB)，或用 compressobj/decompressobj 增量处理——注意 compress() 可能返回空 bytes（别拿它判断结束）、必须 flush()（否则尾部丢失、解压报 unexpected end of file）、unused_data 里装着多成员 gzip 的剩余部分。gzip 头里有 mtime 和原始文件名，所以同样输入每次压出的字节都不同——可复现构建要传 mtime=0（等价 gzip -n）；尾部的 ISIZE 只有 4 字节会回绕且能被伪造，不能当作「解压后大小」的依据。安全上要防解压炸弹：用 decompress(data, max_length) 或边解边计数，并限制压缩比（超过 100 倍可疑），lzma 还有额外的解压内存风险（最高 1.5GB，用 memlimit）。最后两个默认值差异：Python 的 gzip 默认压缩级别是 9（命令行 gzip 是 6，而 6→9 体积只小 8% 但耗时翻倍），gzip.open() 默认是二进制模式（要文本得写 "rt" 并给 encoding）。