import json
import os
import re
import subprocess
import unicodedata
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 can emit multiple complete JSON blocks in sequence.
        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:
    """Remove BOM markers, ANSI control sequences, and NUL bytes from MeshCtrl output."""
    output = output.lstrip("\ufeff")
    output = output.replace("\x00", "")
    return re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", output)


def _parse_json_output(output: str) -> Any:
    """Parse JSON even when surrounded by notices or when multiple JSON blocks are present."""
    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:
            # An invalid detail block must not produce a misleading default message.
            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
    # possible pipe or buffer limits with very large MeshCtrl JSON output.
    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")

    # Remove empty stderr files. Successful high-frequency queries such as the
    # presence check may discard their raw stdout file after it has been parsed.
    # For MeshCtrl errors, the file is retained regardless of retain_stdout.
    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
    # The observed failure ends near 64 KiB in the middle of a JSON object.
    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}")
            # Some MeshCentral or MeshCtrl versions return a
            # corrupted, very large JSON block with --details. Fall back to basic data automatically.
            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]:
    """Read dotted paths; multiple alternatives may be separated with |."""
    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]]:
    """Flatten MeshCentral network entries while retaining the interface name.

    ``net.netif2`` is keyed by interface name. Keeping that name allows us to
    prefer real/primary interfaces over loopback, container and tunnel entries.
    """
    entries: list[dict[str, Any]] = []
    if isinstance(value, dict):
        for adapter_name, group in value.items():
            if not isinstance(group, list):
                continue
            for item in group:
                if not isinstance(item, dict):
                    continue
                entry = dict(item)
                entry.setdefault('_adapter_name', str(adapter_name or ''))
                entries.append(entry)
    elif isinstance(value, list):
        entries.extend(dict(item) for item in value if isinstance(item, dict))
    return entries


def _network_interface_rank(name: Any) -> int:
    """Return a stable preference rank for common physical/virtual NIC names."""
    value = str(name or '').strip().casefold()
    if not value:
        return 80

    # Explicitly de-prioritize interfaces that are usually local/container/tunnel
    # plumbing and should not become the AssetManager management address.
    low_priority = (
        'loopback', 'docker', 'veth', 'virbr', 'tun', 'tap', 'tailscale',
        'wg', 'zt', 'br-',
    )
    if value == 'lo' or value.startswith(low_priority):
        return 200

    # Common Linux, Windows and hypervisor interface names. Comparison is
    # intentionally case-insensitive.
    if value.startswith(('ethernet', 'eth')):
        return 10
    if value.startswith(('eno', 'ens', 'enp', 'enx')):
        return 15
    if value == 'vmbr0':
        return 20
    if value.startswith('vmbr'):
        return 22
    if value.startswith(('bond', 'team')):
        return 25
    if value.startswith(('wlan', 'wlp', 'wifi', 'wi-fi', 'wireless')):
        return 30
    if value.startswith(('vethernet', 'hyper-v', 'vmnet')):
        return 35
    if value.startswith('br'):
        return 40
    return 60


def _network_entry_active(entry: dict[str, Any]) -> bool:
    status = str(entry.get('status') or '').strip().casefold()
    return status not in {'down', 'inactive', 'disabled', 'disconnected', 'notpresent'}


def _valid_ipv4(value: Any) -> bool:
    try:
        import ipaddress
        address = ipaddress.ip_address(str(value or '').strip())
    except ValueError:
        return False
    return address.version == 4 and not address.is_loopback and not address.is_unspecified


def _valid_network_mac(value: Any) -> str | None:
    mac = _normalize_mac(value)
    if not mac:
        return None
    raw = re.sub(r'[^0-9A-Fa-f]', '', mac)
    # Reject malformed, all-zero and broadcast placeholders. Locally administered
    # addresses remain valid because they are common for virtual interfaces.
    if len(raw) != 12 or set(raw.casefold()) == {'0'} or set(raw.casefold()) == {'f'}:
        return None
    return ':'.join(raw[index:index + 2] for index in range(0, 12, 2)).upper()


def _preferred_network_entries(value: Any) -> list[dict[str, Any]]:
    """Order usable network records by interface suitability and source order."""
    candidates: list[tuple[int, int, dict[str, Any]]] = []
    for index, entry in enumerate(_network_entries(value)):
        if entry.get('type') == 'loopback' or not _network_entry_active(entry):
            continue
        rank = _network_interface_rank(entry.get('_adapter_name'))
        candidates.append((rank, index, entry))
    candidates.sort(key=lambda item: (item[0], item[1]))
    return [entry for _, _, entry in candidates]


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':
        entries = _preferred_network_entries(source)
        # Prefer a usable IPv4 address on well-known management interfaces.
        for entry in entries:
            if entry.get('family') == 'IPv4' and _valid_ipv4(entry.get('address')):
                return str(entry.get('address')).strip()
        return None
    if transform == 'active_ipv4_mac':
        entries = _preferred_network_entries(source)
        # Prefer the MAC from an interface that also has usable IPv4 data. This
        # keeps IP and MAC aligned in the common case and avoids all-zero values.
        for entry in entries:
            if entry.get('family') == 'IPv4' and _valid_ipv4(entry.get('address')):
                mac = _valid_network_mac(entry.get('mac'))
                if mac:
                    return mac
        # Some MeshCentral inventories expose MAC data in a separate record.
        for entry in entries:
            mac = _valid_network_mac(entry.get('mac'))
            if mac:
                return 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:
    # Most asset columns are text fields. Integer and foreign-key fields
    # are written with the appropriate type so dynamic mappings remain safe there as well.
    if target in {"mesh_mtype", "mesh_icon", "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]:
    """Build asset values exclusively from enabled database 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, '', [], {}):
            # Lower priorities are processed first; later rules may override earlier values.
            mapped[target] = value
    return mapped

def _normalize_manufacturer(value: Any) -> str:
    """Normalize manufacturer names for safe serial-number matching."""
    if value in (None, ""):
        return ""
    text = unicodedata.normalize("NFKD", str(value)).casefold()
    text = "".join(ch for ch in text if not unicodedata.combining(ch))
    tokens = re.findall(r"[a-z0-9]+", text)
    suffixes = {
        "inc", "incorporated", "corp", "corporation", "company", "co",
        "ltd", "limited", "llc", "gmbh", "ag", "kg", "se", "plc",
    }
    while tokens and tokens[-1] in suffixes:
        tokens.pop()
    return "".join(tokens)


def _normalized_serial_value(value: Any) -> str:
    if value in (None, ""):
        return ""
    return str(value).replace("\x00", "").strip().casefold()


def _invalid_serial_reason(mapped: dict[str, Any], cfg: dict[str, Any]) -> str | None:
    """Return a human-readable reason when a reported serial number is a configured placeholder."""
    serial = _normalized_serial_value(mapped.get("serial_number"))
    if not serial:
        return None

    global_values = {
        _normalized_serial_value(value)
        for value in (cfg.get("invalid_serial_values") or [])
        if _normalized_serial_value(value)
    }
    if serial in global_values:
        return "global configured placeholder"

    manufacturer = _normalize_manufacturer(mapped.get("manufacturer"))
    if manufacturer:
        for rule in cfg.get("manufacturer_invalid_serial_rules") or []:
            if not isinstance(rule, dict):
                continue
            if _normalize_manufacturer(rule.get("manufacturer")) != manufacturer:
                continue
            values = {
                _normalized_serial_value(value)
                for value in (rule.get("values") or [])
                if _normalized_serial_value(value)
            }
            if serial in values:
                return f"configured placeholder for {rule.get('manufacturer') or mapped.get('manufacturer')}"
    return None


def _serial_identity_key(mapped: dict[str, Any], *, include_manufacturer: bool) -> tuple[str, str] | None:
    serial = _normalized_serial_value(mapped.get("serial_number"))
    if not serial:
        return None
    manufacturer = _normalize_manufacturer(mapped.get("manufacturer")) if include_manufacturer else ""
    return serial, manufacturer


def _normalized_mac_identity(value: Any) -> str:
    mac = _valid_network_mac(value)
    return mac.casefold() if mac else ""


def _find_asset(
    db: Session,
    mapped: dict[str, Any],
    order: list[str],
    *,
    serial_match_manufacturer: bool = False,
    serial_is_invalid: bool = False,
    serial_is_ambiguous: bool = False,
) -> tuple[Asset | None, str | None, str | None]:
    incoming_mac = _normalized_mac_identity(mapped.get("mac_address"))
    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 == "mac_address" and incoming_mac:
            matches = db.query(Asset).filter(func.lower(Asset.mac_address) == str(mapped.get("mac_address")).lower()).all()
        elif method == "serial_number" and mapped.get("serial_number") and not serial_is_invalid and not serial_is_ambiguous:
            matches = db.query(Asset).filter(func.lower(Asset.serial_number) == mapped["serial_number"].lower()).all()
            if serial_match_manufacturer:
                incoming_manufacturer = _normalize_manufacturer(mapped.get("manufacturer"))
                if not incoming_manufacturer:
                    return None, method, "Seriennummer stimmt möglicherweise überein, aber der Hersteller fehlt in den MeshCentral-Daten."
                matches = [
                    asset for asset in matches
                    if _normalize_manufacturer(asset.manufacturer) == incoming_manufacturer
                ]
            # A shared DMI/mainboard serial is common for containers and VMs.
            # Never accept such a serial match when both sides have a valid but
            # different MAC address. This avoids reassigning another guest's
            # MeshCentral node ID.
            if incoming_mac:
                matches = [
                    asset for asset in matches
                    if not _normalized_mac_identity(asset.mac_address)
                    or _normalized_mac_identity(asset.mac_address) == incoming_mac
                ]
        else:
            continue
        if len(matches) == 1:
            return matches[0], method, None
        if len(matches) > 1:
            detail = "Seriennummer und Hersteller" if method == "serial_number" and serial_match_manufacturer else method
            return None, method, f"Mehrere Assets stimmen über {detail} ü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, *, dry_run: bool = False) -> SyncResult:
    log = log or (lambda level, message: None)
    cfg = load_config()["meshcentral"]
    run = SyncRun(status="running", devices_found=0, created_count=0, updated_count=0, conflict_count=0, error_count=0)
    if not dry_run:
        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", "Dry-Run gestartet; es werden keine Datenbankänderungen gespeichert." if dry_run else "Synchronisierung gestartet.")
        log("info", "Asset-Zuordnung: " + " → ".join(cfg.get("match_order", ["node_id", "mac_address", "serial_number"])))
        if cfg.get("serial_match_manufacturer", False):
            log("info", "Seriennummern-Abgleich erfordert zusätzlich einen normalisierten Herstellervergleich.")
        invalid_global_count = len(cfg.get("invalid_serial_values") or [])
        invalid_manufacturer_count = len(cfg.get("manufacturer_invalid_serial_rules") or [])
        if invalid_global_count or invalid_manufacturer_count:
            log("info", f"Dummy-Seriennummernfilter aktiv: {invalid_global_count} globale Werte, {invalid_manufacturer_count} Herstellerregel(n). Node-ID bleibt die stabile Geräteidentität.")
        devices = fetch_devices(log)
        run.devices_found = len(devices)
        log("info", f"Verarbeite {len(devices)} Gerät(e).")

        # Determine serial numbers that are already ambiguous within the current
        # MeshCentral inventory. This is essential for containers/VMs where the
        # host DMI/mainboard serial may be exposed identically to several guests.
        serial_identity_counts: dict[tuple[str, str], int] = {}
        for raw_for_identity in devices:
            mapped_for_identity = map_device(raw_for_identity, field_mappings)
            if _invalid_serial_reason(mapped_for_identity, cfg):
                continue
            identity_key = _serial_identity_key(
                mapped_for_identity,
                include_manufacturer=bool(cfg.get("serial_match_manufacturer", False)),
            )
            if identity_key:
                serial_identity_counts[identity_key] = serial_identity_counts.get(identity_key, 0) + 1
        ambiguous_serial_keys = {key for key, count in serial_identity_counts.items() if count > 1}
        if ambiguous_serial_keys:
            log("warning", f"{len(ambiguous_serial_keys)} Seriennummern-Identität(en) sind im aktuellen MeshCentral-Bestand mehrfach vorhanden und werden nicht zum Matching verwendet.")
        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)
                invalid_serial_reason = _invalid_serial_reason(mapped, cfg)
                serial_identity_key = _serial_identity_key(
                    mapped,
                    include_manufacturer=bool(cfg.get("serial_match_manufacturer", False)),
                )
                serial_is_ambiguous = bool(serial_identity_key and serial_identity_key in ambiguous_serial_keys)
                if serial_is_ambiguous:
                    log("warning", f"{mapped.get('name', node_id)}: Seriennummer '{mapped.get('serial_number')}' ist im aktuellen MeshCentral-Bestand nicht eindeutig und wird nicht zum Matching verwendet. Node-ID/MAC haben Vorrang.")
                if invalid_serial_reason:
                    reported_serial = mapped.get("serial_number")
                    log("warning", f"{mapped.get('name', node_id)}: gemeldete Seriennummer '{reported_serial}' als Dummy erkannt ({invalid_serial_reason}); Seriennummer wird weder zum Matching noch zum Überschreiben verwendet. Node-ID: {node_id}.")
                asset, match_method, match_error = _find_asset(
                    db,
                    mapped,
                    cfg.get("match_order", ["node_id", "mac_address", "serial_number"]),
                    serial_match_manufacturer=bool(cfg.get("serial_match_manufacturer", False)),
                    serial_is_invalid=bool(invalid_serial_reason),
                    serial_is_ambiguous=serial_is_ambiguous,
                )
                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_icon = str(mapped.get("mesh_icon") or 0)
                    mapped_category_id = category_mapping.get(device_icon)
                    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-Icon {device_icon}; 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)}: {'würde als neues Asset' if dry_run else 'neues Asset'} in Kategorie '{category.name}' (ID {category.id}) {'angelegt werden' if dry_run else 'angelegt'}." )
                elif match_method == "serial_number" and not asset.mesh_node_id:
                    asset.mesh_node_id = node_id
                    match_note = "über Seriennummer + Hersteller" if cfg.get("serial_match_manufacturer", False) else "über Seriennummer"
                    log("info", f"{asset.name}: {match_note} zugeordnet{' (Dry-Run)' if dry_run else ''}.")

                conflicts: dict[str, Any] = {}
                history_changes: dict[str, dict[str, str]] = {}
                changed = False

                # Only enabled mapping rules determine which
                # system and custom fields are synchronized. There is no
                # feste Feld-Whitelist mehr.
                for target, incoming in mapped.items():
                    if target == "mesh_device_type" or incoming in (None, "", [], {}):
                        continue
                    if target == "serial_number" and invalid_serial_reason:
                        continue

                    definition = field_definitions.get(target)
                    if definition is None:
                        # The target field must exist in the central field catalog.
                        # This validation is also logged before the device loop starts.
                        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 {'würden aktualisiert' if dry_run else '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()
        if dry_run:
            db.rollback()
            log("success", f"Dry-Run beendet: {run.devices_found} gefunden, {run.created_count} würden neu angelegt, {run.updated_count} würden geändert, {run.conflict_count} Konflikte, {run.error_count} Fehler. Es wurden keine Änderungen gespeichert.")
        else:
            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()
        if not dry_run and run.id is not None:
            stored_run = db.query(SyncRun).filter(SyncRun.id == run.id).first()
            if stored_run is not None:
                run = stored_run
        run.status = "error"
        run.error_count = int(run.error_count or 0) + 1
        run.message = str(error)
        run.finished_at = datetime.utcnow()
        if not dry_run and run.id is not None:
            db.commit()
        log("error", f"{'Dry-Run' if dry_run else 'Synchronisierung'} abgebrochen: {error}")
    return SyncResult(run=run)
from .presence_state import extract_last_seen

