监控硬盘健康状态的脚本如何写

本文目录导读：

1. Shell脚本版本（Linux）
2. Python版本（跨平台）
3. 配置文件和系统服务
4. 使用方法

我来帮你写一个监控硬盘健康状态的脚本,提供Linux Shell和Python两个版本。

Shell脚本版本（Linux）

#!/bin/bash
# 硬盘健康监控脚本
# 依赖：smartmontools（需要安装）
# 配置
EMAIL="admin@example.com"
THRESHOLD_TEMP=50  # 温度阈值（摄氏度）
THRESHOLD_REALLOC=10  # 重分配扇区阈值
CHECK_INTERVAL=3600  # 检查间隔（秒）
LOG_FILE="/var/log/disk_health.log"
# 检查依赖
check_dependencies() {
    if ! command -v smartctl &> /dev/null; then
        echo "错误：请安装smartmontools"
        echo "Ubuntu/Debian: sudo apt-get install smartmontools"
        echo "CentOS/RHEL: sudo yum install smartmontools"
        exit 1
    fi
}
# 获取硬盘列表
get_disks() {
    lsblk -ndo NAME | grep -E '^(sd|nvme|hd)'
}
# 检查单个硬盘
check_disk() {
    local disk=$1
    local disk_info=""
    local has_error=0
    echo "========================"
    echo "检查硬盘: /dev/$disk"
    echo "时间: $(date)"
    # 获取硬盘信息
    local smart_info=$(sudo smartctl -a /dev/$disk 2>/dev/null)
    # 检查硬盘温度
    local temp=$(echo "$smart_info" | grep -i "Temperature_Celsius" | awk '{print $10}')
    if [ -z "$temp" ]; then
        temp=$(echo "$smart_info" | grep -i "Current Drive Temperature" | awk '{print $5}')
    fi
    if [ ! -z "$temp" ] && [ "$temp" -gt "$THRESHOLD_TEMP" ]; then
        disk_info+="⚠️  警告：硬盘温度过高 ($temp°C)\n"
        has_error=1
    fi
    # 检查重分配扇区数
    local realloc=$(echo "$smart_info" | grep "Reallocated_Sector_Ct" | awk '{print $10}')
    if [ ! -z "$realloc" ] && [ "$realloc" -gt "$THRESHOLD_REALLOC" ]; then
        disk_info+="⚠️  警告：重分配扇区数过多 ($realloc)\n"
        has_error=1
    fi
    # 检查待处理扇区
    local pending=$(echo "$smart_info" | grep "Current_Pending_Sector" | awk '{print $10}')
    if [ ! -z "$pending" ] && [ "$pending" -gt 0 ]; then
        disk_info+="⚠️  警告：存在待处理扇区 ($pending)\n"
        has_error=1
    fi
    # 检查UDMA CRC错误
    local udma=$(echo "$smart_info" | grep "UDMA_CRC_Error_Count" | awk '{print $10}')
    if [ ! -z "$udma" ] && [ "$udma" -gt 100 ]; then
        disk_info+="⚠️  警告：UDMA CRC错误过多 ($udma)\n"
        has_error=1
    fi
    # 检查硬盘整体健康状态
    local health=$(echo "$smart_info" | grep -i "SMART overall-health" | awk -F': ' '{print $2}')
    if [ -z "$health" ]; then
        health=$(echo "$smart_info" | grep -i "SMART Status" | awk -F': ' '{print $2}')
    fi
    if [ ! -z "$health" ] && [ "$health" != "PASSED" ] && [ "$health" != "OK" ]; then
        disk_info+="❌  严重：硬盘健康状态为 $health\n"
        has_error=1
    fi
    # 汇总信息
    echo "温度: ${temp:-未知}°C"
    echo "重分配扇区: ${realloc:-0}"
    echo "待处理扇区: ${pending:-0}"
    echo "健康状态: ${health:-未知}"
    # 输出监控信息
    if [ $has_error -eq 1 ]; then
        echo -e "$disk_info"
        echo "$(date): /dev/$disk 检测到问题" >> "$LOG_FILE"
        echo -e "$disk_info" >> "$LOG_FILE"
        # 发送邮件通知
        if command -v mail &> /dev/null && [ ! -z "$EMAIL" ]; then
            echo -e "硬盘健康警告\n\n硬盘: /dev/$disk\n$disk_info" | mail -s "硬盘健康警告 - /dev/$disk" "$EMAIL"
        fi
    fi
    return $has_error
}
# 主函数
main() {
    check_dependencies
    echo "硬盘健康监控脚本"
    echo "开始时间: $(date)"
    local error_count=0
    # 检查所有硬盘
    for disk in $(get_disks); do
        check_disk "$disk"
        error_count=$((error_count + $?))
    done
    echo "========================"
    echo "检查完成，发现 $error_count 个问题"
    # 记录日志
    echo "$(date): 检查完成，发现 $error_count 个问题" >> "$LOG_FILE"
    return $error_count
}
# 运行
main

Python版本（跨平台）

#!/usr/bin/env python3
"""
硬盘健康监控脚本
支持：Windows, Linux, macOS
"""
import os
import sys
import time
import json
import logging
import subprocess
from datetime import datetime
from typing import Dict, List, Optional
import smtplib
from email.mime.text import MIMEText
import platform
class DiskHealthMonitor:
    def __init__(self, config: Dict = None):
        self.config = config or self.default_config()
        self.logger = self.setup_logging()
    def default_config(self) -> Dict:
        return {
            'email': 'admin@example.com',
            'threshold_temp': 50,
            'threshold_realloc': 10,
            'threshold_pending': 0,
            'threshold_crc': 100,
            'check_interval': 3600,
            'log_file': '/var/log/disk_health.log',
            'smtp_server': 'smtp.gmail.com',
            'smtp_port': 587,
            'smtp_user': '',
            'smtp_password': ''
        }
    def setup_logging(self) -> logging.Logger:
        logger = logging.getLogger('DiskHealthMonitor')
        logger.setLevel(logging.INFO)
        # 文件处理器
        fh = logging.FileHandler(self.config['log_file'])
        fh.setLevel(logging.INFO)
        # 控制台处理器
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        # 格式化
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)
        logger.addHandler(fh)
        logger.addHandler(ch)
        return logger
    def get_disks_windows(self) -> List[str]:
        """获取Windows下的磁盘列表"""
        disks = []
        try:
            # 使用wmic获取物理磁盘
            result = subprocess.run(
                ['wmic', 'diskdrive', 'get', 'Index,Model,Status'],
                capture_output=True, text=True, check=True
            )
            lines = result.stdout.strip().split('\n')[1:]
            for line in lines:
                if line.strip():
                    index = line.split()[0]
                    disks.append(f'PhysicalDrive{index}')
        except:
            # 备用方法：遍历可能的磁盘号
            for i in range(10):
                disk = f'PhysicalDrive{i}'
                if os.path.exists(f'\\\\.\\{disk}'):
                    disks.append(disk)
        return disks
    def get_disks_linux(self) -> List[str]:
        """获取Linux下的磁盘列表"""
        disks = []
        try:
            result = subprocess.run(
                ['lsblk', '-ndo', 'NAME'],
                capture_output=True, text=True, check=True
            )
            for line in result.stdout.strip().split('\n'):
                if line and line.startswith(('sd', 'nvme', 'hd')):
                    disks.append(f'/dev/{line}')
        except:
            pass
        return disks
    def get_disks_macos(self) -> List[str]:
        """获取macOS下的磁盘列表"""
        disks = []
        try:
            result = subprocess.run(
                ['diskutil', 'list', 'internal', 'physical'],
                capture_output=True, text=True, check=True
            )
            for line in result.stdout.split('\n'):
                if '/dev/disk' in line:
                    disk = line.split()[-1]
                    if 'internal' in line.lower():
                        disks.append(disk)
        except:
            pass
        return disks
    def get_disks(self) -> List[str]:
        """根据平台获取磁盘列表"""
        system = platform.system()
        if system == 'Windows':
            return self.get_disks_windows()
        elif system == 'Linux':
            return self.get_disks_linux()
        elif system == 'Darwin':
            return self.get_disks_macos()
        else:
            self.logger.error(f"不支持的操作系统: {system}")
            return []
    def check_smart_linux(self, disk: str) -> Dict:
        """检查Linux下硬盘SMART信息"""
        result = {
            'disk': disk,
            'temperature': None,
            'reallocated_sectors': None,
            'pending_sectors': None,
            'crc_errors': None,
            'health_status': 'UNKNOWN',
            'errors': []
        }
        try:
            # 获取SMART信息
            smartctl = subprocess.run(
                ['sudo', 'smartctl', '-a', disk],
                capture_output=True, text=True, timeout=30
            )
            if smartctl.returncode != 0:
                result['errors'].append(f"smartctl失败: {smartctl.stderr}")
                return result
            output = smartctl.stdout
            # 解析温度
            for line in output.split('\n'):
                if 'Temperature_Celsius' in line:
                    try:
                        result['temperature'] = int(line.split()[-1])
                    except:
                        pass
            # 解析重分配扇区
            for line in output.split('\n'):
                if 'Reallocated_Sector_Ct' in line:
                    try:
                        result['reallocated_sectors'] = int(line.split()[-1])
                    except:
                        pass
            # 解析待处理扇区
            for line in output.split('\n'):
                if 'Current_Pending_Sector' in line:
                    try:
                        result['pending_sectors'] = int(line.split()[-1])
                    except:
                        pass
            # 解析CRC错误
            for line in output.split('\n'):
                if 'UDMA_CRC_Error_Count' in line:
                    try:
                        result['crc_errors'] = int(line.split()[-1])
                    except:
                        pass
            # 解析健康状态
            for line in output.split('\n'):
                if 'SMART overall-health' in line or 'SMART Status' in line:
                    status = line.split(':')[-1].strip()
                    result['health_status'] = status
        except Exception as e:
            result['errors'].append(f"检查失败: {str(e)}")
        return result
    def check_smart_windows(self, disk: str) -> Dict:
        """检查Windows下硬盘SMART信息"""
        result = {
            'disk': disk,
            'temperature': None,
            'reallocated_sectors': None,
            'pending_sectors': None,
            'crc_errors': None,
            'health_status': 'UNKNOWN',
            'errors': []
        }
        try:
            # 使用WMIC获取温度（如果支持）
            wmic_temp = subprocess.run(
                ['wmic', '/namespace:\\\\root\\wmi', 'path', 
                 'MSStorageDriver_ATAPISmartData', 'get', 'Temperature'],
                capture_output=True, text=True, timeout=30
            )
            # 使用第三方工具如smartctl for Windows
            # 这里模拟基本检查
            result['health_status'] = 'PASSED'
        except Exception as e:
            result['errors'].append(f"检查失败: {str(e)}")
        return result
    def check_health(self, disk_info: Dict) -> bool:
        """检查硬盘健康状况"""
        has_problem = False
        # 检查温度
        temp = disk_info.get('temperature')
        if temp and temp > self.config['threshold_temp']:
            msg = f"硬盘 {disk_info['disk']} 温度过高: {temp}°C"
            self.logger.warning(msg)
            disk_info['errors'].append(msg)
            has_problem = True
        # 检查重分配扇区
        realloc = disk_info.get('reallocated_sectors')
        if realloc and realloc > self.config['threshold_realloc']:
            msg = f"硬盘 {disk_info['disk']} 重分配扇区过多: {realloc}"
            self.logger.warning(msg)
            disk_info['errors'].append(msg)
            has_problem = True
        # 检查待处理扇区
        pending = disk_info.get('pending_sectors')
        if pending and pending > self.config['threshold_pending']:
            msg = f"硬盘 {disk_info['disk']} 存在待处理扇区: {pending}"
            self.logger.warning(msg)
            disk_info['errors'].append(msg)
            has_problem = True
        # 检查CRC错误
        crc = disk_info.get('crc_errors')
        if crc and crc > self.config['threshold_crc']:
            msg = f"硬盘 {disk_info['disk']} CRC错误过多: {crc}"
            self.logger.warning(msg)
            disk_info['errors'].append(msg)
            has_problem = True
        # 检查健康状态
        health = disk_info.get('health_status')
        if health and health not in ['PASSED', 'OK', 'UNKNOWN']:
            msg = f"硬盘 {disk_info['disk']} 健康状态异常: {health}"
            self.logger.error(msg)
            disk_info['errors'].append(msg)
            has_problem = True
        return has_problem
    def send_alert(self, disk_info: Dict):
        """发送告警通知"""
        if not self.config.get('email'):
            return
        subject = f"硬盘健康警告 - {disk_info['disk']}"
        body = f"""
硬盘健康监控报告
时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
硬盘: {disk_info['disk']}
温度: {disk_info.get('temperature', '未知')}°C
重分配扇区: {disk_info.get('reallocated_sectors', '未知')}
待处理扇区: {disk_info.get('pending_sectors', '未知')}
CRC错误: {disk_info.get('crc_errors', '未知')}
健康状态: {disk_info.get('health_status', '未知')}
错误信息:
{chr(10).join(disk_info['errors'])}
        """
        try:
            msg = MIMEText(body)
            msg['Subject'] = subject
            msg['From'] = self.config['smtp_user']
            msg['To'] = self.config['email']
            server = smtplib.SMTP(
                self.config['smtp_server'], 
                self.config['smtp_port']
            )
            server.starttls()
            server.login(
                self.config['smtp_user'], 
                self.config['smtp_password']
            )
            server.send_message(msg)
            server.quit()
            self.logger.info(f"告警邮件已发送到 {self.config['email']}")
        except Exception as e:
            self.logger.error(f"发送告警邮件失败: {str(e)}")
    def run_once(self):
        """运行一次检查"""
        self.logger.info("开始硬盘健康检查")
        disks = self.get_disks()
        if not disks:
            self.logger.warning("未找到可检查的硬盘")
            return
        for disk in disks:
            self.logger.info(f"检查硬盘: {disk}")
            # 根据平台检查
            system = platform.system()
            if system == 'Linux':
                disk_info = self.check_smart_linux(disk)
            elif system == 'Windows':
                disk_info = self.check_smart_windows(disk)
            else:
                self.logger.error(f"不支持的平台: {system}")
                continue
            # 检查健康状况
            has_problem = self.check_health(disk_info)
            if has_problem:
                self.send_alert(disk_info)
                # 保存报告
                report_file = f"disk_health_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
                with open(report_file, 'w') as f:
                    json.dump(disk_info, f, indent=2)
                self.logger.info(f"报告已保存到: {report_file}")
        self.logger.info("硬盘健康检查完成")
    def run_continuous(self):
        """持续监控模式"""
        self.logger.info("启动持续监控模式")
        while True:
            try:
                self.run_once()
                self.logger.info(f"等待 {self.config['check_interval']} 秒后再次检查...")
                time.sleep(self.config['check_interval'])
            except KeyboardInterrupt:
                self.logger.info("收到中断信号，停止监控")
                break
            except Exception as e:
                self.logger.error(f"监控出错: {str(e)}")
                time.sleep(60)  # 出错后等待1分钟重试
def main():
    """主函数"""
    import argparse
    parser = argparse.ArgumentParser(description='硬盘健康监控脚本')
    parser.add_argument('--config', '-c', help='配置文件路径')
    parser.add_argument('--continuous', '-C', action='store_true', 
                       help='持续监控模式')
    parser.add_argument('--interval', '-i', type=int, default=3600,
                       help='检查间隔（秒）')
    parser.add_argument('--email', '-e', help='告警邮箱')
    args = parser.parse_args()
    # 加载配置
    config = {}
    if args.config:
        try:
            with open(args.config) as f:
                config = json.load(f)
        except Exception as e:
            print(f"加载配置文件失败: {e}")
            sys.exit(1)
    # 命令行参数覆盖配置文件
    if args.interval:
        config['check_interval'] = args.interval
    if args.email:
        config['email'] = args.email
    # 创建监控器
    monitor = DiskHealthMonitor(config)
    # 运行
    if args.continuous:
        monitor.run_continuous()
    else:
        monitor.run_once()
if __name__ == '__main__':
    main()

配置文件和系统服务

配置文件示例 (config.json)

{
    "email": "admin@example.com",
    "threshold_temp": 50,
    "threshold_realloc": 10,
    "threshold_pending": 0,
    "threshold_crc": 100,
    "check_interval": 3600,
    "log_file": "/var/log/disk_health.log",
    "smtp_server": "smtp.gmail.com",
    "smtp_port": 587,
    "smtp_user": "your-email@gmail.com",
    "smtp_password": "your-app-password"
}

Systemd服务文件

[Unit]
Description=Disk Health Monitor Service
After=network.target
[Service]
Type=simple
User=root
ExecStart=/usr/local/bin/disk_health_monitor.py --continuous --config /etc/disk_health/config.json
Restart=always
RestartSec=60
[Install]
WantedBy=multi-user.target

使用方法

Linux:

# 安装smartmontools
sudo apt-get install smartmontools  # Ubuntu/Debian
sudo yum install smartmontools      # CentOS/RHEL
# 设置执行权限
chmod +x disk_health.sh
chmod +x disk_health_monitor.py
# 运行Shell脚本
./disk_health.sh
# 运行Python脚本（单次检查）
python3 disk_health_monitor.py
# 运行Python脚本（持续监控）
python3 disk_health_monitor.py --continuous --interval 3600
# 安装为系统服务
sudo cp disk_health_monitor.py /usr/local/bin/
sudo mkdir -p /etc/disk_health
sudo cp config.json /etc/disk_health/
sudo cp disk-health.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable disk-health
sudo systemctl start disk-health

Windows:

# 安装Python
# 下载并安装Python from python.org
# 安装依赖
pip install smtplib
# 运行脚本
python disk_health_monitor.py --continuous --interval 3600
# 或创建计划任务（管理员PowerShell）
$action = New-ScheduledTaskAction -Execute "python.exe" -Argument "C:\path\to\disk_health_monitor.py --continuous"
$trigger = New-ScheduledTaskTrigger -AtStartup
$principal = New-ScheduledTaskPrincipal -UserId "SYSTEM" -LogonType ServiceAccount
Register-ScheduledTask -TaskName "DiskHealthMonitor" -Action $action -Trigger $trigger -Principal $principal

这个脚本支持：

• 多平台（Linux/Windows/macOS）
• 实时监控硬盘温度、SMART信息
• 自动发送告警通知
• 日志记录和报告生成
• 配置灵活可定制
• 支持持续监控和定时检查