Python：业务服务监控（文件差异 + 告警）

目录

• 1. difflib 文本差异
  • 1.1 差异符号说明
  • 1.2 行级差异
  • 1.3 生成 HTML 报告
  • 1.4 统一格式 diff（适合 patch / 邮件）
• 2. 用 hashlib 替代逐行 diff
• 3. 配置文件监控实战：Nginx
• 4. 实时文件监控：watchdog
• 5. 告警：把监控结果发出去
  • 5.1 邮件
  • 5.2 钉钉 / 飞书 / Slack Webhook
  • 5.3 整合：一个“检查 → 报告 → 告警”函数
• 6. 监控体系搭建建议
• 7. 常见问题

---

业务监控的常见需求：

• 配置漂移检测：Nginx / 业务配置被改后，对比新旧版本并告警。
• 文件变更监控：日志/配置/数据文件被改时，立即感知。
• 接口 / 服务可用性：定时检查健康端点（参见 检查 URL）。

本章聚焦前两类：使用标准库 difflib + hashlib 完成差异检测与告警。

---

1. difflib 文本差异

1.1 差异符号说明

difflib.Differ 返回的每一行带有 2 字符的“差异前缀”：

符号	含义
-	仅在第一份序列
+	仅在第二份序列
	两份序列共有
?	行内差异标记（具体到字符）
^	差异字符所在位置（用于 ? 行）

1.2 行级差异

import difflib

text1 = """\
This module provides classes and functions for comparing sequences.
including HTML and context and unified diffs.
difflib document v7.4
add string
"""
text2 = """\
This module provides classes and functions for Comparing sequences.
including HTML and context and unified diffs.
difflib document v7.5
"""

diff = difflib.Differ().compare(text1.splitlines(), text2.splitlines())
print("\n".join(diff))

1.3 生成 HTML 报告

from pathlib import Path
import difflib


def generate_diff_html(file1: Path, file2: Path, out: Path) -> int:
    """返回差异行数（0 表示相同）。"""
    a = file1.read_text(encoding="utf-8").splitlines(keepends=True)
    b = file2.read_text(encoding="utf-8").splitlines(keepends=True)

    html = difflib.HtmlDiff().make_file(a, b, fromdesc=file1.name, todesc=file2.name)
    out.write_text(html, encoding="utf-8")

    # 简单粗略估计差异行数
    changes = sum(1 for line in difflib.unified_diff(a, b) if line.startswith(("+ ", "- ")))
    return changes

keepends=True 让 HtmlDiff 保留行尾换行符，渲染更接近 IDE。

1.4 统一格式 diff（适合 patch / 邮件）

import difflib

a = "one\ntwo\nthree\n".splitlines(keepends=True)
b = "one\nTWO\nthree\n".splitlines(keepends=True)

print("".join(difflib.unified_diff(a, b, fromfile="a", tofile="b", n=1)))

---

2. 用 hashlib 替代逐行 diff

如果只是想知道 变了多少 / 变了没，不需要看具体内容，用哈希更快：

import hashlib
from pathlib import Path


def file_signature(path: Path, algo: str = "sha256") -> str:
    h = hashlib.new(algo)
    h.update(path.read_bytes())
    return h.hexdigest()

把“文件指纹”存到本地，巡检时只比较指纹。指纹不同再走 difflib 出报告。

---

3. 配置文件监控实战：Nginx

from __future__ import annotations

import difflib
import shutil
from datetime import datetime
from pathlib import Path


CONFIG = Path("/etc/nginx/nginx.conf")
BACKUP = Path("/tmp/nginx.conf.bak")
REPORT_DIR = Path("./reports")


def diff_nginx_config() -> Path | None:
    """和上次备份对比，若有差异则输出 HTML 报告。"""
    if not CONFIG.exists():
        raise FileNotFoundError(CONFIG)

    REPORT_DIR.mkdir(exist_ok=True)
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    out_file = REPORT_DIR / f"nginx-diff-{stamp}.html"

    if BACKUP.exists():
        a = BACKUP.read_text(encoding="utf-8").splitlines(keepends=True)
        b = CONFIG.read_text(encoding="utf-8").splitlines(keepends=True)

        diff_iter = difflib.unified_diff(a, b, fromfile="backup", tofile="current")
        changes = [line for line in diff_iter if line.startswith(("+", "-"))
                   and not line.startswith(("+++", "---"))]

        if not changes:
            print("nginx 配置无变化")
            return None

        html = difflib.HtmlDiff().make_file(a, b, fromdesc="backup", todesc="current")
        out_file.write_text(html, encoding="utf-8")
        print(f"nginx 配置有变化，报告：{out_file}")
        return out_file

    # 没有备份就直接拷贝一份
    shutil.copy2(CONFIG, BACKUP)
    print("首次运行，已生成备份")
    return None

⚠️ 真实线上读取 /etc/nginx/nginx.conf 通常需要 root 权限。可以用 Watchdog
库做实时监控，避免每次轮询。

---

4. 实时文件监控：watchdog

pip install watchdog

import time
from pathlib import Path
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer


class ChangeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.is_directory:
            return
        print(f"[modified] {event.src_path} @ {time.strftime('%H:%M:%S')}")

    def on_created(self, event):
        if event.is_directory:
            return
        print(f"[created]  {event.src_path}")


def watch(path: str | Path) -> None:
    obs = Observer()
    obs.schedule(ChangeHandler(), str(path), recursive=True)
    obs.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        obs.stop()
    obs.join()


if __name__ == "__main__":
    watch("./config")

---

5. 告警：把监控结果发出去

5.1 邮件

import smtplib
from email.message import EmailMessage


def send_email(subject: str, body: str, to: str) -> None:
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = "monitor@example.com"
    msg["To"] = to
    msg.set_content(body)

    with smtplib.SMTP("smtp.example.com", 587) as s:
        s.starttls()
        s.login("user", "pass")
        s.send_message(msg)

5.2 钉钉 / 飞书 / Slack Webhook

import requests


def send_dingtalk(text: str, webhook: str) -> None:
    requests.post(webhook, json={"msgtype": "text", "text": {"content": text}}, timeout=5)

5.3 整合：一个“检查 → 报告 → 告警”函数

def monitor():
    report = diff_nginx_config()
    if report:
        send_dingtalk(f"nginx 配置有变更：{report}", WEBHOOK_URL)

---

6. 监控体系搭建建议

维度	建议
巡检频率	配置类 5–10 分钟；日志类实时（watchdog）
报告存储	报告 HTML 归档到 reports/，文件名带时间戳
阈值	同一文件 1 小时变更 > N 次 → 高优先级告警
告警去重	同一变更 5 分钟内不重复推送
权限	读配置：sudoers 受控；告警 webhook：单独 token
安全	报告里不要带明文密钥 / cookie
可观测性	把每次巡检结果（耗时、变更条数）打点到 metrics 系统

---

7. 常见问题

• HtmlDiff 中文乱码？ 给模板注入 CSS：overflow-x: auto; white-space: pre;。
• 大文件 diff 慢？ 优先用哈希对比“变了没”，只有变更后才走 difflib。
• /etc/nginx/nginx.conf 没权限？ 监听 inotify（watchdog + inotify_simple）
  或把配置以只读权限提供给监控用户。
• 告警风暴？ 加去重窗口（5 分钟内同源不重复推送）。
• 想对接 Prometheus？ 暴露 /metrics 端点，把巡检结果写成 Gauge / Counter
  （用 prometheus_client）。