春江暮客

春江暮客的个人学习分享网站

用 Python 生成可复现的 gzip:固定时间戳、移除文件名并验证哈希

2026-10-24 技术
用 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 上验证。

  1. 创建新的演示目录,后续命令在同一目录执行。重复实验时换一个目录名:
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
  1. 保留打印出的 Python 与 zlib 运行时版本,便于比较不同机器的结果。zlib 官方文档解释了运行时版本常量。
  2. 将下一节保存为 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. 执行并验证产物

  1. 查看帮助,再生成两份 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 显示的两个哈希应一致,文件名部分不同。

  1. 检查头部、压缩字节和完整解压结果:
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. 验证失败与边界

  1. 再次写入已有目标,使用 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。

  1. 将下列代码保存为 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,先固定时间戳、文件名策略和压缩等级,再固定并记录运行环境。比较压缩字节,同时完整解压检查原始内容。把拒绝覆盖和失败退出接入自动化任务,避免把半成品当成可复现产物。

标签

1024 12306 ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn chatgpt checkpoint cli cloudflare codex cofoldarena concurrency copy cpu监控 cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging decimal deployment desktop-app devtools disk-space disown dns docker dovecot download electron encoding esm esm2 esm3 esmc esmfold2 faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io flashppi flask folium frontend game generator git github-actions google google-research grep gzip harness hls html http hugo imports indexnow javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics mutation mysql nanobert nanobodies networkx nginx normalize numpy ollama omegatherm pandas password pathlib pdb pep-723 phaser pillow pip postfix preprocessing print protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security selenium seo sequence-identity sha256 shell signals sklearn socket solana somaticsignatures spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tarfile tcp tensorflow tkinter tls toml tron tronpy troubleshooting turtle typesafe-ai ubuntu unicode urllib usdt utf-8 uv vhhbert vite webp wordcloud wordpress workflow yaml zipfile zoneinfo 后台 寓言 概率 经济 贸易 迅雷解析 钱包

友情链接

其它