1069 lines
49 KiB
Python
1069 lines
49 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .config import load_config
|
|
from .models import Asset, AssetHistory, Category, SyncRun, FieldDefinition, AssetFieldValue, MeshFieldMapping
|
|
from .fields import ASSET_FIELDS
|
|
|
|
|
|
@dataclass
|
|
class SyncResult:
|
|
run: SyncRun
|
|
|
|
|
|
def _first(data: dict[str, Any], *paths: str) -> Any:
|
|
for path in paths:
|
|
value: Any = data
|
|
for part in path.split("."):
|
|
if isinstance(value, dict):
|
|
value = value.get(part)
|
|
else:
|
|
value = None
|
|
if value is None:
|
|
break
|
|
if value not in (None, "", [], {}):
|
|
return value
|
|
return None
|
|
|
|
|
|
|
|
def _find_key_recursive(data: Any, keys: set[str]) -> Any:
|
|
if isinstance(data, dict):
|
|
for key, value in data.items():
|
|
if key.lower() in keys and value not in (None, "", [], {}):
|
|
return value
|
|
for value in data.values():
|
|
found = _find_key_recursive(value, keys)
|
|
if found not in (None, "", [], {}):
|
|
return found
|
|
elif isinstance(data, list):
|
|
for item in data:
|
|
found = _find_key_recursive(item, keys)
|
|
if found not in (None, "", [], {}):
|
|
return found
|
|
return None
|
|
|
|
|
|
def _normalize_serial(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
result = str(value).strip()
|
|
return result or None
|
|
|
|
|
|
def _normalize_mac(value: Any) -> str | None:
|
|
if not value:
|
|
return None
|
|
if isinstance(value, list):
|
|
value = value[0] if value else None
|
|
if isinstance(value, dict):
|
|
value = value.get("mac") or value.get("address")
|
|
if not value:
|
|
return None
|
|
raw = re.sub(r"[^0-9A-Fa-f]", "", str(value))
|
|
if len(raw) == 12:
|
|
return ":".join(raw[i:i+2] for i in range(0, 12, 2)).upper()
|
|
return str(value).strip()
|
|
|
|
|
|
def _extract_devices(payload: Any) -> list[dict[str, Any]]:
|
|
if isinstance(payload, list):
|
|
direct = [item for item in payload if isinstance(item, dict)]
|
|
# MeshCtrl kann mehrere vollständige JSON-Blöcke nacheinander ausgeben.
|
|
nested: list[dict[str, Any]] = []
|
|
for item in payload:
|
|
if isinstance(item, (list, dict)):
|
|
try:
|
|
nested.extend(_extract_devices(item))
|
|
except ValueError:
|
|
pass
|
|
return nested or direct
|
|
if isinstance(payload, dict):
|
|
for key in ("nodes", "devices", "result"):
|
|
value = payload.get(key)
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
if isinstance(value, dict):
|
|
flattened = []
|
|
for group in value.values():
|
|
if isinstance(group, list):
|
|
flattened.extend(x for x in group if isinstance(x, dict))
|
|
elif isinstance(group, dict):
|
|
flattened.append(group)
|
|
if flattened:
|
|
return flattened
|
|
if any(key in payload for key in ("_id", "nodeid", "name")):
|
|
return [payload]
|
|
raise ValueError("MeshCtrl-Ausgabe enthält keine erkennbare Geräteliste.")
|
|
|
|
|
|
def _strip_terminal_noise(output: str) -> str:
|
|
"""Entfernt BOM, ANSI-Steuersequenzen und NUL-Zeichen aus MeshCtrl-Ausgaben."""
|
|
output = output.lstrip("\ufeff")
|
|
output = output.replace("\x00", "")
|
|
return re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", output)
|
|
|
|
|
|
def _parse_json_output(output: str) -> Any:
|
|
"""Liest auch JSON mit vor-/nachgelagerten Hinweisen oder mehreren JSON-Blöcken."""
|
|
cleaned = _strip_terminal_noise(output).strip()
|
|
if not cleaned:
|
|
raise ValueError("MeshCtrl lieferte eine leere Ausgabe.")
|
|
|
|
try:
|
|
return json.loads(cleaned)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
decoder = json.JSONDecoder()
|
|
values: list[Any] = []
|
|
position = 0
|
|
|
|
while position < len(cleaned):
|
|
starts = [i for i in (cleaned.find("[", position), cleaned.find("{", position)) if i >= 0]
|
|
if not starts:
|
|
break
|
|
start = min(starts)
|
|
try:
|
|
value, end = decoder.raw_decode(cleaned, start)
|
|
values.append(value)
|
|
position = end
|
|
except json.JSONDecodeError as exc:
|
|
# Ein nicht valider Detailblock darf nicht zu einer irreführenden Standardmeldung führen.
|
|
context_start = max(0, exc.pos - 120)
|
|
context_end = min(len(cleaned), exc.pos + 120)
|
|
context = cleaned[context_start:context_end].replace("\n", "\\n")
|
|
raise ValueError(
|
|
f"Ungültige MeshCtrl-JSON-Ausgabe bei Zeichen {exc.pos}. "
|
|
f"Ausschnitt: {context}"
|
|
) from exc
|
|
|
|
if not values:
|
|
raise ValueError(f"MeshCtrl lieferte kein erkennbares JSON: {cleaned[:300]}")
|
|
if len(values) == 1:
|
|
return values[0]
|
|
return values
|
|
|
|
|
|
def _run_meshctrl(
|
|
cfg: dict[str, Any],
|
|
action: str = "ListDevices",
|
|
include_details: bool = False,
|
|
extra_args: list[str] | None = None,
|
|
retain_stdout: bool = True,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
command = [
|
|
"node", cfg["meshctrl_path"], action,
|
|
"--url", cfg["url"], "--loginuser", cfg["username"],
|
|
"--loginpass", cfg["password"], "--json"
|
|
]
|
|
if include_details:
|
|
command.append("--details")
|
|
if extra_args:
|
|
command.extend(extra_args)
|
|
if cfg.get("tenant"):
|
|
command += ["--tenant", cfg["tenant"]]
|
|
if not cfg.get("verify_tls", True):
|
|
command.append("--noverify")
|
|
|
|
# stdout/stderr bewusst direkt in Dateien schreiben. Dadurch umgehen wir
|
|
# mögliche Pipe-/Puffergrenzen bei sehr großen MeshCtrl-JSON-Ausgaben.
|
|
diagnostic_dir = Path(os.getenv("DIAGNOSTIC_DIR", "/app/data/logs/diagnostics"))
|
|
diagnostic_dir.mkdir(parents=True, exist_ok=True)
|
|
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
|
mode = "details" if include_details else "basic"
|
|
safe_action = "".join(ch for ch in action if ch.isalnum() or ch in "-_" )
|
|
stdout_path = diagnostic_dir / f"meshctrl-{timestamp}-{safe_action}-{mode}.stdout.json"
|
|
stderr_path = diagnostic_dir / f"meshctrl-{timestamp}-{safe_action}-{mode}.stderr.log"
|
|
|
|
with stdout_path.open("wb") as stdout_file, stderr_path.open("wb") as stderr_file:
|
|
completed = subprocess.run(
|
|
command,
|
|
stdout=stdout_file,
|
|
stderr=stderr_file,
|
|
timeout=int(cfg.get("timeout_seconds", 600)),
|
|
env=os.environ.copy(),
|
|
)
|
|
|
|
stdout_bytes = stdout_path.read_bytes()
|
|
stderr_bytes = stderr_path.read_bytes()
|
|
stdout_text = stdout_bytes.decode("utf-8", errors="replace")
|
|
stderr_text = stderr_bytes.decode("utf-8", errors="replace")
|
|
|
|
# Leere stderr-Dateien entfernen. Erfolgreiche, häufige Abfragen wie die
|
|
# Präsenzprüfung können ihre stdout-Rohdatei nach dem Einlesen verwerfen.
|
|
# Bei MeshCtrl-Fehlern bleibt die Datei unabhängig von retain_stdout erhalten.
|
|
if not stderr_bytes:
|
|
stderr_path.unlink(missing_ok=True)
|
|
if not retain_stdout and completed.returncode == 0:
|
|
stdout_path.unlink(missing_ok=True)
|
|
|
|
return subprocess.CompletedProcess(
|
|
args=command,
|
|
returncode=completed.returncode,
|
|
stdout=stdout_text,
|
|
stderr=stderr_text,
|
|
)
|
|
|
|
|
|
|
|
|
|
def _looks_truncated(output: str) -> bool:
|
|
cleaned = _strip_terminal_noise(output).rstrip()
|
|
if not cleaned:
|
|
return False
|
|
# Der beobachtete Fehler endet nahe 64 KiB mitten in einem JSON-Objekt.
|
|
return len(cleaned) >= 65000 and cleaned[-1] not in "]}"
|
|
|
|
|
|
def _extract_group_ids(payload: Any) -> list[tuple[str, str]]:
|
|
groups: list[tuple[str, str]] = []
|
|
items = payload if isinstance(payload, list) else list(payload.values()) if isinstance(payload, dict) else []
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
raw_id = item.get("_id") or item.get("id") or item.get("meshid")
|
|
name = item.get("name") or item.get("meshname") or str(raw_id or "Gruppe")
|
|
if not raw_id:
|
|
continue
|
|
group_id = str(raw_id)
|
|
if group_id.startswith("mesh//"):
|
|
group_id = group_id.split("/", 2)[-1]
|
|
groups.append((group_id, str(name)))
|
|
return groups
|
|
|
|
|
|
def _fetch_devices_by_group(
|
|
cfg: dict[str, Any],
|
|
include_details: bool,
|
|
log: Callable[[str, str], None],
|
|
*,
|
|
retain_stdout: bool = True,
|
|
) -> list[dict[str, Any]]:
|
|
log("warning", "Gesamtausgabe ist abgeschnitten. Geräte werden deshalb gruppenweise abgerufen.")
|
|
group_result = _run_meshctrl(cfg, "ListDeviceGroups", retain_stdout=retain_stdout)
|
|
if group_result.returncode != 0:
|
|
raise RuntimeError((group_result.stderr or group_result.stdout or "ListDeviceGroups fehlgeschlagen").strip())
|
|
groups = _extract_group_ids(_parse_json_output(group_result.stdout))
|
|
if not groups:
|
|
raise RuntimeError("MeshCtrl lieferte keine auswertbaren Gerätegruppen.")
|
|
log("info", f"{len(groups)} Gerätegruppe(n) gefunden.")
|
|
devices: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for index, (group_id, group_name) in enumerate(groups, start=1):
|
|
log("info", f"Gruppe {index}/{len(groups)} wird gelesen: {group_name}")
|
|
result = _run_meshctrl(
|
|
cfg,
|
|
"ListDevices",
|
|
include_details,
|
|
["--id", group_id],
|
|
retain_stdout=retain_stdout,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Gruppe {group_name}: {(result.stderr or result.stdout or 'MeshCtrl-Fehler').strip()}")
|
|
if _looks_truncated(result.stdout):
|
|
raise RuntimeError(f"Auch die Ausgabe der Gruppe {group_name} wurde bei etwa 64 KiB abgeschnitten.")
|
|
group_devices = _extract_devices(_parse_json_output(result.stdout))
|
|
log("success", f"Gruppe {group_name}: {len(group_devices)} Gerät(e) gelesen.")
|
|
for device in group_devices:
|
|
node_id = str(device.get("_id") or device.get("nodeid") or device.get("nodeId") or "")
|
|
key = node_id or json.dumps(device, sort_keys=True, default=str)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
devices.append(device)
|
|
return devices
|
|
|
|
|
|
def _device_summary(raw: dict[str, Any]) -> dict[str, str]:
|
|
"""Return a compact, robust device description for manual linking.
|
|
|
|
MeshCtrl output differs slightly between versions and between basic and
|
|
detail mode. The summary therefore checks the known paths first and then
|
|
uses conservative recursive fallbacks for manufacturer and serial number.
|
|
"""
|
|
node = raw.get("node") if isinstance(raw.get("node"), dict) else {}
|
|
sys_data = raw.get("sys") if isinstance(raw.get("sys"), dict) else {}
|
|
hardware = sys_data.get("hardware") if isinstance(sys_data.get("hardware"), dict) else {}
|
|
identifiers = hardware.get("identifiers") if isinstance(hardware.get("identifiers"), dict) else {}
|
|
|
|
node_id = _first(
|
|
raw,
|
|
"node._id",
|
|
"node.nodeid",
|
|
"node.nodeId",
|
|
"node.id",
|
|
"_id",
|
|
"nodeid",
|
|
"nodeId",
|
|
"id",
|
|
)
|
|
name = _first(raw, "node.name", "node.rname", "name", "rname", "hostname")
|
|
manufacturer = (
|
|
identifiers.get("bios_vendor")
|
|
or identifiers.get("board_vendor")
|
|
or _first(raw, "sys.hardware.windows.computer.Manufacturer", "node.manufacturer", "manufacturer")
|
|
or _find_key_recursive(raw, {"manufacturer", "bios_vendor", "board_vendor", "vendor"})
|
|
)
|
|
serial_number = (
|
|
identifiers.get("bios_serial")
|
|
or identifiers.get("board_serial")
|
|
or _first(raw, "sys.hardware.windows.bios.SerialNumber", "node.serial_number", "node.serialNumber", "serial_number", "serialNumber")
|
|
or _find_key_recursive(raw, {"serialnumber", "serial_number", "bios_serial", "board_serial"})
|
|
)
|
|
group = _first(raw, "node.groupname", "groupname", "meshname", "group")
|
|
operating_system = _first(raw, "node.osdesc", "osdesc", "operatingSystem", "os")
|
|
|
|
return {
|
|
"node_id": str(node_id or "").strip(),
|
|
"name": str(name or node_id or "").strip(),
|
|
"manufacturer": str(manufacturer or "").strip(),
|
|
"serial_number": str(serial_number or "").strip(),
|
|
"group": str(group or "").strip(),
|
|
"operating_system": str(operating_system or "").strip(),
|
|
}
|
|
|
|
|
|
def fetch_device_summaries_for_linking() -> list[dict[str, str]]:
|
|
"""Fetch MeshCentral devices on demand for the administrator link page.
|
|
|
|
Successful raw output is deleted immediately so this manual lookup does
|
|
not grow the diagnostics directory. Failed MeshCtrl calls are retained by
|
|
``_run_meshctrl`` for troubleshooting.
|
|
"""
|
|
cfg = load_config()["meshcentral"]
|
|
if not cfg.get("enabled"):
|
|
raise RuntimeError("MeshCentral-Synchronisierung ist in config.json deaktiviert.")
|
|
required = ["url", "username", "password", "meshctrl_path"]
|
|
missing = [key for key in required if not cfg.get(key)]
|
|
if missing:
|
|
raise RuntimeError("Fehlende MeshCentral-Konfiguration: " + ", ".join(missing))
|
|
|
|
attempts: list[str] = []
|
|
# Detail mode is preferred because manufacturer and serial number are
|
|
# normally only available there. Basic mode remains a compatibility
|
|
# fallback for older MeshCentral/MeshCtrl combinations.
|
|
for include_details in (True, False):
|
|
result = _run_meshctrl(
|
|
cfg,
|
|
"ListDevices",
|
|
include_details,
|
|
retain_stdout=False,
|
|
)
|
|
if result.returncode != 0:
|
|
attempts.append((result.stderr or result.stdout or "MeshCtrl-Fehler").strip())
|
|
continue
|
|
try:
|
|
if _looks_truncated(result.stdout):
|
|
devices = _fetch_devices_by_group(
|
|
cfg,
|
|
include_details,
|
|
lambda _level, _message: None,
|
|
retain_stdout=False,
|
|
)
|
|
else:
|
|
devices = _extract_devices(_parse_json_output(result.stdout))
|
|
except (ValueError, json.JSONDecodeError, RuntimeError) as exc:
|
|
attempts.append(str(exc))
|
|
continue
|
|
|
|
summaries: dict[str, dict[str, str]] = {}
|
|
for raw in devices:
|
|
summary = _device_summary(raw)
|
|
node_id = summary["node_id"]
|
|
if node_id:
|
|
summaries[node_id] = summary
|
|
if summaries:
|
|
return sorted(
|
|
summaries.values(),
|
|
key=lambda item: (item["name"].casefold(), item["node_id"].casefold()),
|
|
)
|
|
|
|
detail = " | ".join(item for item in attempts if item) or "keine auswertbare Geräteliste"
|
|
raise RuntimeError("MeshCentral-Geräteliste konnte nicht geladen werden: " + detail)
|
|
|
|
|
|
def fetch_devices(log: Callable[[str, str], None] | None = None) -> list[dict[str, Any]]:
|
|
log = log or (lambda level, message: None)
|
|
cfg = load_config()["meshcentral"]
|
|
if not cfg.get("enabled"):
|
|
raise RuntimeError("MeshCentral-Synchronisierung ist in config.json deaktiviert.")
|
|
required = ["url", "username", "password", "meshctrl_path"]
|
|
missing = [key for key in required if not cfg.get(key)]
|
|
if missing:
|
|
raise RuntimeError("Fehlende MeshCentral-Konfiguration: " + ", ".join(missing))
|
|
|
|
attempts: list[str] = []
|
|
use_details = cfg.get("include_details", True)
|
|
|
|
for include_details in ([True, False] if use_details else [False]):
|
|
mode = "mit Detaildaten" if include_details else "mit Basisdaten"
|
|
log("info", f"MeshCtrl-Abfrage wird {mode} gestartet.")
|
|
result = _run_meshctrl(cfg, "ListDevices", include_details)
|
|
log("debug", f"MeshCtrl beendet: Exitcode {result.returncode}, Ausgabe {len(result.stdout or '')} Zeichen.")
|
|
if result.returncode != 0:
|
|
message = (result.stderr or result.stdout or "MeshCtrl-Fehler").strip()
|
|
attempts.append(("mit Details: " if include_details else "ohne Details: ") + message)
|
|
log("error", f"MeshCtrl-Fehler: {message[:500]}")
|
|
continue
|
|
try:
|
|
if _looks_truncated(result.stdout):
|
|
log("warning", f"MeshCtrl-Ausgabe endet nach {len(result.stdout)} Zeichen mitten im JSON.")
|
|
devices = _fetch_devices_by_group(cfg, include_details, log)
|
|
else:
|
|
devices = _extract_devices(_parse_json_output(result.stdout))
|
|
log("success", f"{len(devices)} Gerät(e) aus MeshCentral gelesen.")
|
|
return devices
|
|
except (ValueError, json.JSONDecodeError, RuntimeError) as exc:
|
|
attempts.append(("mit Details: " if include_details else "ohne Details: ") + str(exc))
|
|
log("warning", f"JSON-Auswertung {mode} fehlgeschlagen: {exc}")
|
|
# Manche MeshCentral-/MeshCtrl-Versionen liefern bei --details einen
|
|
# beschädigten sehr großen JSON-Block. Dann automatisch Basisdaten abrufen.
|
|
continue
|
|
|
|
raise RuntimeError("MeshCentral-Import fehlgeschlagen. " + " | ".join(attempts))
|
|
|
|
|
|
def _bytes_to_gb(value: Any, binary: bool = False) -> str | None:
|
|
try:
|
|
size = int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
divisor = 1024 ** 3 if binary else 1000 ** 3
|
|
number = round(size / divisor, 2)
|
|
return f"{number:g}"
|
|
|
|
|
|
|
|
def _path_values(data: Any, path: str) -> list[Any]:
|
|
"""Liest Punktpfade; mehrere Alternativen können mit | getrennt werden."""
|
|
for alternative in [item.strip() for item in str(path or "").split("|") if item.strip()]:
|
|
values = [data]
|
|
for part in alternative.split('.'):
|
|
next_values: list[Any] = []
|
|
list_mode = part.endswith('[]')
|
|
key = part[:-2] if list_mode else part
|
|
for value in values:
|
|
candidate = value.get(key) if isinstance(value, dict) else None
|
|
if list_mode and isinstance(candidate, list):
|
|
next_values.extend(candidate)
|
|
elif candidate is not None:
|
|
next_values.append(candidate)
|
|
values = next_values
|
|
if not values:
|
|
break
|
|
if values:
|
|
return values
|
|
return []
|
|
|
|
|
|
def _network_entries(value: Any) -> list[dict[str, Any]]:
|
|
entries: list[dict[str, Any]] = []
|
|
if isinstance(value, dict):
|
|
for group in value.values():
|
|
if isinstance(group, list):
|
|
entries.extend(item for item in group if isinstance(item, dict))
|
|
elif isinstance(value, list):
|
|
entries.extend(item for item in value if isinstance(item, dict))
|
|
return entries
|
|
|
|
|
|
def _convert_mapping_value(values: list[Any], transform: str, separator: str) -> Any:
|
|
if not values:
|
|
return None
|
|
source = values[0]
|
|
if transform == 'active_ipv4_address':
|
|
for entry in _network_entries(source):
|
|
if entry.get('family') == 'IPv4' and entry.get('status') == 'up' and entry.get('type') != 'loopback':
|
|
return entry.get('address')
|
|
return None
|
|
if transform == 'active_ipv4_mac':
|
|
for entry in _network_entries(source):
|
|
if entry.get('family') == 'IPv4' and entry.get('status') == 'up' and entry.get('type') != 'loopback':
|
|
return _normalize_mac(entry.get('mac'))
|
|
return None
|
|
if transform == 'normalize_serial':
|
|
return _normalize_serial(source)
|
|
if transform == 'normalize_mac':
|
|
return _normalize_mac(source)
|
|
if transform == 'sum_memory_gb':
|
|
modules = source if isinstance(source, list) else values
|
|
total = 0
|
|
for module in modules:
|
|
if isinstance(module, dict):
|
|
try: total += int(module.get('Capacity') or 0)
|
|
except (TypeError, ValueError): pass
|
|
return round(total / (1024 ** 3), 2) if total else None
|
|
if transform == 'format_memory_modules':
|
|
modules = source if isinstance(source, list) else values
|
|
lines: list[str] = []
|
|
for module in modules:
|
|
if not isinstance(module, dict):
|
|
continue
|
|
try: capacity = int(module.get('Capacity') or 0)
|
|
except (TypeError, ValueError): capacity = 0
|
|
parts = [
|
|
str(module.get('DeviceLocator') or 'Steckplatz unbekannt'),
|
|
str(module.get('Description') or 'Speichermodul'),
|
|
f"{_bytes_to_gb(capacity, binary=True)} GB" if capacity else None,
|
|
str(module.get('Manufacturer') or '').strip() or None,
|
|
str(module.get('PartNumber') or '').strip() or None,
|
|
f"{module.get('Speed')} MHz" if module.get('Speed') else None,
|
|
]
|
|
lines.append(' | '.join(part for part in parts if part))
|
|
return '\n────────────────────\n'.join(lines) or None
|
|
if transform == 'sum_storage_gb':
|
|
drives = source if isinstance(source, list) else values
|
|
total = 0
|
|
for drive in drives:
|
|
if isinstance(drive, dict):
|
|
try: total += int(drive.get('Size') or 0)
|
|
except (TypeError, ValueError): pass
|
|
return round(total / (1000 ** 3), 2) if total else None
|
|
if transform == 'format_drives':
|
|
drives = source if isinstance(source, list) else values
|
|
lines: list[str] = []
|
|
for drive in drives:
|
|
if not isinstance(drive, dict):
|
|
continue
|
|
try: size = int(drive.get('Size') or 0)
|
|
except (TypeError, ValueError): size = 0
|
|
model = drive.get('Model') or drive.get('Caption') or 'Unbekannter Datenträger'
|
|
size_text = _bytes_to_gb(size) if size else None
|
|
lines.append(f"{model} | {size_text} GB" if size_text else str(model))
|
|
return '\n────────────────────\n'.join(lines) or None
|
|
if transform == 'gpu_names':
|
|
gpu = source if isinstance(source, list) else values
|
|
names = [str(item.get('Name')) for item in gpu if isinstance(item, dict) and item.get('Name')]
|
|
return separator.join(dict.fromkeys(names)) or None
|
|
if transform == 'active_antivirus':
|
|
products = source if isinstance(source, list) else values
|
|
names = [str(item.get('product')) for item in products if isinstance(item, dict) and item.get('enabled') is True and item.get('product')]
|
|
return separator.join(dict.fromkeys(names)) or None
|
|
return None
|
|
|
|
|
|
def _mapped_value(raw: dict[str, Any], mapping: dict[str, Any]) -> Any:
|
|
values = _path_values(raw, str(mapping.get('source_path') or ''))
|
|
mode = str(mapping.get('multi_value_mode') or 'first')
|
|
transform = str(mapping.get('transform') or 'raw')
|
|
separator = str(mapping.get('separator') or '\n')
|
|
|
|
special = {
|
|
'active_ipv4_address', 'active_ipv4_mac', 'normalize_serial', 'normalize_mac',
|
|
'sum_memory_gb', 'format_memory_modules', 'sum_storage_gb', 'format_drives',
|
|
'gpu_names', 'active_antivirus'
|
|
}
|
|
if transform in special:
|
|
return _convert_mapping_value(values, transform, separator)
|
|
|
|
def convert(value: Any) -> Any:
|
|
if transform == 'integer':
|
|
try: return int(value)
|
|
except (TypeError, ValueError): return None
|
|
if transform == 'decimal':
|
|
try: return float(value)
|
|
except (TypeError, ValueError): return None
|
|
if transform == 'boolean':
|
|
if isinstance(value, bool): return value
|
|
return str(value).strip().lower() in {'1', 'true', 'yes', 'on', 'ja'}
|
|
if transform == 'string': return str(value)
|
|
if transform == 'json': return value
|
|
return value
|
|
|
|
converted = [convert(value) for value in values if value not in (None, '', [], {})]
|
|
converted = [value for value in converted if value is not None]
|
|
if not converted:
|
|
return None
|
|
if mode == 'json': return converted if len(converted) > 1 else converted[0]
|
|
if mode == 'count': return len(converted)
|
|
if mode == 'join':
|
|
return separator.join(json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else str(value) for value in converted)
|
|
return converted[0]
|
|
|
|
|
|
def _set_custom_value(db: Session, asset: Asset, definition: FieldDefinition, value: Any) -> None:
|
|
if asset.id is None:
|
|
db.flush()
|
|
row = db.query(AssetFieldValue).filter(AssetFieldValue.asset_id==asset.id, AssetFieldValue.field_definition_id==definition.id).first()
|
|
if not row: row=AssetFieldValue(asset_id=asset.id, field_definition_id=definition.id); db.add(row)
|
|
row.value_text=row.value_number=row.value_date=None; row.value_boolean=None; row.value_datetime=None; row.value_json=None
|
|
if definition.data_type == 'json': row.value_json=value
|
|
elif definition.data_type == 'boolean': row.value_boolean=bool(value)
|
|
elif definition.data_type in {'integer','decimal'}: row.value_number=str(value)
|
|
elif definition.data_type == 'date': row.value_date=str(value)
|
|
else: row.value_text=str(value)
|
|
|
|
|
|
def _is_empty_value(value: Any) -> bool:
|
|
return value in (None, "", [], {})
|
|
|
|
|
|
def _display_mapping_value(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
|
return str(value)
|
|
|
|
|
|
def _values_equal(current: Any, incoming: Any) -> bool:
|
|
if _is_empty_value(current) and _is_empty_value(incoming):
|
|
return True
|
|
if isinstance(current, (dict, list)) or isinstance(incoming, (dict, list)):
|
|
return _display_mapping_value(current) == _display_mapping_value(incoming)
|
|
return str(current) == str(incoming)
|
|
|
|
|
|
def _custom_value(row: AssetFieldValue | None, definition: FieldDefinition) -> Any:
|
|
if row is None:
|
|
return None
|
|
if definition.data_type == "json":
|
|
return row.value_json
|
|
if definition.data_type in {"integer", "decimal"}:
|
|
return row.value_number
|
|
if definition.data_type == "boolean":
|
|
return row.value_boolean
|
|
if definition.data_type == "date":
|
|
return row.value_date
|
|
if definition.data_type == "datetime":
|
|
return row.value_datetime
|
|
return row.value_text
|
|
|
|
|
|
def _coerce_system_value(target: str, value: Any) -> Any:
|
|
# Die meisten Asset-Spalten sind Textfelder. Integer-/Fremdschlüsselfelder
|
|
# werden typgerecht geschrieben, damit dynamische Mappings auch dort sicher sind.
|
|
if target in {"mesh_mtype", "parent_asset_id", "category_id"}:
|
|
try:
|
|
return int(value) if value not in (None, "") else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if isinstance(value, (dict, list)):
|
|
return json.dumps(value, ensure_ascii=False)
|
|
return value if value is None else str(value)
|
|
|
|
|
|
def map_device(raw: dict[str, Any], field_mappings: list[dict[str, Any]] | None = None) -> dict[str, Any]:
|
|
"""Erzeugt Assetwerte ausschließlich aus den aktivierten Datenbank-Mappings."""
|
|
mapped: dict[str, Any] = {}
|
|
for mapping in field_mappings or []:
|
|
if not isinstance(mapping, dict) or not mapping.get('enabled', True):
|
|
continue
|
|
target = str(mapping.get('target_field') or '').strip()
|
|
if not target:
|
|
continue
|
|
value = _mapped_value(raw, mapping)
|
|
if value not in (None, '', [], {}):
|
|
# Niedrigere Priorität wird zuerst verarbeitet; spätere Regeln dürfen überschreiben.
|
|
mapped[target] = value
|
|
return mapped
|
|
|
|
def _find_asset(db: Session, mapped: dict[str, Any], order: list[str]) -> tuple[Asset | None, str | None, str | None]:
|
|
for method in order:
|
|
if method == "node_id" and mapped.get("mesh_node_id"):
|
|
matches = db.query(Asset).filter(Asset.mesh_node_id == mapped["mesh_node_id"]).all()
|
|
elif method == "serial_number" and mapped.get("serial_number"):
|
|
matches = db.query(Asset).filter(func.lower(Asset.serial_number) == mapped["serial_number"].lower()).all()
|
|
else:
|
|
continue
|
|
if len(matches) == 1:
|
|
return matches[0], method, None
|
|
if len(matches) > 1:
|
|
return None, method, f"Mehrere Assets stimmen über {method} überein."
|
|
return None, None, None
|
|
|
|
|
|
|
|
|
|
def _first_nonempty_path(raw: dict[str, Any], paths: list[str]) -> Any:
|
|
for path in paths:
|
|
value: Any = raw
|
|
for part in path.split('.'):
|
|
if not isinstance(value, dict) or part not in value:
|
|
value = None
|
|
break
|
|
value = value.get(part)
|
|
if value not in (None, '', [], {}):
|
|
return value
|
|
return None
|
|
|
|
|
|
def _inventory_list(value: Any) -> list[Any]:
|
|
return value if isinstance(value, list) else []
|
|
|
|
|
|
def _inventory_dict(value: Any) -> dict[str, Any]:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _network_inventory(raw: dict[str, Any]) -> list[dict[str, Any]]:
|
|
net = _inventory_dict(raw.get("net"))
|
|
interfaces = _inventory_dict(net.get("netif2"))
|
|
result: list[dict[str, Any]] = []
|
|
for adapter_name, entries_value in interfaces.items():
|
|
entries = _inventory_list(entries_value)
|
|
if not entries:
|
|
continue
|
|
non_loopback = [entry for entry in entries if isinstance(entry, dict) and entry.get("type") != "loopback"]
|
|
if not non_loopback:
|
|
continue
|
|
ipv4 = [str(entry.get("address")) for entry in non_loopback if entry.get("family") == "IPv4" and entry.get("address")]
|
|
ipv6 = [str(entry.get("address")) for entry in non_loopback if entry.get("family") == "IPv6" and entry.get("address")]
|
|
macs = list(dict.fromkeys(str(entry.get("mac")) for entry in non_loopback if entry.get("mac") and entry.get("mac") != "00:00:00:00:00:00"))
|
|
gateways = list(dict.fromkeys(str(entry.get("gateway")) for entry in non_loopback if entry.get("gateway")))
|
|
netmasks = list(dict.fromkeys(str(entry.get("netmask")) for entry in non_loopback if entry.get("netmask")))
|
|
fqdns = list(dict.fromkeys(str(entry.get("fqdn")) for entry in non_loopback if entry.get("fqdn")))
|
|
statuses = list(dict.fromkeys(str(entry.get("status")) for entry in non_loopback if entry.get("status")))
|
|
types = list(dict.fromkeys(str(entry.get("type")) for entry in non_loopback if entry.get("type")))
|
|
indices = list(dict.fromkeys(entry.get("index") for entry in non_loopback if entry.get("index") is not None))
|
|
record: dict[str, Any] = {"inventory.property.name": adapter_name}
|
|
if statuses: record["inventory.property.status"] = statuses[0] if len(statuses) == 1 else statuses
|
|
if types: record["inventory.property.type"] = types[0] if len(types) == 1 else types
|
|
if macs: record["inventory.property.mac_address"] = macs[0] if len(macs) == 1 else macs
|
|
if ipv4: record["inventory.property.ipv4_address"] = ipv4[0] if len(ipv4) == 1 else ipv4
|
|
if ipv6: record["inventory.property.ipv6_address"] = ipv6[0] if len(ipv6) == 1 else ipv6
|
|
if gateways: record["inventory.property.gateway"] = gateways[0] if len(gateways) == 1 else gateways
|
|
if netmasks: record["inventory.property.netmask"] = netmasks[0] if len(netmasks) == 1 else netmasks
|
|
if fqdns: record["inventory.property.dns_suffix"] = fqdns[0] if len(fqdns) == 1 else fqdns
|
|
if indices: record["inventory.property.interface_index"] = indices[0] if len(indices) == 1 else indices
|
|
result.append(record)
|
|
return result
|
|
|
|
|
|
def _security_inventory(raw: dict[str, Any]) -> dict[str, Any]:
|
|
node = _inventory_dict(raw.get("node"))
|
|
antivirus_entries = _inventory_list(node.get("av"))
|
|
antivirus: list[dict[str, Any]] = []
|
|
seen: set[tuple[Any, Any, Any]] = set()
|
|
for item in antivirus_entries:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
key = (item.get("product"), item.get("enabled"), item.get("updated"))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
antivirus.append({
|
|
"inventory.property.name": item.get("product") or "inventory.value.unknown",
|
|
"inventory.property.enabled": item.get("enabled"),
|
|
"inventory.property.up_to_date": item.get("updated"),
|
|
})
|
|
result: dict[str, Any] = {}
|
|
if antivirus:
|
|
result["inventory.group.antivirus"] = antivirus
|
|
if isinstance(node.get("wsc"), dict) and node.get("wsc"):
|
|
result["inventory.group.windows_security_center"] = node.get("wsc")
|
|
if isinstance(node.get("defender"), dict) and node.get("defender"):
|
|
result["inventory.group.microsoft_defender"] = node.get("defender")
|
|
return result
|
|
|
|
|
|
def _build_hardware_inventory(raw: dict[str, Any]) -> dict[str, Any]:
|
|
"""Build a language-neutral inventory. Known labels are i18n keys."""
|
|
sys_data = _inventory_dict(raw.get("sys"))
|
|
hardware = _inventory_dict(sys_data.get("hardware"))
|
|
windows = _inventory_dict(hardware.get("windows"))
|
|
identifiers = _inventory_dict(hardware.get("identifiers"))
|
|
node = _inventory_dict(raw.get("node"))
|
|
inventory: dict[str, Any] = {}
|
|
|
|
system_overview = {
|
|
"inventory.property.device_name": node.get("name") or node.get("rname"),
|
|
"inventory.property.manufacturer": identifiers.get("bios_vendor") or identifiers.get("board_vendor"),
|
|
"inventory.property.model": identifiers.get("product_name"),
|
|
"inventory.property.product_uuid": identifiers.get("product_uuid"),
|
|
"inventory.property.serial_number": identifiers.get("bios_serial"),
|
|
"inventory.property.mesh_group": node.get("groupname"),
|
|
"inventory.property.operating_system": node.get("osdesc"),
|
|
}
|
|
system_overview = {k:v for k,v in system_overview.items() if v not in (None,"",[],{})}
|
|
if system_overview: inventory["inventory.section.system_overview"] = system_overview
|
|
|
|
bios = {
|
|
"inventory.property.manufacturer": identifiers.get("bios_vendor"),
|
|
"inventory.property.version": identifiers.get("bios_version"),
|
|
"inventory.property.date": identifiers.get("bios_date"),
|
|
"inventory.property.mode": identifiers.get("bios_mode"),
|
|
"inventory.property.serial_number": identifiers.get("bios_serial"),
|
|
}
|
|
bios = {k:v for k,v in bios.items() if v not in (None,"",[],{})}
|
|
if bios: inventory["inventory.section.bios"] = bios
|
|
|
|
board = {
|
|
"inventory.property.manufacturer": identifiers.get("board_vendor"),
|
|
"inventory.property.model": identifiers.get("board_name"),
|
|
"inventory.property.version": identifiers.get("board_version"),
|
|
"inventory.property.serial_number": identifiers.get("board_serial"),
|
|
}
|
|
board = {k:v for k,v in board.items() if v not in (None,"",[],{})}
|
|
if board: inventory["inventory.section.mainboard"] = board
|
|
|
|
if windows.get("cpu"): inventory["inventory.section.processors"] = windows.get("cpu")
|
|
if windows.get("memory"): inventory["inventory.section.memory"] = windows.get("memory")
|
|
|
|
storage: dict[str, Any] = {}
|
|
if windows.get("drives"): storage["inventory.group.physical_drives"] = windows.get("drives")
|
|
if identifiers.get("storage_devices"): storage["inventory.group.device_identifiers"] = identifiers.get("storage_devices")
|
|
if windows.get("partitions"): storage["inventory.group.partitions"] = windows.get("partitions")
|
|
if windows.get("volumes"): storage["inventory.group.volumes"] = windows.get("volumes")
|
|
if storage: inventory["inventory.section.storage"] = storage
|
|
|
|
if windows.get("gpu"): inventory["inventory.section.graphics"] = windows.get("gpu")
|
|
|
|
network_details: dict[str, Any] = {}
|
|
adapters = _network_inventory(raw)
|
|
if adapters: network_details["inventory.group.adapters"] = adapters
|
|
hardware_network = _inventory_dict(hardware.get("network"))
|
|
if hardware_network.get("dns"): network_details["inventory.group.dns_servers"] = hardware_network.get("dns")
|
|
if network_details: inventory["inventory.section.network"] = network_details
|
|
|
|
if hardware.get("tpm"): inventory["inventory.section.tpm"] = hardware.get("tpm")
|
|
if windows.get("osinfo"): inventory["inventory.section.os_details"] = windows.get("osinfo")
|
|
security = _security_inventory(raw)
|
|
if security: inventory["inventory.section.security"] = security
|
|
|
|
runtime: dict[str, Any] = {}
|
|
if node.get("agent"): runtime["inventory.group.mesh_agent"] = node.get("agent")
|
|
if hardware.get("agentvers"): runtime["inventory.group.agent_build"] = hardware.get("agentvers")
|
|
if node.get("intelamt"): runtime["inventory.group.intel_amt"] = node.get("intelamt")
|
|
if node.get("users"): runtime["inventory.group.known_users"] = node.get("users")
|
|
if node.get("lusers"): runtime["inventory.group.logged_on_users"] = node.get("lusers")
|
|
if node.get("lastbootuptime"): runtime["inventory.property.last_boot_unix_ms"] = node.get("lastbootuptime")
|
|
if raw.get("lastConnect"): runtime["inventory.group.last_connection"] = raw.get("lastConnect")
|
|
if runtime: inventory["inventory.section.mesh_runtime"] = runtime
|
|
|
|
known_hardware_keys = {"agentvers", "identifiers", "network", "tpm", "windows"}
|
|
extra_hardware = {k:v for k,v in hardware.items() if k not in known_hardware_keys and v not in (None,"",[],{})}
|
|
if extra_hardware: inventory["inventory.section.other_hardware"] = extra_hardware
|
|
known_windows_keys = {"cpu", "memory", "drives", "partitions", "volumes", "gpu", "osinfo", "software"}
|
|
extra_windows = {k:v for k,v in windows.items() if k not in known_windows_keys and v not in (None,"",[],{})}
|
|
if extra_windows: inventory["inventory.section.other_windows_hardware"] = extra_windows
|
|
return inventory
|
|
|
|
def _inventory_segments(raw: dict[str, Any]) -> tuple[dict[str, Any], list[Any] | dict[str, Any]]:
|
|
"""Extract readable inventory snapshots while retaining mesh_source_data as raw truth."""
|
|
hardware = _build_hardware_inventory(raw)
|
|
software = _first_nonempty_path(raw, [
|
|
'sys.software', 'sys.hardware.windows.software', 'software',
|
|
'installedSoftware', 'installedApplications', 'node.software'
|
|
])
|
|
if not isinstance(software, (list, dict)):
|
|
software = []
|
|
return hardware, software
|
|
|
|
def synchronize(db: Session, log: Callable[[str, str], None] | None = None) -> SyncResult:
|
|
log = log or (lambda level, message: None)
|
|
cfg = load_config()["meshcentral"]
|
|
run = SyncRun(status="running")
|
|
db.add(run)
|
|
db.commit()
|
|
db.refresh(run)
|
|
now = datetime.utcnow()
|
|
seen_node_ids: set[str] = set()
|
|
mapping_rows = db.query(MeshFieldMapping).filter(MeshFieldMapping.enabled.is_(True)).order_by(MeshFieldMapping.priority, MeshFieldMapping.id).all()
|
|
field_mappings = [{
|
|
"source_path": row.source_path, "target_field": row.target_field_name,
|
|
"transform": row.transform, "multi_value_mode": row.multi_value_mode,
|
|
"separator": row.separator, "enabled": row.enabled, "update_rule": row.update_rule,
|
|
} for row in mapping_rows]
|
|
mapping_rules = {row.target_field_name: row.update_rule for row in mapping_rows}
|
|
field_definitions = {
|
|
definition.field_name: definition
|
|
for definition in db.query(FieldDefinition).filter(FieldDefinition.is_active.is_(True)).all()
|
|
}
|
|
|
|
try:
|
|
log("info", f"{len(mapping_rows)} aktive Feld-Mapping(s) geladen.")
|
|
unknown_targets = sorted({row.target_field_name for row in mapping_rows if row.target_field_name not in field_definitions})
|
|
for target in unknown_targets:
|
|
log("error", f"Aktives Mapping verweist auf unbekanntes oder inaktives Zielfeld '{target}'.")
|
|
run.error_count += 1
|
|
missing_columns = sorted({
|
|
row.target_field_name for row in mapping_rows
|
|
if row.target_field_name in field_definitions
|
|
and field_definitions[row.target_field_name].is_system
|
|
and not hasattr(Asset, row.target_field_name)
|
|
})
|
|
for target in missing_columns:
|
|
log("error", f"Systemfeld '{target}' besitzt keine Asset-Datenbankspalte und wird übersprungen.")
|
|
run.error_count += 1
|
|
log("info", "Synchronisierung gestartet.")
|
|
devices = fetch_devices(log)
|
|
run.devices_found = len(devices)
|
|
log("info", f"Verarbeite {len(devices)} Gerät(e).")
|
|
fallback_id = int(cfg.get("fallback_category_id") or 1)
|
|
fallback_category = db.query(Category).filter(Category.id == fallback_id).first()
|
|
if fallback_category is None and fallback_id != 1:
|
|
fallback_category = db.query(Category).filter(Category.id == 1).first()
|
|
if fallback_category is None:
|
|
fallback_category = db.query(Category).order_by(Category.id.asc()).first()
|
|
if fallback_category is None:
|
|
raise RuntimeError("Es ist keine Asset-Kategorie vorhanden. Bitte zuerst eine Kategorie anlegen.")
|
|
category_mapping = cfg.get("category_mapping") or {}
|
|
log("info", f"Fallback-Kategorie: {fallback_category.name} (ID {fallback_category.id}).")
|
|
|
|
for raw in devices:
|
|
try:
|
|
mapped = map_device(raw, field_mappings)
|
|
node_id = mapped.get("mesh_node_id")
|
|
if not node_id:
|
|
run.error_count += 1
|
|
log("error", "Gerät ohne Node-ID wurde übersprungen.")
|
|
continue
|
|
seen_node_ids.add(node_id)
|
|
asset, match_method, match_error = _find_asset(db, mapped, cfg.get("match_order", ["node_id", "serial_number"]))
|
|
if match_error:
|
|
run.conflict_count += 1
|
|
log("warning", f"{mapped.get('name', node_id)}: {match_error}")
|
|
continue
|
|
is_new = asset is None
|
|
if asset is None:
|
|
if not cfg.get("create_missing_assets", True):
|
|
continue
|
|
device_type = str(mapped.get("mesh_mtype") or 0)
|
|
mapped_category_id = category_mapping.get(device_type)
|
|
try:
|
|
mapped_category_id = int(mapped_category_id) if mapped_category_id not in (None, "") else None
|
|
except (TypeError, ValueError):
|
|
mapped_category_id = None
|
|
category = db.query(Category).filter(Category.id == mapped_category_id).first() if mapped_category_id else None
|
|
if category is None:
|
|
category = fallback_category
|
|
log("warning", f"{mapped.get('name', node_id)}: keine gültige Kategorie für MeshCentral-Typ {device_type}; Fallback '{category.name}' (ID {category.id}) verwendet.")
|
|
asset = Asset(category_id=category.id, name=mapped.get("name", node_id), mesh_node_id=node_id)
|
|
db.add(asset)
|
|
run.created_count += 1
|
|
log("success", f"{mapped.get('name', node_id)}: neues Asset in Kategorie '{category.name}' (ID {category.id}) angelegt.")
|
|
elif match_method == "serial_number" and not asset.mesh_node_id:
|
|
asset.mesh_node_id = node_id
|
|
log("info", f"{asset.name}: über Seriennummer zugeordnet.")
|
|
|
|
conflicts: dict[str, Any] = {}
|
|
history_changes: dict[str, dict[str, str]] = {}
|
|
changed = False
|
|
|
|
# Ausschließlich die aktiven Mapping-Regeln bestimmen, welche
|
|
# System- und Custom-Felder synchronisiert werden. Es gibt keine
|
|
# feste Feld-Whitelist mehr.
|
|
for target, incoming in mapped.items():
|
|
if target == "mesh_device_type" or incoming in (None, "", [], {}):
|
|
continue
|
|
|
|
definition = field_definitions.get(target)
|
|
if definition is None:
|
|
# Das Zielfeld muss im zentralen Feldkatalog existieren.
|
|
# Die Prüfung wird zusätzlich vor der Geräteschleife protokolliert.
|
|
continue
|
|
|
|
rule = mapping_rules.get(target, "fill_empty")
|
|
label = definition.label or ASSET_FIELDS.get(target, target)
|
|
|
|
if definition.is_system:
|
|
if not hasattr(asset, target):
|
|
log("error", f"{asset.name}: Systemfeld '{target}' besitzt keine Asset-Datenbankspalte.")
|
|
run.error_count += 1
|
|
continue
|
|
current = getattr(asset, target)
|
|
else:
|
|
current_row = db.query(AssetFieldValue).filter(
|
|
AssetFieldValue.asset_id == asset.id,
|
|
AssetFieldValue.field_definition_id == definition.id,
|
|
).first()
|
|
current = _custom_value(current_row, definition)
|
|
|
|
if _values_equal(current, incoming):
|
|
continue
|
|
|
|
should_write = _is_empty_value(current) or rule == "meshcentral_wins"
|
|
if rule == "local_wins" and not _is_empty_value(current):
|
|
should_write = False
|
|
|
|
if should_write:
|
|
old_display = _display_mapping_value(current)
|
|
new_display = _display_mapping_value(incoming)
|
|
if definition.is_system:
|
|
setattr(asset, target, _coerce_system_value(target, incoming))
|
|
else:
|
|
_set_custom_value(db, asset, definition, incoming)
|
|
history_changes[label] = {"old": old_display, "new": new_display}
|
|
changed = True
|
|
log("debug", f"{asset.name}: Mapping '{target}' angewendet ({rule}).")
|
|
elif rule == "fill_empty":
|
|
conflicts[target] = {
|
|
"local": current,
|
|
"meshcentral": incoming,
|
|
"rule": rule,
|
|
}
|
|
|
|
if history_changes:
|
|
db.flush()
|
|
db.add(AssetHistory(
|
|
asset_id=asset.id,
|
|
source="meshcentral-create" if is_new else "meshcentral-sync",
|
|
changed_by="MeshCentral",
|
|
changes=history_changes,
|
|
))
|
|
|
|
asset.mesh_last_sync = now
|
|
reported_last_seen = extract_last_seen(raw)
|
|
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
|
|
asset.mesh_source_data = raw if cfg.get("store_source_json", True) else {}
|
|
hardware_snapshot, software_snapshot = _inventory_segments(raw)
|
|
asset.mesh_hardware_data = hardware_snapshot
|
|
asset.mesh_software_data = software_snapshot
|
|
asset.mesh_inventory_updated_at = now
|
|
asset.mesh_conflicts = conflicts
|
|
if conflicts:
|
|
asset.mesh_sync_status = "conflict"
|
|
asset.mesh_sync_message = f"{len(conflicts)} Feldkonflikt(e)"
|
|
run.conflict_count += 1
|
|
log("warning", f"{asset.name}: {len(conflicts)} Feldkonflikt(e).")
|
|
elif is_new:
|
|
asset.mesh_sync_status = "created"
|
|
asset.mesh_sync_message = "Neu aus MeshCentral importiert"
|
|
else:
|
|
asset.mesh_sync_status = "updated" if changed else "synced"
|
|
asset.mesh_sync_message = "Über Seriennummer zugeordnet" if match_method == "serial_number" else "Synchronisiert"
|
|
if changed:
|
|
run.updated_count += 1
|
|
log("info", f"{asset.name}: Daten aktualisiert.")
|
|
else:
|
|
log("debug", f"{asset.name}: keine Änderungen.")
|
|
except Exception as item_error:
|
|
run.error_count += 1
|
|
log("error", f"Gerätefehler: {item_error}")
|
|
run.message = ((run.message or "") + f"\nGerätefehler: {item_error}").strip()
|
|
|
|
if cfg.get("mark_missing_devices", True):
|
|
linked = db.query(Asset).filter(Asset.mesh_node_id.isnot(None)).all()
|
|
for asset in linked:
|
|
if asset.mesh_node_id not in seen_node_ids:
|
|
asset.mesh_sync_status = "missing"
|
|
asset.mesh_sync_message = "Beim letzten MeshCentral-Import nicht gefunden"
|
|
log("warning", f"{asset.name}: in MeshCentral nicht mehr gefunden.")
|
|
|
|
run.status = "success" if run.error_count == 0 else "partial"
|
|
run.finished_at = datetime.utcnow()
|
|
db.commit()
|
|
log("success", f"Synchronisierung beendet: {run.created_count} neu, {run.updated_count} geändert, {run.conflict_count} Konflikte, {run.error_count} Fehler.")
|
|
except Exception as error:
|
|
db.rollback()
|
|
run = db.query(SyncRun).filter(SyncRun.id == run.id).first()
|
|
run.status = "error"
|
|
run.error_count += 1
|
|
run.message = str(error)
|
|
run.finished_at = datetime.utcnow()
|
|
db.commit()
|
|
log("error", f"Synchronisierung abgebrochen: {error}")
|
|
return SyncResult(run=run)
|
|
from .presence_state import extract_last_seen
|
|
|