115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
"""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"
|