日志分析与统计

巡检与排障常需：统计 ERROR 数量、按 IP/用户聚合、找最近异常栈。Python 标准库足够处理 GB 级以下的文本日志；更大文件用逐行读取，避免 read() 整文件进内存。

---

逐行扫描与计数

#!/usr/bin/env python3
"""统计日志级别、输出 Top N 错误消息。"""
from collections import Counter
from pathlib import Path
import re

LOG_PATH = Path("/var/log/app/application.log")
LEVEL_PATTERN = re.compile(r"\b(ERROR|WARN|INFO|DEBUG)\b")

def count_levels(path: Path) -> Counter:
    levels = Counter()
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            m = LEVEL_PATTERN.search(line)
            if m:
                levels[m.group(1)] += 1
    return levels

def top_error_messages(path: Path, top_n: int = 10) -> list[tuple[str, int]]:
    """简单归一化：去掉时间戳与数字，便于聚合相似错误。"""
    errors = Counter()
    ts_strip = re.compile(r"\d{4}-\d{2}-\d{2}[T \d:.]+")
    num_strip = re.compile(r"\b\d+\b")
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            if "ERROR" not in line:
                continue
            msg = ts_strip.sub("", line)
            msg = num_strip.sub("N", msg).strip()
            errors[msg[:200]] += 1  # 截断过长行
    return errors.most_common(top_n)

if __name__ == "__main__":
    if not LOG_PATH.exists():
        print(f"文件不存在: {LOG_PATH}")
    else:
        print("级别统计:", dict(count_levels(LOG_PATH)))
        for msg, cnt in top_error_messages(LOG_PATH):
            print(f"{cnt:5d}  {msg[:80]}...")

---

时间窗口过滤

from datetime import datetime, timedelta
import re

# 假设格式: 2026-07-23 14:30:01,123 ERROR ...
TS_RE = re.compile(r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})")

def errors_in_last_hours(path: Path, hours: int = 1) -> int:
    cutoff = datetime.now() - timedelta(hours=hours)
    count = 0
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            if "ERROR" not in line:
                continue
            m = TS_RE.match(line)
            if not m:
                continue
            ts = datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S")
            if ts >= cutoff:
                count += 1
    return count

---

多文件与 glob

from pathlib import Path

def scan_log_dir(log_dir: Path, pattern: str = "*.log") -> dict[str, int]:
    """每个文件的 ERROR 行数。"""
    summary = {}
    for log_file in sorted(log_dir.glob(pattern)):
        if not log_file.is_file():
            continue
        err = sum(1 for line in log_file.open(errors="replace") if "ERROR" in line)
        summary[log_file.name] = err
    return summary

---

结合 journalctl（systemd 日志）

import subprocess
import json

def journal_errors_since(unit: str, since: str = "1 hour ago") -> list[dict]:
    """
    since 格式同 journalctl，如 '2026-07-23 00:00:00'、'1 hour ago'。
    """
    cmd = [
        "journalctl", "-u", unit,
        "--since", since,
        "-p", "err",  # priority >= err
        "-o", "json",
        "--no-pager",
    ]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
    if r.returncode != 0:
        raise RuntimeError(r.stderr)
    entries = []
    for line in r.stdout.splitlines():
        line = line.strip()
        if line:
            entries.append(json.loads(line))
    return entries

技巧	说明
errors="replace"	二进制或乱码行不中断脚本
逐行读取	大文件内存友好
阈值告警	errors_in_last_hours 超阈值则 exit 1

提示

nginx/access 日志可结合第二阶段做 IP 统计；K8s 容器日志优先 kubectl logs 或 Loki API。