如何通过Python实现Linux服务器存储容量与昨日数据的自动对比

以下是可直接落地的Python实现方案，覆盖SSH连接、数据解析、差异对比、Excel存储、告警通知全流程：

前置依赖安装 #

执行以下命令安装所需第三方库：

pip install paramiko pandas openpyxl

完整实现代码 #

import paramiko
import pandas as pd
import os
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

# -------------------------- 配置项 --------------------------
# 服务器SSH配置
SSH_CONFIG = {
    "host": "你的服务器IP",
    "port": 22,
    "username": "ssh用户名",
    "password": "ssh密码（如果用密钥认证可以删掉这行）",
    # "private_key_path": "~/.ssh/id_rsa" # 密钥认证路径，推荐使用
}
# 存储配置
EXCEL_SAVE_PATH = "./disk_capacity_record.xlsx"
# 告警配置
ALERT_THRESHOLD = 5 # 容量日涨幅超过5%则告警，可自行调整
NOTIFY_EMAILS = ["zhangsan@example.com", "lisi@example.com"] # 通知邮箱列表
# 邮箱发送配置（用其他通知方式可以删掉这部分）
SMTP_CONFIG = {
    "server": "smtp.example.com",
    "port": 465,
    "user": "alert@example.com",
    "password": "邮箱授权码"
}
# -----------------------------------------------------------

def get_disk_capacity():
    """通过SSH获取目标服务器df -h结果"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 连接逻辑，优先用密钥，没有用密码
    if "private_key_path" in SSH_CONFIG:
        private_key = paramiko.RSAKey.from_private_key_file(os.path.expanduser(SSH_CONFIG["private_key_path"]))
        ssh.connect(SSH_CONFIG["host"], port=SSH_CONFIG["port"], username=SSH_CONFIG["username"], pkey=private_key)
    else:
        ssh.connect(SSH_CONFIG["host"], port=SSH_CONFIG["port"], username=SSH_CONFIG["username"], password=SSH_CONFIG["password"])
    
    stdin, stdout, stderr = ssh.exec_command("df -h")
    output = stdout.read().decode().strip().split("\n")
    ssh.close()

    # 解析结果
    disk_data = []
    for line in output[1:]: # 跳过表头行
        parts = list(filter(None, line.split())) # 拆分并过滤空字符串
        if len(parts) >= 6:
            disk_data.append({
                "文件系统": parts[0],
                "总容量": parts[1],
                "已用容量": parts[2],
                "可用容量": parts[3],
                "使用率": parts[4],
                "挂载点": parts[5]
            })
    return pd.DataFrame(disk_data)

def compare_with_yesterday(today_df):
    """和昨日数据对比，返回对比结果和告警列表"""
    today_str = datetime.now().strftime("%Y-%m-%d")
    if not os.path.exists(EXCEL_SAVE_PATH):
        # 首次运行无历史数据，直接存储今日数据
        with pd.ExcelWriter(EXCEL_SAVE_PATH, engine="openpyxl", mode="w") as writer:
            today_df.to_excel(writer, sheet_name=today_str, index=False)
        return today_df, []
    
    # 读取昨日数据
    yesterday_str = (datetime.now() - pd.Timedelta(days=1)).strftime("%Y-%m-%d")
    try:
        yesterday_df = pd.read_excel(EXCEL_SAVE_PATH, sheet_name=yesterday_str)
    except ValueError:
        # 找不到昨日数据，直接存今日数据
        with pd.ExcelWriter(EXCEL_SAVE_PATH, engine="openpyxl", mode="a", if_sheet_exists="replace") as writer:
            today_df.to_excel(writer, sheet_name=today_str, index=False)
        return today_df, []
    
    # 合并数据对比涨幅
    compare_df = pd.merge(today_df, yesterday_df, on="挂载点", suffixes=("_今日", "_昨日"), how="left")
    # 计算使用率涨幅
    compare_df["使用率涨幅(%)"] = compare_df.apply(lambda x: int(x["使用率_今日"].strip("%")) - int(x["使用率_昨日"].strip("%")), axis=1)
    # 筛选需要告警的条目
    alert_list = compare_df[compare_df["使用率涨幅(%)"] >= ALERT_THRESHOLD].to_dict("records")
    # 存储今日数据
    with pd.ExcelWriter(EXCEL_SAVE_PATH, engine="openpyxl", mode="a", if_sheet_exists="replace") as writer:
        compare_df.to_excel(writer, sheet_name=today_str, index=False)
    return compare_df, alert_list

def send_alert(alert_list):
    """发送告警通知，这里以邮件为例，可自行替换为企业微信/钉钉机器人通知"""
    if not alert_list:
        return
    # 构造告警内容
    content = "【磁盘容量告警】以下存储容量日涨幅超过阈值：\n\n"
    for item in alert_list:
        content += f"挂载点：{item['挂载点']}\n今日使用率：{item['使用率_今日']}\n昨日使用率：{item['使用率_昨日']}\n涨幅：{item['使用率涨幅(%)']}%\n\n"
    # 发送邮件
    msg = MIMEText(content, "plain", "utf-8")
    msg["Subject"] = f"磁盘容量告警 {datetime.now().strftime('%Y-%m-%d')}"
    msg["From"] = SMTP_CONFIG["user"]
    msg["To"] = ",".join(NOTIFY_EMAILS)

    with smtplib.SMTP_SSL(SMTP_CONFIG["server"], SMTP_CONFIG["port"]) as smtp:
        smtp.login(SMTP_CONFIG["user"], SMTP_CONFIG["password"])
        smtp.sendmail(SMTP_CONFIG["user"], NOTIFY_EMAILS, msg.as_string())

if __name__ == "__main__":
    today_disk_data = get_disk_capacity()
    compare_result, alert_list = compare_with_yesterday(today_disk_data)
    send_alert(alert_list)

定时执行配置 #

脚本调试正常后，配置定时任务实现每日自动运行：

• Linux系统使用crontab配置，示例（每日9点执行，日志输出到指定文件）：

0 9 * * * /usr/bin/python3 /你存放脚本的绝对路径/disk_check.py >> /var/log/disk_check.log 2>&1

• Windows系统可以使用「任务计划程序」，配置每日固定时间触发执行python脚本即可。

补充说明 #

• 优先使用SSH密钥认证，避免在代码中明文存储服务器密码，安全性更高
• 通知方式可根据团队需求修改，对接企业微信/钉钉/飞书机器人的webhook接口比邮件通知更即时，仅需将send_alert函数中的邮件发送逻辑替换为webhook POST请求即可
• 可根据实际需求调整告警阈值，也可以新增使用率超过固定值（比如80%）就告警的逻辑

内容的提问来源于stack exchange，提问作者Farbod Khalili