60行Python代码编写磁盘自动监控脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
disk_inspect.py —— 每天自动巡检磁盘 + 清理过期日志
全程标准库，无需安装任何第三方包。

用法：
    python3 disk_inspect.py            # 默认 LIST 模式，只打印待删清单，不真删
    python3 disk_inspect.py --apply    # 确认无误后真删过期日志
    python3 disk_inspect.py --help     # 查看帮助
"""

import os
import sys
import time
import shutil
import argparse
from pathlib import Path

# ===== 可配置项：改这一块就够了 =====
PARTITIONS = ["/", "/data", "/var"]          # 要巡检的挂载点，按机器实际情况增删
WARN_THRESHOLD = 80                          # 使用率 >= 80% 记 WARN
CRIT_THRESHOLD = 90                          # 使用率 >= 90% 记 CRIT，并让退出码 = 1
LOG_DIR = "/var/log/app"                     # 要清理的日志目录
LOG_PATTERN = "*.log"                        # 只清 .log 文件，通配符宁窄勿宽
LOG_RETAIN_DAYS = 7                          # 保留最近 7 天，更早的视为过期
REPORT_FILE = "/var/log/disk_report.txt"     # 报告输出位置（不推送，想看时 cat）


def check_disk(partitions, warn, crit):
    """查每个分区的使用率，按阈值分级为 OK / WARN / CRIT。

    关键细节：用 used / total 算百分比，别拿 free 反推——
    部分文件系统的 free 含可回收缓存，和 used 算出来的比例能差三五个点，
    报警阈值会对不上。
    """
    results = []
    for mount in partitions:
        try:
            usage = shutil.disk_usage(mount)
        except OSError:
            # 分区不存在或无法访问，记一行 UNKNOWN，不影响其它分区巡检
            results.append((mount, None, "UNKNOWN"))
            continue
        percent = usage.used / usage.total * 100
        if percent >= crit:
            level = "CRIT"
        elif percent >= warn:
            level = "WARN"
        else:
            level = "OK"
        results.append((mount, percent, level))
    return results


def clean_logs(log_dir, pattern, retain_days, apply):
    """按文件修改时间筛过期日志。

    apply=False（默认）只打印待删清单和能腾出的空间，不真删；
    apply=True 才真删。第一次跑务必先空跑确认，避免通配符写宽误删归档日志。
    """
    cutoff = time.time() - retain_days * 86400   # 过期时间戳，早于此的删
    log_path = Path(log_dir)
    if not log_path.is_dir():
        return [], 0
    to_delete = []
    freed = 0
    for f in log_path.glob(pattern):
        try:
            st = f.stat()
        except OSError:
            continue
        if st.st_mtime < cutoff:
            to_delete.append(f)
            freed += st.st_size
    if apply:
        for f in to_delete:
            try:
                f.unlink()
            except OSError:
                pass
    return to_delete, freed


def main():
    parser = argparse.ArgumentParser(description="磁盘巡检 + 日志清理")
    parser.add_argument("--apply", action="store_true",
                        help="真删过期日志；不加此参数只打印清单（LIST 模式）")
    args = parser.parse_args()

    # 1) 磁盘水位
    disk_results = check_disk(PARTITIONS, WARN_THRESHOLD, CRIT_THRESHOLD)

    # 2) 清日志（默认 LIST 模式，--apply 才真删）
    deleted_files, freed_bytes = clean_logs(
        LOG_DIR, LOG_PATTERN, LOG_RETAIN_DAYS, args.apply)

    # 3) 汇总成报告：每分区一行「挂载点：使用率 级别」，方便 grep CRIT
    lines = ["Disk Inspect Report - " + time.strftime("%Y-%m-%d %H:%M:%S"), ""]
    has_crit = False
    for mount, percent, level in disk_results:
        if level == "UNKNOWN":
            lines.append(f"{mount}: UNKNOWN (无法访问)")
        else:
            lines.append(f"{mount}: {percent:.1f}% {level}")
        if level == "CRIT":
            has_crit = True
    lines.append("")
    action = "已删除" if args.apply else "待删除(LIST 模式)"
    lines.append(f"日志清理: {action} {len(deleted_files)} 个文件, "
                 f"腾出 {freed_bytes / 1024 / 1024:.1f} MB")
    if not args.apply and deleted_files:
        lines.append("确认无误后加 --apply 真删:")
        for f in deleted_files:
            lines.append(f"  - {f}")
    report = "\n".join(lines) + "\n"

    # 写报告（不推送，想看时 cat 一下）
    try:
        with open(REPORT_FILE, "w", encoding="utf-8") as fh:
            fh.write(report)
    except OSError:
        pass  # 写不进报告也不能掩盖磁盘问题，退出码照常返回

    # 同时打到 stdout，方便 cron 追加到日志文件
    print(report, end="")

    # 4) 退出码：有 CRIT 就非 0，告警链路据此触发通知
    sys.exit(1 if has_crit else 0)


if __name__ == "__main__":
    main()

python3 disk_inspect.py            # LIST 模式：只打印待删清单，不真删
python3 disk_inspect.py --apply    # 确认无误后真删过期日志

    # 写报告（不推送，想看时 cat 一下）
    try:
        with open(REPORT_FILE, "w", encoding="utf-8") as fh:
            fh.write(report)
    except OSError:
        pass  # 写不进报告也不能掩盖磁盘问题，退出码照常返回

    # 同时打到 stdout，方便 cron 追加到日志文件
    print(report, end="")

    # 退出码：有 CRIT 就非 0，告警链路据此触发通知
    sys.exit(1 if has_crit else 0)