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" may be a list OR a dict keyed by timestamp — handle both smart_raw = device_data.get("smart") or {} if isinstance(smart_raw, dict): smart_values = list(smart_raw.values()) elif isinstance(smart_raw, list): smart_values = smart_raw else: smart_values = [] # Sort by date if values have a date field, otherwise take last entry if smart_values and isinstance(smart_values[0], dict) and "date" in smart_values[0]: smart_values = sorted(smart_values, key=lambda x: x.get("date", "")) latest_smart = smart_values[-1] if smart_values else {} name = device_info.get("device_name") or device_path.split("/")[-1] model = device_info.get("model_name") or "Unknown" status_code = latest_smart.get("Status", -1) if status_code == 0: status = "passed" elif status_code > 0: status = "failed" else: status = "unknown" # Temperature: try top-level "temp", then nested attrs temperature: int | None = None temp_val = latest_smart.get("temp") or latest_smart.get("temperature") if temp_val is not None: try: temperature = int(temp_val) except (TypeError, ValueError): pass devices.append(ScrutinyDevice(name=name, model=model, status=status, temperature=temperature)) return sorted(devices, key=lambda d: d.name)