72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from app.clients.base import BaseHTTPClient
|
|
from app.config import Settings
|
|
from app.models.sources import ScrutinyDevice, ScrutinySnapshot
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ScrutinyClient(BaseHTTPClient):
|
|
"""
|
|
Reads SMART disk health data from Scrutiny's /api/summary endpoint.
|
|
No authentication required.
|
|
"""
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
super().__init__(settings, "scrutiny", settings.scrutiny_base_url)
|
|
|
|
async def fetch_summary(self) -> ScrutinySnapshot:
|
|
snapshot = ScrutinySnapshot()
|
|
if not self.base_url:
|
|
logger.info("scrutiny skipped: base URL missing")
|
|
return snapshot
|
|
|
|
data = await self._request_json("GET", "/api/summary")
|
|
|
|
if data is None:
|
|
logger.warning("scrutiny: empty or failed response")
|
|
return snapshot
|
|
|
|
devices = self._parse_devices(data)
|
|
failed = sum(1 for d in devices if d.status == "failed")
|
|
overall: str = "online" if failed == 0 and len(devices) > 0 else ("offline" if failed > 0 else "offline")
|
|
|
|
result = ScrutinySnapshot(
|
|
source_status="online",
|
|
devices=devices,
|
|
overall_status=overall,
|
|
failed_count=failed,
|
|
total_count=len(devices),
|
|
)
|
|
logger.info("scrutiny summary: %s devices, %s failed", len(devices), failed)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _parse_devices(data: dict) -> list[ScrutinyDevice]:
|
|
devices: list[ScrutinyDevice] = []
|
|
summary: dict = (data.get("data") or {}).get("summary") or {}
|
|
|
|
for device_path, device_data in summary.items():
|
|
device_info: dict = device_data.get("device") or {}
|
|
smart_list: list = device_data.get("smart") or []
|
|
|
|
name = device_info.get("device_name") or device_path.split("/")[-1]
|
|
model = device_info.get("model_name") or "Unknown"
|
|
|
|
latest_smart = smart_list[-1] if smart_list else {}
|
|
status_code = latest_smart.get("Status", -1)
|
|
if status_code == 0:
|
|
status = "passed"
|
|
elif status_code > 0:
|
|
status = "failed"
|
|
else:
|
|
status = "unknown"
|
|
|
|
devices.append(ScrutinyDevice(name=name, model=model, status=status))
|
|
|
|
return sorted(devices, key=lambda d: d.name)
|