Initialer Import des AssetManagers

app/presence_state.py

@@ -0,0 +1,114 @@
+"""Shared MeshCentral presence and last-seen helpers.
+
+The MeshCentral CLI uses slightly different key names and nesting depending on
+whether ListDevices is called with --details and on the MeshCentral version.
+This module keeps timestamp parsing and state derivation in one place.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+
+def _nested(data: Any, *path: str) -> Any:
+    current = data
+    for key in path:
+        if not isinstance(current, dict):
+            return None
+        current = current.get(key)
+    return current
+
+
+def parse_mesh_datetime(value: Any) -> datetime | None:
+    """Parse MeshCentral Unix timestamps (seconds/ms) and ISO strings as UTC-naive."""
+    if value is None or isinstance(value, bool):
+        return None
+    if isinstance(value, datetime):
+        if value.tzinfo:
+            return value.astimezone(timezone.utc).replace(tzinfo=None)
+        return value
+    if isinstance(value, dict):
+        for key in ("time", "timestamp", "value"):
+            parsed = parse_mesh_datetime(value.get(key))
+            if parsed:
+                return parsed
+        return None
+    if isinstance(value, (int, float)):
+        numeric = float(value)
+        if numeric <= 0:
+            return None
+        # MeshCentral commonly emits JavaScript timestamps in milliseconds.
+        if numeric > 100_000_000_000:
+            numeric /= 1000.0
+        try:
+            return datetime.fromtimestamp(numeric, tz=timezone.utc).replace(tzinfo=None)
+        except (OverflowError, OSError, ValueError):
+            return None
+
+    text = str(value).strip()
+    if not text or text in {"0", "-"}:
+        return None
+    try:
+        return parse_mesh_datetime(float(text))
+    except ValueError:
+        pass
+    try:
+        normalized = text[:-1] + "+00:00" if text.endswith("Z") else text
+        parsed = datetime.fromisoformat(normalized)
+        if parsed.tzinfo:
+            parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
+        return parsed
+    except ValueError:
+        return None
+
+
+def extract_last_seen(device: dict[str, Any] | None) -> datetime | None:
+    """Return the best historical connection timestamp exposed by MeshCentral."""
+    if not isinstance(device, dict):
+        return None
+    candidates = (
+        device.get("lastConnect"),
+        device.get("lastconnect"),
+        device.get("lastSeen"),
+        device.get("lastseen"),
+        device.get("lastOnline"),
+        device.get("lastonline"),
+        _nested(device, "node", "lastConnect"),
+        _nested(device, "node", "lastconnect"),
+        _nested(device, "node", "lastSeen"),
+        _nested(device, "node", "lastseen"),
+    )
+    for candidate in candidates:
+        parsed = parse_mesh_datetime(candidate)
+        if parsed:
+            return parsed
+    return None
+
+
+def extract_last_address(device: dict[str, Any] | None) -> str | None:
+    if not isinstance(device, dict):
+        return None
+    candidates = (
+        _nested(device, "lastConnect", "addr"),
+        _nested(device, "lastconnect", "addr"),
+        device.get("lastaddr"),
+        device.get("lastAddress"),
+        _nested(device, "node", "lastaddr"),
+        _nested(device, "node", "lastAddress"),
+        _nested(device, "node", "ip"),
+        _nested(device, "node", "host"),
+    )
+    for value in candidates:
+        text = str(value or "").strip()
+        if text:
+            return text
+    return None
+
+
+def derive_presence_state(live_online: bool | None, last_seen: datetime | None) -> str:
+    """Derive UI state: live true=online; historical connection=offline; else unknown."""
+    if live_online is True:
+        return "online"
+    if last_seen is not None:
+        return "offline"
+    return "unknown"