import logging import os import threading from pathlib import Path from datetime import datetime from typing import Any from .config import load_config from .database import SessionLocal from .meshcentral import _extract_devices, _parse_json_output, _run_meshctrl from .models import Asset from .presence_state import derive_presence_state, extract_last_address, extract_last_seen logger = logging.getLogger("assetmanager.presence") def _normalized_node_id(value: Any) -> str: text = str(value or "").strip() if not text: return "" if text.startswith("node//"): return text[6:] if text.startswith("node/"): return text.split("/")[-1] return text def _device_node_id(device: dict[str, Any]) -> str: node = device.get("node") if isinstance(device.get("node"), dict) else {} return str( device.get("_id") or device.get("nodeid") or device.get("nodeId") or device.get("id") or node.get("_id") or node.get("nodeid") or "" ).strip() def _device_online(device: dict[str, Any]) -> bool | None: for key in ("online", "connected", "isOnline"): if key not in device: continue value = device.get(key) if isinstance(value, bool): return value text = str(value or "").strip().casefold() if text in {"1", "true", "yes", "online", "connected"}: return True if text in {"0", "false", "no", "offline", "disconnected"}: return False # MeshCentral exposes a connection bit mask as `conn`. if "conn" in device: try: return int(device.get("conn") or 0) > 0 except (TypeError, ValueError): pass return None def _presence_config() -> dict[str, Any] | None: config = load_config().get("meshcentral", {}) if not config.get("enabled") or not config.get("presence_enabled", True): return None if not config.get("url") or not config.get("username") or not config.get("password"): return None return config def refresh_mesh_presence() -> dict[str, int]: config = _presence_config() if not config: return {"checked": 0, "online": 0, "offline": 0, "unknown": 0} # Die Präsenzprüfung läuft standardmäßig alle 30 Sekunden. Ihre erfolgreiche # ListDevices-Ausgabe wird nach dem Einlesen nicht als Zeitstempeldatei behalten. # Bei einem Parsefehler speichern wir nur eine einzige überschreibbare Fehlerdatei. result = _run_meshctrl( config, "ListDevices", include_details=False, retain_stdout=False, ) if result.returncode != 0: raise RuntimeError((result.stderr or result.stdout or "MeshCentral presence query failed").strip()) diagnostic_dir = Path(os.getenv("DIAGNOSTIC_DIR", "/app/data/logs/diagnostics")) presence_error_path = diagnostic_dir / "meshctrl-presence-last-error.stdout.json" try: devices = _extract_devices(_parse_json_output(result.stdout)) except Exception: diagnostic_dir.mkdir(parents=True, exist_ok=True) presence_error_path.write_text(result.stdout or "", encoding="utf-8", errors="replace") raise else: presence_error_path.unlink(missing_ok=True) by_exact: dict[str, dict[str, Any]] = {} by_normalized: dict[str, dict[str, Any]] = {} for device in devices: node_id = _device_node_id(device) if not node_id: continue by_exact[node_id] = device by_normalized[_normalized_node_id(node_id)] = device now = datetime.utcnow() counters = {"checked": 0, "online": 0, "offline": 0, "unknown": 0} db = SessionLocal() try: assets = db.query(Asset).filter(Asset.mesh_node_id.isnot(None), Asset.mesh_node_id != "").all() for asset in assets: counters["checked"] += 1 node_id = str(asset.mesh_node_id or "").strip() device = by_exact.get(node_id) or by_normalized.get(_normalized_node_id(node_id)) state_value = _device_online(device) if device else None reported_last_seen = extract_last_seen(device) if reported_last_seen and (asset.mesh_last_seen is None or reported_last_seen > asset.mesh_last_seen): asset.mesh_last_seen = reported_last_seen # The live query is authoritative for "online". An offline label is # only meaningful once MeshCentral has supplied a historical # connection time. Devices without such history remain "unknown". effective_last_seen = asset.mesh_last_seen derived_state = derive_presence_state(state_value, effective_last_seen) asset.mesh_online_state = derived_state asset.mesh_online = True if derived_state == "online" else (False if derived_state == "offline" else None) if derived_state == "online": asset.mesh_last_online_at = now counters[derived_state] += 1 asset.mesh_presence_updated_at = now asset.mesh_presence_source = "meshcentral" logger.debug( "MeshCentral presence asset=%s node_id=%s live=%s last_seen=%s last_address=%s derived_state=%s", asset.name, node_id, state_value, effective_last_seen, extract_last_address(device), derived_state, ) db.commit() finally: db.close() logger.info( "MeshCentral presence refreshed: checked=%s online=%s offline=%s unknown=%s", counters["checked"], counters["online"], counters["offline"], counters["unknown"], ) return counters def mesh_presence_loop(stop_event: threading.Event) -> None: while not stop_event.is_set(): config = load_config().get("meshcentral", {}) interval = max(10, min(int(config.get("presence_interval_seconds", 30) or 30), 600)) try: refresh_mesh_presence() except Exception: logger.exception("MeshCentral presence refresh failed") if stop_event.wait(interval): return