基于Python编写一个轻量日志监控脚本_Python

前言

服务器运行过程中，磁盘空间不足、程序连接失败或者服务异常，通常都会在系统日志或应用日志中留下记录。但如果没有专门的监控工具，这些信息往往要等到人工登录服务器检查时才会被发现。

对于家庭服务器、测试主机和数量不多的小型服务来说，部署一套完整的日志平台可能有些复杂。更直接的办法，是持续读取关键日志文件，在出现error、failed或者no space left on device等指定关键词时，主动发送一条告警消息。

本文会使用python编写一个轻量日志监控脚本，记录当前读取位置，只处理后续新增的日志内容，并在匹配到异常关键词后，通过钉钉机器人将主机名、时间、日志路径和具体内容发送到群聊。

完成日志告警后，还会将脚本配置为systemd服务，使其能够随系统启动并持续运行。最后通过cpolar映射服务器的ssh端口，收到告警后可以从外部网络远程登录主机，继续检查磁盘、进程和服务状态。

1.测试效果

效果预览

当你的/var/log/syslog或应用日志中出现：

may 08 15:30:01 server cpolar[1234]: failed to write log: no space left on device

你将在钉钉群收到如下消息

【服务器告警】

主机：z4s-vjfl

时间：2026-05-08 15:30:01

日志路径：/var/log/cpolar/access.log.20260508

内容：failed to write log: no space left on device

磁盘空间不足！请立即处理！

通过这个python脚本，你实现了

实时监控任意日志文件
关键词触发精准告警
钉钉消息秒级触达手机
低资源占用，适合嵌入式设备
零依赖复杂框架，纯标准库 + requests
延伸场景：监控nginx错误日志 → 发现攻击行为
监控数据库慢查询日志 → 优化sql
监控docker容器日志 → 异常自动重启

让机器替你“盯屏”，你只管睡觉！

2.获取钉钉webhook和secret

打开钉钉群 → 点击右上角设置：

找到智能群助手 → 添加机器人：

添加自定义机器人：

点击添加：

给机器人起个名字，我这里是“服务器告警”：

设置发消息关键词，因为现在钉钉对安全严格，所以需要设置限制，，也可以设置加签或者ip地址：

点击“加签”，复制生成的秘钥，留着备用：

点击完成后，复制生成的webhook，留着备用：

3.编写python监控脚本

创建要监控的日志文件：

mkdir /var/log/ceshi/
touch /var/log/ceshi/access.log.20260508

创建文件log_monitor.py：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import time
import json
import hmac
import hashlib
import base64
import urllib.parse
from datetime import datetime
from pathlib import path
import requests
# ====== 配置区（请按需修改）======
log_file = "/var/log/ceshi/access.log.20260508"  # 要监控的日志文件
keywords = ["error", "failed", "exception", "no space left on device", "disk full"]
hostname = os.uname().nodename  # 自动获取主机名
# 钉钉机器人配置（从钉钉后台获取）
dingtalk_webhook = "https://oapi.dingtalk.com/robot/send?access_token=dbf63c2e3c2f2dd"
dingtalk_secret = "sec8bb4d908c1039a4"
# ====== 钉钉签名函数 ======
def get_dingtalk_sign():
    timestamp = str(round(time.time() * 1000))
    secret_enc = dingtalk_secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, dingtalk_secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign
# ====== 发送钉钉消息 ======
def send_dingtalk_alert(message):
    timestamp, sign = get_dingtalk_sign()
    webhook_url = f"{dingtalk_webhook}×tamp={timestamp}&sign={sign}"
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "【服务器告警】",
            "text": message
        }
    }
    try:
        response = requests.post(webhook_url, json=data, timeout=10)
        if response.status_code != 200:
            print(f"[!] 钉钉发送失败: {response.text}")
    except exception as e:
        print(f"[!] 钉钉请求异常: {e}")
# ====== 监控日志主函数 ======
def monitor_log():
    log_path = path(log_file)
    if not log_path.exists():
        print(f"[!] 日志文件不存在: {log_file}")
        return
    # 获取文件初始大小（用于断点续读）
    file_size = log_path.stat().st_size
    print(f"[+] 开始监控日志: {log_file} (初始大小: {file_size} bytes)")
    while true:
        try:
            current_size = log_path.stat().st_size
            if current_size < file_size:
                # 日志被轮转（如 logrotate），重置位置
                print("[+] 检测到日志轮转，重新开始读取")
                file_size = 0
            if current_size > file_size:
                with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                    f.seek(file_size)  # 从上次位置读起
                    lines = f.readlines()
                    file_size = current_size  # 更新已读位置
                    for line in lines:
                        line_lower = line.lower()
                        for keyword in keywords:
                            if keyword in line_lower:
                                # 构建告警消息
                                now = datetime.now().strftime("%y-%m-%d %h:%m:%s")
                                msg = (
                                    f"## 🔴 **【服务器告警】**\n\n"
                                    f"- **主机**: `{hostname}`\n"
                                    f"- **时间**: `{now}`\n"
                                    f"- **日志路径**: `{log_file}`\n"
                                    f"- **内容**: `{line.strip()}`\n\n"
                                    f"> ⚠️ **检测到关键词: `{keyword}`**"
                                )
                                print(f"[alert] {line.strip()}")
                                send_dingtalk_alert(msg)
                                break  # 避免重复告警同一行
            time.sleep(1)  # 每秒检查一次
        except keyboardinterrupt:
            print("\n[+] 监控已停止")
            break
        except exception as e:
            print(f"[!] 监控异常: {e}")
            time.sleep(5)
if __name__ == "__main__":
    monitor_log()

4.安装依赖 & 赋予执行权限

进入你的项目目录:

cd /root/ceshi

创建虚拟环境（python 3.6+ 自带 venv）

python3 -m venv venv

激活虚拟环境:

source venv/bin/activate

此时提示符会变成 (venv)，再安装包

pip install requests

5.测试运行

运行你的脚本：

python log_monitor.py

编辑/var/log/ceshi/access.log.20260508监控文件：

vi access.log.20260508

may 08 15:30:01 server cpolar[1234]: failed to write log: no space left on device

钉钉成功发送告警：

6.运行 & 开机自启

方式 1：使用nohup临时运行

nohup python3 /path/to/log_monitor.py > /var/log/log_monitor.log 2>&1 &

方式 2：创建systemd服务（推荐，支持开机自启）

创建服务文件：

sudo vim /etc/systemd/system/log-monitor.service

[unit]
description=log monitor for server alerts
after=network.target

[service]
type=simple
user=root
execstart=/usr/bin/python3 /ceshi/log_monitor.py
restart=always
restartsec=10

[install]
wantedby=multi-user.target

启动服务：

sudo systemctl daemon-reload
sudo systemctl enable --now log-monitor.service
sudo systemctl status log-monitor

在现实运维中，很多开发者、创客甚至中小企业都把关键服务部署在家庭nas、老旧服务器或内网测试机上——它们没有固定公网ip，藏在路由器后面，平时靠cpolar暴露个web界面勉强够用。可一旦半夜磁盘爆满、docker容器崩溃、系统日志里刷出“no space left on device”或“cpolar failed to reconnect”，你却远在千里之外出差、旅游，甚至正在熟睡。这时候，人工巡检毫无意义，而商业监控平台又太重、太贵。

但如果你提前做了两件事：

第一，在服务器上跑一个轻量python脚本，持续扫描 /var/log/cpolar/ 或系统日志，一旦发现关键词就通过钉钉机器人推送告警到手机；

第二，用cpolar配置一条tcp隧道，把本地22端口（ssh）映射到公网，哪怕只是临时分配的地址。

那么当故障发生时，你的手机会立刻收到钉钉消息：“【告警z4s主机磁盘已满！建议通过 cpolar ssh登录清理”。你只需打开终端，输入ssh -p 22222 root@192.xxx.xxx.xxx，就能像坐在机器前一样，删日志、重启服务、查进程——整个过程不到五分钟，无需回家，不用求人。这不仅是技术组合，更是对“运维最后一公里”的真实补全：让每一台沉默的内网设备，都能在危急时刻主动呼救，并为你留一扇随时可进的门。 尤其对于用极空间、树莓派、旧笔记本搭建家庭实验室的人来说，这套方案成本近乎为零，却能极大提升系统的可靠性和你的安心感。