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)