Python 大日志统计：逐行读取，避免一次性加载整个文件

排查应用日志时，一个常见需求是统计 INFO、WARNING、ERROR 等级的出现次数。文件只有几兆时，readlines() 很方便；日志变大后，把全部行装进列表会占用额外内存。下面用 Python 标准库写一个逐行统计示例，并明确它的适用边界。

1. 先约定日志格式

本文处理的每条记录以大写等级开头，例如：

INFO service started
ERROR request timeout
  traceback detail
WARNING retry scheduled
INFO service stopped

异常堆栈的续行没有等级前缀，因此不单独计为一条 ERROR。这个约定比直接搜索整行是否包含 ERROR 更可靠：业务正文里也可能出现这个单词。

2. 最小实现

from collections import Counter
import re

def count_levels(lines):
    counts = Counter()
    for line in lines:
        match = re.match(r'^(DEBUG|INFO|WARNING|ERROR|CRITICAL)\b', line)
        if match:
            counts[match.group(1)] += 1
    return counts

with open('app.log', encoding='utf-8') as log_file:
    counts = count_levels(log_file)

for level in ('DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'):
    print(f'{level}: {counts[level]}')

文件对象可以逐行迭代，不必先 read() 或 readlines()。with 会在离开代码块时关闭文件，包括处理中抛出异常的情况。Counter 为没有出现的等级返回 0，因此输出能保持固定顺序。

正则表达式的 ^ 限制从行首匹配，\b 限制单词边界，避免把 ERRORISH 这样的前缀算成 ERROR。这里没有自动去掉行首空格，因为是否允许缩进属于日志格式的一部分，应明确决定。

3. 用小样本验证规则

from io import StringIO

sample = (
    'INFO start\nERROR timeout\n  traceback detail\n'
    'WARNING retry\nINFO done\nERRORISH ignored\n'
)
assert count_levels(StringIO(sample)) == Counter(
    INFO=2, ERROR=1, WARNING=1
)
assert count_levels(StringIO('')) == Counter()

这两条断言覆盖了合法等级、空文件、堆栈续行和相似前缀。统计函数接收可迭代的行，而不是负责打开路径，所以测试无需生成真实日志文件。

4. 三个实际边界

第一，逐行读取不代表内存绝对恒定：内存还受单行长度影响，超长单行仍可能占用较多内存。第二，本例固定按 UTF-8 解码，编码错误会显式报错；不要为了“跑完”就默认忽略解码错误，否则结果可能不完整。第三，如果日志以时间戳开头，或者采用 JSON 格式，需要按实际结构解析等级；不要直接套用本例的行首规则。

这种写法适合对已收集的、格式明确的日志做一次离线统计。它不负责实时追踪、轮转文件发现或跨文件去重；需求出现后再增加对应处理即可。示例只读文件，不修改原始日志。

参考资料

• Python 官方教程：文件读写，文档路径 docs.python.org/3/tutorial/inputoutput.html#reading-and-writing-files
• Python 官方文档：Counter，文档路径 docs.python.org/3/library/collections.html#collections.Counter
• Python 官方文档：正则表达式，文档路径 docs.python.org/3/library/re.html

本文由 AI 辅助整理，示例已在本地 Python 环境通过上述断言自检。