用 Python 读取日志末尾 N 行：deque 窗口、编码与边界验证

日志文件很大，只想给失败报告附上最后 20 行，却写成 readlines()[-20:]，结果把整份文件都装进内存。本文面向初学者到中级读者，用 Python 标准库 deque 保留末尾 N 行，并验证中文、空行和末尾没有换行符时的结果。

示例处理已经写完、读取期间不再变化的普通 UTF-8 日志。对于持续增长的活动日志、FIFO 或设备文件，请先换成稳定的普通文件副本。

1. 方法：逐行读取，只保留末尾窗口

把文件迭代器交给 deque(source, maxlen=N)，队列满后会丢弃最早加入的行。Python 官方 deque 示例也展示了这种末尾窗口用法。

这仍然会从头扫描整个文件，时间随输入长度增长。队列最多保留 N 行，但长行本身可能很大，因此限制的是行数，不是固定字节数；读入超长单行时也需要对应内存。只在终端看大型普通文件的末尾，可以直接用系统 tail -n 20 task.log；下面的方法适合把结果接进已有 Python 自动化。

2. 准备：生成可重复的小日志

需要 Python 3.10 或更新版本，无需安装第三方包。本文命令在 Linux、Python 3.11.4 上验证，shell 示例适用于 bash 和 zsh。

在新目录生成五行日志，最后一行故意不写换行符：

python3 --version
mkdir tail-demo
cd tail-demo
python3 - <<'PYCODE'
from pathlib import Path
Path("task.log").write_text(
    "01 INFO start\n02 INFO loaded\n03 WARNING retry\n"
    "04 INFO recovered\n05 ERROR 连接失败",
    encoding="utf-8",
)
PYCODE

将下一节代码保存为当前目录的 tail_log.py。后续操作都在这个演示目录运行。

3. 实现：完整命令行工具

"""Print the last N lines of a completed UTF-8 log. Python 3.10+."""
import argparse
from collections import deque
from pathlib import Path
import sys


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return number


def tail_lines(path, count):
    if count < 1:
        raise ValueError("count must be at least 1")
    with Path(path).open(encoding="utf-8", errors="strict") as source:
        return deque(source, maxlen=count)


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("path", type=Path, help="completed UTF-8 text log")
    parser.add_argument("-n", "--lines", type=positive_int, default=20)
    args = parser.parse_args()
    try:
        recent = tail_lines(args.path, args.lines)
        sys.stdout.writelines(recent)
        sys.stdout.flush()
    except (OSError, UnicodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


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

脚本先读完再输出。输入路径错误或任意位置出现坏 UTF-8 时，返回 1，标准输出为空，即使坏字节位于最终保留范围之外也一样。命令行参数错误返回 2，成功返回 0；空文件成功且不输出内容。

文件采用明确的 UTF-8 严格解码。Python open 文档说明，默认文本换行处理会把 CRLF 和单独 CR 转为 \n。因此输出保留文本行内容与末尾是否有换行，但不承诺原始换行字节完全相同；这里也不添加时间戳或行号。

writelines() 不会额外插入分隔符。输出写入失败仍可能留下部分结果，所以自动化必须同时检查退出码；重定向到报告文件时选新路径，绝不能重定向到输入日志本身，否则 shell 会提前截断输入。

4. 使用：提取最近三行，再筛选错误

查看帮助、提取末尾三行，并逐字节核对输出：

python3 tail_log.py --help
python3 -X utf8 tail_log.py -n 3 task.log > recent.log
python3 - <<'PYCODE'
from pathlib import Path
expected = "03 WARNING retry\n04 INFO recovered\n05 ERROR 连接失败"
assert Path("recent.log").read_bytes() == expected.encode("utf-8")
print("PASS: last 3 lines, UTF-8, no added newline")
PYCODE

预期输出 PASS: last 3 lines, UTF-8, no added newline。-X utf8 让这个调用的标准输出使用 UTF-8；文件读取编码仍由代码中的 encoding 明确指定。若定时任务里只有中文输出失败，可继续看标准输出编码排障。

需要接入 Python 报告逻辑时，可直接导入函数。把下面代码保存为 recent_errors.py，运行 python3 -X utf8 recent_errors.py：

from tail_log import tail_lines

for line in tail_lines("task.log", 3):
    if "ERROR" in line:
        print(line, end="")

输出为 05 ERROR 连接失败，末尾仍没有换行。这里是先选最后三行，再筛选 ERROR；如果先筛选整份文件再取三行，得到的会是最后三个错误，含义不同。函数读取失败会直接抛出异常，调用方应按自己的任务策略处理。

5. 验证：边界输入与失败退出码

将下面代码保存为 check_tail.py，运行 python3 check_tail.py。测试只在自动清理的临时目录创建数据：

from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory

script = Path("tail_log.py").resolve()


def run(path, count):
    return subprocess.run(
        [sys.executable, "-X", "utf8", str(script), "-n", str(count), str(path)],
        capture_output=True,
        timeout=10,
    )


with TemporaryDirectory(prefix="tail-check-") as workspace:
    folder = Path(workspace)
    cases = [
        (b"", 3, b""),
        (b"one\n", 5, b"one\n"),
        (b"a\nb\nc", 2, b"b\nc"),
        (b"a\n\nb\n", 2, b"\nb\n"),
        ("甲\n乙\n丙".encode("utf-8"), 1, "丙".encode("utf-8")),
        (b"a\r\nb\r\n", 1, b"b\n"),
    ]
    for index, (content, count, expected) in enumerate(cases):
        path = folder / f"case {index}.log"
        path.write_bytes(content)
        result = run(path, count)
        assert (result.returncode, result.stdout, result.stderr) == (0, expected, b"")
        assert path.read_bytes() == content

    bad = folder / "invalid.log"
    bad.write_bytes(b"bad:\xff\nlater valid line\n")
    for path in (bad, folder / "missing.log", folder):
        result = run(path, 1)
        assert result.returncode == 1 and result.stdout == b""
        assert b"ERROR:" in result.stderr

    for count in (0, -1, "oops"):
        result = run(folder / "case 0.log", count)
        assert result.returncode == 2 and result.stdout == b""

print("PASS: content, line endings, input preserved, read errors, bad arguments")

预期输出 PASS: content, line endings, input preserved, read errors, bad arguments。测试包含带空格路径、空文件、行数不足、空行、中文、CRLF，以及位于末尾窗口之前的坏编码。

再运行这个规模稍大的例子，核对留下的两行：

python3 - <<'PYCODE'
from pathlib import Path
from tail_log import tail_lines

path = Path("many.log")
with path.open("x", encoding="utf-8") as output:
    for index in range(100_000):
        output.write(f"row={index}\n")
recent = tail_lines(path, 2)
assert list(recent) == ["row=99998\n", "row=99999\n"]
assert recent.maxlen == 2
print("PASS: retained 2 of 100000 lines")
PYCODE

预期输出 PASS: retained 2 of 100000 lines。many.log 用 x 模式创建，重复运行会拒绝覆盖；换一个新目录再试。这项检查验证保留行数和内容，并非内存性能基准。

6. 排障：读不到、输出错与范围限制

问题	直接处理
`No such file or directory`	检查实际工作目录，自动化中传入日志的绝对路径。
`Permission denied`	用任务运行账户检查文件和父目录权限，选择该账户有权读取的日志。
UTF-8 解码失败	确认生产端的真实编码；明确修改读取编码后重测，不用 `errors="ignore"` 静默丢字。
输出中文时报编码错误	用上面的 `python3 -X utf8` 调用，并检查失败进程的标准流设置。
读取很慢	这个算法要扫描全文件；缩小输入范围，或仅查看文件末尾时使用系统 tail。
输出出现半条业务记录	工具按文本行计数；多行异常栈会被截断，需要完整事件时应采用业务记录解析。

工具不跟踪文件新增内容，也不拼接轮转备份。活动文件在读取中被改写或轮转时，结果不是一致快照；想控制日志体积，先为写入方配置按大小轮转日志。遇到 .gz 备份时，不要用普通文本方式打开，可参考gzip 日志流式读取与完整性检查。

7. 总结

用 deque 保存末尾 N 行，可以避免一次保存整份日志，便于生成失败摘要。先验证行尾、编码和非零退出码，再把函数接进自动化报告。文件越大扫描越慢，单行越长占用越多，选方法时要同时考虑这两个边界。