用 Python 生成可复现的 gzip:固定时间戳、移除文件名并验证哈希
重复压缩同一份 CSV,解压后内容没变,.gz 的 SHA-256 却变了。这会让备份去重、构建缓存和产物比较多出无效变更。
本文面向初学者到中级读者,用 Python 标准库固定 gzip 头部信息,流式生成新文件,再检查压缩字节和解压内容。示例适用于已经写完、处理期间不再变化的普通文件。
1. 方法:固定头部,再比较结果
gzip 头部可以携带时间和原文件名。RFC 1952 的字段定义说明,MTIME = 0 表示不提供时间戳,FNAME 标志位决定是否带文件名。下面显式指定不同头部信息,稳定复现“内容相同、压缩字节不同”,无需等待时钟走动:
python3 - <<'PY_HEADER'
import gzip
from io import BytesIO
payload = b"id,value\n1,alpha\n"
def pack(name, stamp):
buffer = BytesIO()
with gzip.GzipFile(filename=name, mode="wb", fileobj=buffer,
mtime=stamp, compresslevel=9) as file:
file.write(payload)
return buffer.getvalue()
a = pack("a.csv", 1700000000)
b = pack("a.csv", 1700000001)
c = pack("b.csv", 1700000000)
assert a != b and a != c
assert gzip.decompress(a) == gzip.decompress(b) == gzip.decompress(c) == payload
print("OK: same content, different gzip bytes")
PY_HEADER
预期输出 OK: same content, different gzip bytes。因此,判断原始内容是否一致,应比较原始字节或解压后的哈希;判断发布产物是否逐字节一致,才比较 .gz 本身。可以结合数据集 SHA-256 清单教程记录输入版本。
本文固定三个参数:filename=""、mtime=0、compresslevel=9。可复现范围限定为相同输入字节、相同脚本和 Python/zlib 环境;升级压缩实现后,应重新测试,不能把它当成跨版本的永久哈希约定。
2. 准备环境和输入
需要 Python 3.9+ 和已有的 gzip、zlib 标准库模块,无第三方依赖。下列命令适用于 Linux 的 bash 或 zsh;示例在 Python 3.11.4 上验证。
- 创建新的演示目录,后续命令在同一目录执行。重复实验时换一个目录名:
python3 --version
python3 -c 'import platform, zlib; print(platform.python_version(), zlib.ZLIB_RUNTIME_VERSION)'
mkdir reproducible-gzip-demo
cd reproducible-gzip-demo
printf 'id,value\n1,alpha\n2,beta\n' > first.csv
cp first.csv second.csv
- 保留打印出的 Python 与 zlib 运行时版本,便于比较不同机器的结果。zlib 官方文档解释了运行时版本常量。
- 将下一节保存为
reproducible_gzip.py。两个输入文件内容相同,但名称不同,用于验证文件名不会进入产物。
3. 完整实现:流式压缩并拒绝覆盖
"""Create a reproducible gzip in a new file. Python 3.9+."""
import argparse
import gzip
from pathlib import Path
import shutil
import sys
import zlib
def compress_file(source, target):
created = False
try:
with source.open("rb") as src:
with target.open("xb") as raw:
created = True
with gzip.GzipFile(
filename="", mode="wb", fileobj=raw,
mtime=0, compresslevel=9,
) as packed:
shutil.copyfileobj(src, packed, length=1024 * 1024)
except BaseException:
if created:
target.unlink(missing_ok=True)
raise
def main():
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("source", type=Path)
parser.add_argument("target", type=Path)
args = parser.parse_args()
try:
compress_file(args.source, args.target)
except (OSError, zlib.error) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
print("OK: gzip complete")
return 0
if __name__ == "__main__":
raise SystemExit(main())
fileobj 指向目标文件,空 filename 防止把目标名称写入头部。mtime=0 去掉时间变化。Python 的 GzipFile 文档也说明,关闭 gzip 包装层不会关闭传入的文件对象,所以代码用了两层 with。
shutil.copyfileobj按 1 MiB 分块复制,不会先把整个输入读进内存。xb 独占创建目标文件,目标已存在时直接失败;源文件保持原样。Python 捕获到异常或 Ctrl+C 时,会尝试删除本次创建的未完成输出,不会删除原先已存在的目标。
这里假设目录可信,运行期间没有其他进程替换目标。压缩尚未结束时目标文件已经可见;强杀或断电仍可能留下半成品。需要向其他进程交付结果时,可参考原子写入教程,在同目录暂存,成功关闭后再发布。
4. 执行并验证产物
- 查看帮助,再生成两份 gzip:
python3 reproducible_gzip.py --help
python3 reproducible_gzip.py first.csv first.gz
python3 reproducible_gzip.py second.csv second.gz
sha256sum first.gz second.gz
两次压缩各输出 OK: gzip complete;sha256sum 显示的两个哈希应一致,文件名部分不同。
- 检查头部、压缩字节和完整解压结果:
python3 - <<'PY_CHECK'
import gzip
import hashlib
from pathlib import Path
first = Path("first.gz").read_bytes()
second = Path("second.gz").read_bytes()
assert first == second
assert first[:3] == b"\x1f\x8b\x08"
assert first[3] & 0x08 == 0 # No FNAME field.
assert first[4:8] == b"\x00" * 4 # MTIME = 0.
assert gzip.decompress(first) == Path("first.csv").read_bytes()
assert gzip.decompress(second) == Path("second.csv").read_bytes()
print("OK: equal gzip bytes, zero timestamp, no filename, exact round trip")
print("sha256:", hashlib.sha256(first).hexdigest())
PY_CHECK
头部断言适用于本脚本生成的 gzip;它不是任意 gzip 文件的完整校验器。gzip.decompress() 在这里读取很小的演示文件;大型日志的完整读取可参考gzip 日志完整性检查。
两种校验都要保留:相同的坏产物也可能有相同哈希,而只验证解压内容又发现不了头部造成的无效变更。
5. 验证失败与边界
- 再次写入已有目标,使用
if捕获失败,兼容启用了set -e的脚本:
if python3 reproducible_gzip.py first.csv first.gz; then
printf 'UNEXPECTED: existing output was accepted\n' >&2
else
printf 'exit=%s\n' "$?"
fi
预期 stderr 包含 ERROR: 和文件已存在的原因,随后输出 exit=1,原 first.gz 保持不变。正常执行返回 0,文件或压缩错误返回 1,命令行参数错误返回 2。
- 将下列代码保存为
check_gzip.py,执行python3 check_gzip.py。它使用自动清理的临时目录,验证空文件、UTF-8、多块二进制、重复执行、输入输出同路径和缺失输入:
"""Run beside reproducible_gzip.py, using only disposable test files."""
import gzip
from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory
script = Path("reproducible_gzip.py").resolve()
def run(source, target):
return subprocess.run(
[sys.executable, str(script), str(source), str(target)],
capture_output=True, text=True, timeout=30,
)
with TemporaryDirectory(prefix="gzip-check-") as directory:
root = Path(directory)
for number, payload in enumerate((b"", "中文\n".encode("utf-8"),
bytes(range(256)) * 9000)):
source = root / f"input {number}.bin"
target = root / f"output {number}.gz"
other = root / f"other {number}.gz"
source.write_bytes(payload)
assert run(source, target).returncode == 0
assert run(source, other).returncode == 0
saved = target.read_bytes()
assert saved == other.read_bytes()
assert gzip.decompress(saved) == payload
assert source.read_bytes() == payload
result = run(source, target)
assert result.returncode == 1 and not result.stdout
assert target.read_bytes() == saved
result = run(source, source)
assert result.returncode == 1
assert source.read_bytes() == payload
missing = root / "missing.bin"
output = root / "missing.gz"
result = run(missing, output)
assert result.returncode == 1 and not output.exists()
assert "ERROR:" in result.stderr
print("OK: empty, Unicode, multi-chunk binary, overwrite and missing-input checks")
预期输出最后一行 OK: empty, Unicode, multi-chunk binary, overwrite and missing-input checks。这些测试同时检查输入未被修改,以及已存在目标不会被覆盖。
6. 常见问题与修复
| 现象 | 具体修复 |
|---|---|
| 解压内容相同,哈希不同 | 检查是否遗漏 filename="" 或 mtime=0,并核对压缩等级和 Python/zlib 版本。 |
| 升级 Python 后结果变化 | 固定显式参数并重跑上述测试;需要长期字节一致时固定运行环境。 |
File exists |
换一个新的目标名;确认不再需要旧产物后再由调用者处理,不要把代码改成静默覆盖。 |
No such file or directory |
检查源路径和目标父目录;本工具不会创建父目录。 |
Permission denied 或磁盘空间不足 |
修复目标目录权限或释放空间,再使用新目标重试;先核查残留半成品。 |
| 解压内容本身不同 | 检查输入生成过程、编码、换行和行顺序;gzip 不会把语义相同的文本规范成相同字节。 |
不要依赖某个 Python 版本的默认参数。当前 gzip 文档的版本记录记录了 gzip.compress() 在 3.14 将默认 mtime 改为 0,以及 3.15 将默认压缩等级改为 6。本文使用流式 GzipFile 并显式设置全部三个参数。
总结
让相同输入生成稳定的 gzip,先固定时间戳、文件名策略和压缩等级,再固定并记录运行环境。比较压缩字节,同时完整解压检查原始内容。把拒绝覆盖和失败退出接入自动化任务,避免把半成品当成可复现产物。
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/python-reproducible-gzip.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。