smart_checker.py - master - triv/cursor_examples

smart_checker.py
371 строка14 KB
triv
upload files
22 окт 2025, 14:19
22 окт 2025, 14:193b30bce
#!/usr/bin/env python3
"""
Скрипт для проверки SMART дисков и сохранения данных в SQLite базу данных.
Работает на Linux системах с установленным smartmontools.
"""

import sqlite3
import subprocess
import json
import os
import glob
import re
from datetime import datetime
from typing import List, Dict, Optional, Tuple
import logging

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('smart_checker.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class SmartChecker:
    """Класс для проверки SMART дисков и работы с базой данных."""
    
    def __init__(self, db_path: str = "smart_data.db"):
        """
        Инициализация SmartChecker.
        
        Args:
            db_path: Путь к файлу базы данных SQLite
        """
        self.db_path = db_path
        self.init_database()
    
    def init_database(self) -> None:
        """Создание таблицы в базе данных, если она не существует."""
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    CREATE TABLE IF NOT EXISTS smart_data (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        disk_name TEXT NOT NULL,
                        check_date DATETIME NOT NULL,
                        health_status TEXT,
                        temperature INTEGER,
                        power_on_hours INTEGER,
                        reallocated_sectors INTEGER,
                        pending_sectors INTEGER,
                        uncorrectable_sectors INTEGER,
                        raw_data TEXT,
                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
                    )
                ''')
                conn.commit()
                logger.info("База данных инициализирована успешно")
        except sqlite3.Error as e:
            logger.error(f"Ошибка при инициализации базы данных: {e}")
            raise
    
    def discover_disks(self) -> List[str]:
        """
        Обнаружение всех доступных дисков в системе.
        
        Returns:
            Список путей к дискам
        """
        disks = []
        
        # Поиск SATA/SCSI дисков
        sata_disks = glob.glob('/dev/sd*')
        # Поиск NVMe дисков
        nvme_disks = glob.glob('/dev/nvme*')
        # Поиск IDE дисков
        ide_disks = glob.glob('/dev/hd*')
        
        all_disks = sata_disks + nvme_disks + ide_disks
        
        for disk in all_disks:
            if self._is_smart_capable(disk):
                disks.append(disk)
        
        logger.info(f"Найдено {len(disks)} дисков с поддержкой SMART: {disks}")
        return disks
    
    def _is_smart_capable(self, disk_path: str) -> bool:
        """
        Проверка поддержки SMART для диска.
        
        Args:
            disk_path: Путь к диску
            
        Returns:
            True если диск поддерживает SMART
        """
        try:
            result = subprocess.run(
                ['smartctl', '-i', disk_path],
                capture_output=True,
                text=True,
                timeout=10
            )
            return result.returncode == 0 and 'SMART support is: Enabled' in result.stdout
        except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
            return False
    
    def get_smart_data(self, disk_path: str) -> Optional[Dict]:
        """
        Получение SMART данных для диска.
        
        Args:
            disk_path: Путь к диску
            
        Returns:
            Словарь с SMART данными или None при ошибке
        """
        try:
            # Получение общей информации о SMART
            result = subprocess.run(
                ['smartctl', '-H', disk_path],
                capture_output=True,
                text=True,
                timeout=30
            )
            
            if result.returncode != 0:
                logger.warning(f"Не удалось получить SMART данные для {disk_path}")
                return None
            
            # Получение детальной информации
            detail_result = subprocess.run(
                ['smartctl', '-A', disk_path],
                capture_output=True,
                text=True,
                timeout=30
            )
            
            smart_data = {
                'disk_name': disk_path,
                'health_status': self._parse_health_status(result.stdout),
                'temperature': self._parse_temperature(detail_result.stdout),
                'power_on_hours': self._parse_power_on_hours(detail_result.stdout),
                'reallocated_sectors': self._parse_reallocated_sectors(detail_result.stdout),
                'pending_sectors': self._parse_pending_sectors(detail_result.stdout),
                'uncorrectable_sectors': self._parse_uncorrectable_sectors(detail_result.stdout),
                'raw_data': detail_result.stdout
            }
            
            logger.info(f"SMART данные получены для {disk_path}: {smart_data['health_status']}")
            return smart_data
            
        except subprocess.TimeoutExpired:
            logger.error(f"Таймаут при получении SMART данных для {disk_path}")
            return None
        except Exception as e:
            logger.error(f"Ошибка при получении SMART данных для {disk_path}: {e}")
            return None
    
    def _parse_health_status(self, output: str) -> str:
        """Парсинг статуса здоровья диска."""
        if 'PASSED' in output:
            return 'PASSED'
        elif 'FAILED' in output:
            return 'FAILED'
        else:
            return 'UNKNOWN'
    
    def _parse_temperature(self, output: str) -> Optional[int]:
        """Парсинг температуры диска."""
        # Поиск температуры в различных форматах
        temp_patterns = [
            r'Temperature.*?(\d+)\s*C',
            r'Current Drive Temperature.*?(\d+)',
            r'Airflow Temperature.*?(\d+)'
        ]
        
        for pattern in temp_patterns:
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                return int(match.group(1))
        return None
    
    def _parse_power_on_hours(self, output: str) -> Optional[int]:
        """Парсинг часов работы диска."""
        patterns = [
            r'Power_On_Hours.*?(\d+)',
            r'Power-on Hours.*?(\d+)'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                return int(match.group(1))
        return None
    
    def _parse_reallocated_sectors(self, output: str) -> Optional[int]:
        """Парсинг переназначенных секторов."""
        patterns = [
            r'Reallocated_Sector_Ct.*?(\d+)',
            r'Reallocated Sectors Count.*?(\d+)'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                return int(match.group(1))
        return None
    
    def _parse_pending_sectors(self, output: str) -> Optional[int]:
        """Парсинг секторов в ожидании."""
        patterns = [
            r'Current_Pending_Sector.*?(\d+)',
            r'Pending Sectors Count.*?(\d+)'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                return int(match.group(1))
        return None
    
    def _parse_uncorrectable_sectors(self, output: str) -> Optional[int]:
        """Парсинг неисправимых секторов."""
        patterns = [
            r'Offline_Uncorrectable.*?(\d+)',
            r'Uncorrectable Sector Count.*?(\d+)'
        ]
        
        for pattern in patterns:
            match = re.search(pattern, output, re.IGNORECASE)
            if match:
                return int(match.group(1))
        return None
    
    def save_smart_data(self, smart_data: Dict) -> bool:
        """
        Сохранение SMART данных в базу данных.
        
        Args:
            smart_data: Словарь с SMART данными
            
        Returns:
            True если данные сохранены успешно
        """
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    INSERT INTO smart_data (
                        disk_name, check_date, health_status, temperature,
                        power_on_hours, reallocated_sectors, pending_sectors,
                        uncorrectable_sectors, raw_data
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ''', (
                    smart_data['disk_name'],
                    datetime.now().isoformat(),
                    smart_data['health_status'],
                    smart_data['temperature'],
                    smart_data['power_on_hours'],
                    smart_data['reallocated_sectors'],
                    smart_data['pending_sectors'],
                    smart_data['uncorrectable_sectors'],
                    smart_data['raw_data']
                ))
                conn.commit()
                logger.info(f"SMART данные для {smart_data['disk_name']} сохранены в базу")
                return True
        except sqlite3.Error as e:
            logger.error(f"Ошибка при сохранении данных в базу: {e}")
            return False
    
    def run_check(self) -> None:
        """Основная функция проверки всех дисков."""
        logger.info("Начало проверки SMART дисков")
        
        # Проверка наличия smartctl
        try:
            subprocess.run(['smartctl', '--version'], capture_output=True, check=True)
        except (subprocess.CalledProcessError, FileNotFoundError):
            logger.error("smartctl не найден. Установите smartmontools: sudo apt-get install smartmontools")
            return
        
        # Обнаружение дисков
        disks = self.discover_disks()
        if not disks:
            logger.warning("Диски с поддержкой SMART не найдены")
            return
        
        # Проверка каждого диска
        successful_checks = 0
        for disk in disks:
            logger.info(f"Проверка диска: {disk}")
            smart_data = self.get_smart_data(disk)
            
            if smart_data:
                if self.save_smart_data(smart_data):
                    successful_checks += 1
                else:
                    logger.error(f"Не удалось сохранить данные для {disk}")
            else:
                logger.warning(f"Не удалось получить SMART данные для {disk}")
        
        logger.info(f"Проверка завершена. Успешно обработано: {successful_checks}/{len(disks)} дисков")
    
    def get_latest_data(self) -> List[Dict]:
        """
        Получение последних данных из базы.
        
        Returns:
            Список последних записей для каждого диска
        """
        try:
            with sqlite3.connect(self.db_path) as conn:
                cursor = conn.cursor()
                cursor.execute('''
                    SELECT disk_name, check_date, health_status, temperature,
                           power_on_hours, reallocated_sectors, pending_sectors,
                           uncorrectable_sectors
                    FROM smart_data
                    WHERE id IN (
                        SELECT MAX(id) FROM smart_data GROUP BY disk_name
                    )
                    ORDER BY disk_name
                ''')
                
                columns = [description[0] for description in cursor.description]
                results = []
                for row in cursor.fetchall():
                    results.append(dict(zip(columns, row)))
                
                return results
        except sqlite3.Error as e:
            logger.error(f"Ошибка при получении данных из базы: {e}")
            return []

def main():
    """Главная функция скрипта."""
    print("SMART Disk Checker")
    print("==================")
    
    # Создание экземпляра SmartChecker
    checker = SmartChecker()
    
    # Запуск проверки
    checker.run_check()
    
    # Вывод последних данных
    print("\nПоследние данные SMART:")
    print("-" * 50)
    latest_data = checker.get_latest_data()
    
    if latest_data:
        for data in latest_data:
            print(f"Диск: {data['disk_name']}")
            print(f"  Статус: {data['health_status']}")
            print(f"  Температура: {data['temperature']}°C" if data['temperature'] else "  Температура: N/A")
            print(f"  Часы работы: {data['power_on_hours']}" if data['power_on_hours'] else "  Часы работы: N/A")
            print(f"  Переназначенные сектора: {data['reallocated_sectors']}" if data['reallocated_sectors'] else "  Переназначенные сектора: N/A")
            print(f"  Дата проверки: {data['check_date']}")
            print()
    else:
        print("Данные не найдены")

if __name__ == "__main__":
    main()