Add SMART short test and status

This commit is contained in:
2026-02-08 01:50:39 +03:00
parent 48dc1f38ac
commit a7d5fb5459
3 changed files with 62 additions and 1 deletions

View File

@@ -5,7 +5,7 @@ from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKe
from app import dp, cfg from app import dp, cfg
from auth import is_admin_msg from auth import is_admin_msg
from keyboards import system_info_kb, system_ops_kb, system_logs_kb from keyboards import system_info_kb, system_ops_kb, system_logs_kb
from system_checks import security, disks, hardware from system_checks import security, disks, hardware, list_disks, smart_last_test
from services.http_checks import get_url_checks, check_url from services.http_checks import get_url_checks, check_url
from services.queue import enqueue from services.queue import enqueue
from services.updates import list_updates, apply_updates from services.updates import list_updates, apply_updates
@@ -71,6 +71,52 @@ async def metrics(msg: Message):
await msg.answer(summarize(state.METRICS_STORE, minutes=15), reply_markup=system_info_kb) await msg.answer(summarize(state.METRICS_STORE, minutes=15), reply_markup=system_info_kb)
@dp.message(F.text == "🧪 SMART test")
async def smart_test(msg: Message):
if not is_admin_msg(msg):
return
disks_list = list_disks()
if not disks_list:
await msg.answer("💽 Disks\n\n❌ No disks found", reply_markup=system_info_kb)
return
await msg.answer("🧪 Starting SMART short tests…", reply_markup=system_info_kb)
async def worker():
lines = ["🧪 SMART short tests\n"]
for dev in disks_list:
rc, out = await run_cmd(["sudo", "smartctl", "-t", "short", dev], timeout=20)
if rc != 0:
lines.append(f"🔴 {dev}: {out.strip() or 'error'}")
continue
summary = "started"
for line in out.splitlines():
if "Please wait" in line or "will complete" in line:
summary = line.strip()
break
lines.append(f"🟢 {dev}: {summary}")
await msg.answer("\n".join(lines), reply_markup=system_info_kb)
asyncio.create_task(worker())
@dp.message(F.text == "🧪 SMART status")
async def smart_status(msg: Message):
if not is_admin_msg(msg):
return
disks_list = list_disks()
if not disks_list:
await msg.answer("💽 Disks\n\n❌ No disks found", reply_markup=system_info_kb)
return
lines = ["🧪 SMART last tests\n"]
for dev in disks_list:
last = smart_last_test(dev)
lines.append(f"{dev}: {last}")
await msg.answer("\n".join(lines), reply_markup=system_info_kb)
@dp.message(F.text == "🧾 Audit") @dp.message(F.text == "🧾 Audit")
async def audit_log(msg: Message): async def audit_log(msg: Message):
if not is_admin_msg(msg): if not is_admin_msg(msg):

View File

@@ -64,6 +64,7 @@ system_info_kb = ReplyKeyboardMarkup(
keyboard=[ keyboard=[
[KeyboardButton(text="💽 Disks"), KeyboardButton(text="🔐 Security")], [KeyboardButton(text="💽 Disks"), KeyboardButton(text="🔐 Security")],
[KeyboardButton(text="📈 Metrics"), KeyboardButton(text="🧱 Hardware")], [KeyboardButton(text="📈 Metrics"), KeyboardButton(text="🧱 Hardware")],
[KeyboardButton(text="🧪 SMART test"), KeyboardButton(text="🧪 SMART status")],
[KeyboardButton(text="⬅️ System")], [KeyboardButton(text="⬅️ System")],
], ],
resize_keyboard=True, resize_keyboard=True,

View File

@@ -122,6 +122,20 @@ def disk_temperature(dev: str) -> str:
return "n/a" return "n/a"
def smart_last_test(dev: str) -> str:
out = _cmd(f"smartctl -l selftest {dev}")
if not out or "ERROR:" in out:
return "n/a"
for line in out.splitlines():
if "No self-tests have been logged" in line:
return "no tests"
if line.lstrip().startswith("#"):
return line.strip()
return "n/a"
def disks() -> str: def disks() -> str:
disks = list_disks() disks = list_disks()