用 Python 流式统计 gzip 日志：检查截断、CRC 错误与退出码

历史日志已经压缩成 task.log.gz，为了统计错误行，先解压又会占用额外磁盘。更麻烦的是，复制中断的压缩文件可能还能读出开头几行，让脚本误以为统计成功。

本文面向初学者到中级读者，用 Python 标准库直接逐行读取 gzip 日志、统计指定文本，并验证截断和校验错误时的退出码。适合已经写完、读取期间不会变化的日志备份。

1. 方法：直接读压缩文件，读完再输出

Python 官方 gzip 文档说明，gzip.open() 支持文本模式及编码参数。这里使用 rt 和严格解码；压缩错误、文件错误或编码错误都让任务失败。

处理顺序是：检查 gzip 文件头 → 逐行解压和统计 → 读到末尾并关闭 → 输出 JSON。不要匹配到第一条错误就停止，否则未读取部分的问题可能没有被发现。文件头检查也不能代替完整读取。

已有的 rg 与 Python 日志排障教程介绍了日志筛选和结构化统计。本篇补充直接读取压缩备份的方法；--contains ERROR 只是区分大小写的字面子串匹配，不解析日志级别或 HTTP 状态码。

2. 准备环境和测试日志

需要 Python 3.9 或更新版本，无第三方依赖。示例已在 Linux、Python 3.11.4 下测试；命令可在 bash 或 zsh 运行。

创建新的演示目录，后续命令都在该目录运行。重复实验时换一个目录名：

python3 --version
mkdir gzip-log-demo
cd gzip-log-demo
python3 - <<'PY_DATA'
import gzip
text = "INFO start\nERROR timeout\nINFO retry\nERROR failed"
with gzip.open("task.log.gz", "wt", encoding="utf-8", newline="\n") as file:
    file.write(text)
PY_DATA

测试日志有四行，最后一行没有换行符。将下一节保存为 summarize_gzip.py，也可下载完整脚本。

3. 完整实现

"""Count lines in a completed gzip log; emit JSON only after a full read."""
import argparse
import gzip
import json
from pathlib import Path
import sys
import zlib


def summarize(path, contains, encoding):
    lines = matches = 0
    with path.open("rb") as raw:
        if raw.read(2) != b"\x1f\x8b":
            raise gzip.BadGzipFile("missing gzip header")
        raw.seek(0)
        with gzip.open(raw, "rt", encoding=encoding, errors="strict") as stream:
            for line in stream:
                lines += 1
                if contains is None or contains in line:
                    matches += 1
    return {"file": str(path), "lines": lines, "matches": matches}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("file", type=Path, help="completed .gz text log")
    parser.add_argument("--contains", help="case-sensitive literal substring")
    parser.add_argument("--encoding", default="utf-8", help="text encoding (default: utf-8)")
    args = parser.parse_args()
    if args.contains == "":
        parser.error("contains must not be empty")
    try:
        result = summarize(args.file, args.contains, args.encoding)
    except (OSError, EOFError, zlib.error, UnicodeError, LookupError) as exc:
        print(f"ERROR: {args.file}: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

代码先检查两个 gzip 标识字节，再把同一个文件对象交给解压器。统计过程只保留计数，不调用无参数的 read()，也不把全部行放进列表；内存仍受最长单行和解压缓冲影响，不是固定大小的内存上限。

BadGzipFile 继承自 OSError，因此 OSError 分支同时处理格式错误和文件访问错误。无效压缩数据还可能触发 EOFError 或 zlib.error，代码也会捕获。编码名称无效或严格解码失败，同样返回 1。

4. 运行并验证正常输出

查看帮助，分别统计错误、全部行和不存在的文本：

python3 summarize_gzip.py --help
python3 summarize_gzip.py task.log.gz --contains ERROR
python3 summarize_gzip.py task.log.gz
python3 summarize_gzip.py task.log.gz --contains WARNING

第二条命令输出：

{"file": "task.log.gz", "lines": 4, "matches": 2}

不指定 --contains 时，lines 和 matches 都是 4；匹配 WARNING 时，matches 是 0，退出码仍为 0。没有匹配是一种正常结果。空白行也计为一行，末行不需要换行符。

检查报告字段：

python3 summarize_gzip.py task.log.gz --contains ERROR > report.json
python3 - <<'PY_CHECK'
import json
from pathlib import Path
result = json.loads(Path("report.json").read_text(encoding="utf-8"))
assert result == {"file": "task.log.gz", "lines": 4, "matches": 2}
print("OK: four lines, two matches")
PY_CHECK

预期打印 OK: four lines, two matches。报告字段来自本次完整读取；它没有证明日志业务内容齐全，也不验证文件来源。

5. 制造截断与 CRC 错误

只修改演示目录里的副本。下面分别删除 gzip 尾部八字节、修改尾部 CRC 的一个字节：

python3 - <<'PY_BAD'
from pathlib import Path
original = Path("task.log.gz").read_bytes()
Path("truncated.log.gz").write_bytes(original[:-8])
corrupt = bytearray(original)
corrupt[-8] ^= 1
Path("bad-crc.log.gz").write_bytes(corrupt)
PY_BAD
python3 summarize_gzip.py truncated.log.gz --contains ERROR > failed.json
printf 'exit=%s\n' "$?"
wc -c < failed.json
python3 summarize_gzip.py bad-crc.log.gz --contains ERROR
printf 'exit=%s\n' "$?"

两次运行都应输出 ERROR: 到标准错误，并打印 exit=1。wc 应输出 0，说明失败运行没有向标准输出写成功 JSON。具体异常文本可能随 Python 版本变化，不要依赖完整错误字符串判断成败。

还验证了空内容的有效 gzip、CRLF、Unicode、多个 gzip 成员拼接、非 gzip 文件、缺失文件和错误编码。读取完毕仍不等于所有业务记录都到齐；一次完整压缩了半份日志，也可能拥有正确 CRC。

重定向 > report.json 会在程序启动前清空目标，即使程序后来失败。因此，更新已有报告时先写新的临时路径，成功后再替换：

if python3 summarize_gzip.py task.log.gz --contains ERROR > report.pending.json; then
  mv report.pending.json report.json
else
  rm -f report.pending.json
  printf 'Report update failed\n' >&2
fi

这个片段假定演示目录可信且只有一个运行者。失败时保留已有 report.json；若嵌入任务脚本，还应在失败分支返回非零退出码。更完整的保存方案见 Python 原子写入 JSON。

6. 排错与使用边界

现象	具体处理
`missing gzip header`	确认输入确实是 gzip 文件；改扩展名不会改变格式。
`CRC check failed` 或提前结束	等源文件写完后重新复制；比较源与副本的校验和，不要用部分统计作为完整报告。
`UnicodeDecodeError`	核实日志编码，必要时传 `--encoding gb18030`；不要用忽略解码错误掩盖丢失内容。
`unknown encoding`	修正 `--encoding` 的名称，例如 `utf-8`。
`No such file` 或权限错误	检查文件绝对路径，以及文件和父目录的访问权限。
文件很小却运行很久	查看解压后的数据量和单行长度；压缩体积不能代表读取工作量。

只统计已完成的备份，不扫描正在追加的 gzip 文件。.tar.gz 解压后是归档数据，不能当普通文本日志交给本脚本。日志轮转的生成与保留策略可以接着阅读 Python 日志按大小轮转；该教程的默认处理器不会自动生成 gzip 备份。

7. 小结

直接逐行读取 gzip 日志，可以省去落盘解压这一步，同时保留简单的 JSON 统计接口。把完整读取、严格解码和失败退出码一起验证，再把结果接入自动化任务；更新已有报告时，只在成功后替换文件。