From a0c1193f1458aaad5255719b58a13cdceabc96da Mon Sep 17 00:00:00 2001
From: Micha <michideheld@gmx.de>
Date: Sun, 5 Apr 2026 21:05:24 +0000
Subject: [PATCH] feat: add scrutiny client

---
 .../backend/app/clients/scrutiny_client.py    | 71 +++++++++++++++++++
 1 file changed, 71 insertions(+)
 create mode 100644 apps/dashboard/backend/app/clients/scrutiny_client.py

diff --git a/apps/dashboard/backend/app/clients/scrutiny_client.py b/apps/dashboard/backend/app/clients/scrutiny_client.py
new file mode 100644
index 0000000..b123a09
--- /dev/null
+++ b/apps/dashboard/backend/app/clients/scrutiny_client.py
@@ -0,0 +1,71 @@
+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)