自动化审计日志巡检Python实战 - 第2节

## 2. Python 基础回顾：文件操作与日志处理 做审计日志巡检，你天天要和文件打交道。日志文件怎么读？怎么解析？怎么保证不出错？这些都绕不开 Python 的文件操作和日志模块。这节课我把常用的东西拆开来讲，有的地方我自己踩过坑，会特别提醒你。

2.1 文件操作：你的基本功

2.1.1 打开与关闭文件

Python 内置的 `open()` 函数是起点。它的第一个参数是路径，第二个是模式。我个人习惯把模式记成一组字母：

• 'r' — 只读（默认）
• 'w' — 写，会清空原文件
• 'a' — 追加，在末尾写
• 'b' — 二进制模式，常和上面组合，如 'rb'
• 'x' — 排他创建，文件存在就报错

有个细节我提醒你：**千万别忘了关文件**。以前我见过一个运维脚本，跑着跑着就报“too many open files”，其实就是 `open()` 之后没调用 `close()`。虽然 Python 会在垃圾回收时自动关，但时机不确定，尤其在循环里，文件描述符很容易泄漏。

2.1.2 上下文管理器：with 语句

`with` 会自动释放资源，你只管用。举个例子：

with open('audit.log', 'r', encoding='utf-8') as f:
    content = f.read()

这段代码执行完，文件自动关闭。哪怕中间抛了异常，它也会关。我在项目中测试过，用 `with` 后，再也没出现过文件描述符泄漏的问题。为什么推荐它？因为你写代码时少想一件事，就少一个出错的机会。

2.1.3 读取方式的差异

读取日志文件常用三种方法：`read()`、`readline()`、`readlines()`。我给你列个对比表：

方法	返回值	适用场景
read()	整个字符串	小文件，一次性处理
readline()	一行字符串	逐行处理，内存友好
readlines()	列表（每行一个元素）	需要索引访问的行，但占用内存较大

我个人习惯用 `for line in f:` 逐行遍历，既简洁又不会一次把文件全部加载到内存。比如审计日志几 GB 是常事，`read()` 直接爆内存。你想想看，生产环境日志动辄几百 MB，逐行读最稳妥。

with open('audit.log', 'r', encoding='utf-8') as f:
    for line in f:
        # 处理每一行
        process_line(line)

2.1.4 写入与追加

写文件时最怕什么？路径不存在，或者权限不够。我建议先检查路径，再写。一个实用写法：

import os

log_dir = '/var/log/audit'
if not os.path.exists(log_dir):
    os.makedirs(log_dir, exist_ok=True)

with open(f'{log_dir}/check.log', 'a', encoding='utf-8') as f:
    f.write(f'{datetime.now()} - 巡检通过\n')

这里用 `'a'` 模式追加，不会覆盖之前的记录。`exist_ok=True` 是 Python 3.2 之后新增的参数，可以避免多线程时“文件夹已存在”的异常。

2.1.5 编码与异常处理

**编码问题**：日志文件很可能不是 UTF-8，比如旧系统用 GBK。打开时指定 encoding 是专业做法。我建议统一用 UTF-8 输出，但读取时要灵活。

try:
    with open('audit.log', 'r', encoding='utf-8') as f:
        data = f.read()
except UnicodeDecodeError:
    with open('audit.log', 'r', encoding='gbk') as f:
        data = f.read()

**异常处理**：常见的文件异常有 `FileNotFoundError`、`PermissionError`、`IsADirectoryError`。我曾经一个巡检脚本，因为日志文件被管理员换成了目录，直接崩溃。后来加了异常捕获：

try:
    with open('audit.log', 'r') as f:
        ...
except FileNotFoundError:
    print('日志文件不存在，跳过')
except PermissionError:
    print('权限不足，跳过')

2.2 日志处理：从原始文件到结构化数据

2.2.1 日志的常见格式

审计日志一般有固定格式，比如：

• CSV：字段用逗号分隔，第一行是列名
• JSON：每行一个 JSON 对象，方便解析
• 固定宽度：每个字段占固定字符数，需要切片
• Syslog 格式：时间 + 主机 + 级别 + 消息

在实际项目中，日志格式千奇百怪。我记得有一次接手一个老系统，它的日志是“|”分隔的，注释行还以“#”开头。解析时得先跳过注释行，再按分隔符拆分。这种场景很考验正则和字符串处理能力。

2.2.2 使用正则提取字段

Python 的 `re` 模块是日志解析的核心。比如解析一个典型的审计行：

2025-03-21 14:30:00, INFO, user001, login success

你可以这样写：

import re

pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}), (\w+), (\w+), (.+)'
line = '2025-03-21 14:30:00, INFO, user001, login success'
match = re.match(pattern, line)
if match:
    timestamp, level, user, msg = match.groups()
    print(timestamp, level, user, msg)

我建议把正则表达式提前编译，提高效率：

compiled = re.compile(pattern)
with open('audit.log') as f:
    for line in f:
        m = compiled.match(line)
        if m:
            # 处理匹配的行

2.2.3 使用 csv 模块

如果日志是 CSV 格式，别自己 split，用标准库 `csv` 更安全，它处理引号、转义等边界情况。

import csv
with open('audit.csv', 'r') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row['username'], row['action'])

`DictReader` 会把第一行当作列名，后面每行转成字典，用起来非常方便。

2.2.4 日志的合并与去重

有时候审计日志分散在多个文件中，需要合并分析。简单做法是：

import glob
log_files = glob.glob('/var/log/audit/*.log')
all_lines = []
for fpath in log_files:
    with open(fpath, 'r') as f:
        all_lines.extend(f.readlines())
# 然后排序、去重等操作

不过小心内存，如果文件太大，建议用 `heapq.merge` 做外部归并。我再提个避坑点：**不同文件的时间戳可能不是严格递增**，合并前最好先按时间排序，否则生成的顺序是乱的。我曾经因为忘记排序，分析出的结果完全对不上。

2.3 Python 的 logging 模块：自己写日志

做审计巡检时，你不仅要读别人的日志，也要写自己的巡检日志。`logging` 模块比 `print` 强太多了：可以控制级别、输出到文件、滚动备份。

2.3.1 基本配置

最简单的配置：

import logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='check.log'
)
logging.info('巡检开始')
logging.error('发现了异常记录')

`basicConfig` 只能在第一次调用时生效，后面的调用不会覆盖。我建议程序入口处做一次完整配置。

2.3.2 日志级别

Python 定义了六个级别（从低到高）：

级别	数值	典型场景
DEBUG	10	调试信息，开发时用
INFO	20	确认程序正常运行
WARNING	30	可能有问题，但不影响当前任务
ERROR	40	功能出错，需干预
CRITICAL	50	严重错误，程序可能崩溃

实际巡检中，我一般用 INFO 记录正常操作，WARNING 记录阈值接近报警的情况，ERROR 记录读取失败或数据不一致。这样事后排查时，日志清晰得不得了。

2.3.3 自定义格式

默认格式不够详细。我常用的格式字符串：

format='%(asctime)s [%(levelname)s] %(name)s - %(message)s (%(filename)s:%(lineno)d)'

它会输出时间、级别、记录器名称、消息、文件名和行号。定位问题时，行号是救命稻草。你会发现，加上这些信息后，日志文件完全可以当调试报告用。

2.3.4 文件滚动

如果不限制文件大小，日志会越写越大，直到把磁盘撑爆。用 `RotatingFileHandler` 可以按大小滚动：

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    'audit.log', maxBytes=10*1024*1024, backupCount=5
)
handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))

logger = logging.getLogger('audit')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

`maxBytes` 是 10 MB，超过后自动重命名，保留 5 个备份。还有一个 `TimedRotatingFileHandler`，按时间滚动，比如每天一个文件。

2.3.5 同时输出到控制台和文件

开发时想看实时输出，生产环境又需要文件存档。可以加多个 handler：

import sys

console = logging.StreamHandler(sys.stdout)
console.setLevel(logging.DEBUG)
file_handler = RotatingFileHandler('check.log', maxBytes=5*1024*1024, backupCount=3)
file_handler.setLevel(logging.INFO)

logger = logging.getLogger('')
logger.addHandler(console)
logger.addHandler(file_handler)

这样控制台能看到 DEBUG 信息，日志文件只记录 INFO 以上。我一般调试完把控制台级别调高，避免刷屏。

2.4 本章知识脉络图

下面这张图把文件操作和日志处理的核心知识点串在一起，你看一眼就能回忆起整体结构。

2.5 避坑总结

我把刚才提到的常见坑再整理一遍，你写代码时对照看看：

• 文件用 with 打开，别用裸 open 然后手工 close
• 指定 encoding，尤其在不同操作系统间转移时
• 日志解析用正则或 csv 模块，别简单的 split，字段里可能有逗号
• logging 配置只做一次，多模块共享 logger 时注意重复添加 Handler 的问题
• 别忘了设置日志滚动，否则磁盘警报会让你半夜起床