#!/usr/bin/env python3
# Copyright (c) 2026 Christian Wirtz
# Licensed under the GNU General Public License v2 (see the LICENSE file).
"""Checkmk special agent for NetBox (netbox-community/netbox).

Queries the NetBox REST API and emits Checkmk agent sections as JSON. The design
mirrors the checkmk_gitlab special agent: one section per object type, each
carrying a single JSON payload that the agent-based check plugins parse.

Sections produced:
    <<<netbox_instance:sep(0)>>>     /api/status/ plus API response time and update check
    <<<netbox_health:sep(0)>>>       netbox-healthcheck-plugin (DB / cache / Redis)
    <<<netbox_queues:sep(0)>>>       /api/core/background-queues/   (superuser)
    <<<netbox_workers:sep(0)>>>      /api/core/background-workers/  (superuser)
    <<<netbox_jobs:sep(0)>>>         /api/core/jobs/
    <<<netbox_datasources:sep(0)>>>  /api/core/data-sources/
    <<<netbox_changelog:sep(0)>>>    /api/core/object-changes/ (change rate)
    <<<netbox_tokens:sep(0)>>>       /api/users/tokens/ (expiry overview)
    <<<netbox_objects:sep(0)>>>      object counts per selected endpoint
    <<<netbox_inventory:sep(0)>>>    data-hygiene counters (failed devices, ...)
    <<<netbox_prefixes:sep(0)>>>     utilization of selected IPAM prefixes

The agent is intentionally dependency-free (standard library only) so it can be
shipped inside an MKP and executed on any Checkmk site without extra packages.
"""

from __future__ import annotations

import argparse
import ipaddress
import json
import os
import re
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from fnmatch import fnmatch
from typing import Any, Iterator

USER_AGENT = "checkmk-netbox-special-agent/1.0.0"
DEFAULT_TIMEOUT = 30
# Hard cap so a misconfigured/huge query cannot run away.
MAX_PAGES = 50
GITHUB_RELEASES_URL = "https://api.github.com/repos/netbox-community/netbox/releases/latest"
HEALTHCHECK_PATH = "/plugins/netbox_healthcheck_plugin/healthcheck/"
# NetBox v2 API tokens ("nbt_<key>.<secret>", NetBox 4.5+) are documented with the
# "Bearer" keyword; legacy v1 tokens use "Token". See users/constants.TOKEN_PREFIX.
TOKEN_V2_PREFIX = "nbt_"

# Object types that can be counted via "?limit=1" (the paginated response's
# "count" field). Key = the name used in the ruleset and as the service item
# label source; value = (API path, human readable label).
OBJECT_ENDPOINTS: dict[str, tuple[str, str]] = {
    "sites": ("/dcim/sites/", "Sites"),
    "racks": ("/dcim/racks/", "Racks"),
    "devices": ("/dcim/devices/", "Devices"),
    "device_types": ("/dcim/device-types/", "Device types"),
    "modules": ("/dcim/modules/", "Modules"),
    "interfaces": ("/dcim/interfaces/", "Interfaces"),
    "cables": ("/dcim/cables/", "Cables"),
    "inventory_items": ("/dcim/inventory-items/", "Inventory items"),
    "power_feeds": ("/dcim/power-feeds/", "Power feeds"),
    "prefixes": ("/ipam/prefixes/", "Prefixes"),
    "ip_addresses": ("/ipam/ip-addresses/", "IP addresses"),
    "ip_ranges": ("/ipam/ip-ranges/", "IP ranges"),
    "aggregates": ("/ipam/aggregates/", "Aggregates"),
    "vlans": ("/ipam/vlans/", "VLANs"),
    "vrfs": ("/ipam/vrfs/", "VRFs"),
    "virtual_machines": ("/virtualization/virtual-machines/", "Virtual machines"),
    "clusters": ("/virtualization/clusters/", "Clusters"),
    "circuits": ("/circuits/circuits/", "Circuits"),
    "providers": ("/circuits/providers/", "Providers"),
    "tenants": ("/tenancy/tenants/", "Tenants"),
    "contacts": ("/tenancy/contacts/", "Contacts"),
    "tunnels": ("/vpn/tunnels/", "VPN tunnels"),
    "wireless_lans": ("/wireless/wireless-lans/", "Wireless LANs"),
    "users": ("/users/users/", "Users"),
}

# Data-hygiene counters: key -> (API path, query parameters, label). Each is one
# cheap "?limit=1" request whose "count" is reported.
INVENTORY_COUNTERS: dict[str, tuple[str, dict[str, str], str]] = {
    "devices_failed": ("/dcim/devices/", {"status": "failed"}, "Devices failed"),
    "devices_offline": ("/dcim/devices/", {"status": "offline"}, "Devices offline"),
    "devices_planned": ("/dcim/devices/", {"status": "planned"}, "Devices planned"),
    "devices_staged": ("/dcim/devices/", {"status": "staged"}, "Devices staged"),
    "devices_inventory": ("/dcim/devices/", {"status": "inventory"}, "Devices inventory"),
    "devices_no_primary_ip": ("/dcim/devices/", {"has_primary_ip": "false"},
                              "Devices without primary IP"),
    "vms_failed": ("/virtualization/virtual-machines/", {"status": "failed"}, "VMs failed"),
    "vms_offline": ("/virtualization/virtual-machines/", {"status": "offline"}, "VMs offline"),
    "vms_no_primary_ip": ("/virtualization/virtual-machines/", {"has_primary_ip": "false"},
                          "VMs without primary IP"),
    "ips_deprecated": ("/ipam/ip-addresses/", {"status": "deprecated"}, "IP addresses deprecated"),
    "prefixes_deprecated": ("/ipam/prefixes/", {"status": "deprecated"}, "Prefixes deprecated"),
    "cables_planned": ("/dcim/cables/", {"status": "planned"}, "Cables planned"),
}


def normalize_token(raw: str | None) -> str:
    """Return the bare token value from whatever the user pasted.

    After creating a token, NetBox displays a "token_auth_string" exactly once,
    and that string is the *complete* Authorization header value, assembled from
    Token.get_auth_header_prefix() plus the plaintext:

        Bearer nbt_<key>.<secret>      (v2)
        Token <secret>                 (v1)

    Since that is the only representation the UI ever shows, copying it verbatim
    is the obvious thing to do - so accept it and strip the keyword, instead of
    sending "Authorization: Bearer Bearer nbt_..." and having NetBox reject the
    header.
    """
    token = (raw or "").strip()
    for prefix in ("bearer ", "token "):
        if token.lower().startswith(prefix):
            token = token[len(prefix):].strip()
            break
    return token


def token_problem(token: str) -> str | None:
    """Return a human-readable reason why this token cannot work, or None.

    Catching this here turns NetBox's generic "Invalid authorization header"
    403 into a message that says what to fix.
    """
    if not token:
        return "no API token was given"
    if any(char.isspace() for char in token):
        return (
            "the token contains a space. NetBox shows a new token once, as a complete "
            "header value like 'Bearer nbt_<key>.<secret>' - paste that whole string, "
            "or just the 'nbt_<key>.<secret>' part, but nothing more"
        )
    if token.startswith(TOKEN_V2_PREFIX) and "." not in token:
        return (
            "this looks like a v2 token key without its secret. A v2 token is "
            "'nbt_<key>.<secret>'; the NetBox token list shows only the key, which "
            "cannot authenticate. The full value is shown once, right after creation"
        )
    return None


@dataclass
class Args:
    url: str
    token: str | None
    token_problem: str | None
    timeout: int
    no_cert_check: bool
    max_workers: int
    collect_health: bool
    health_path: str
    collect_queues: bool
    collect_workers: bool
    collect_jobs: bool
    job_window: int
    job_names: list[str]
    collect_datasources: bool
    collect_changelog: bool
    changelog_window: int
    collect_tokens: bool
    collect_objects: list[str]
    collect_inventory: bool
    prefixes: list[str]
    prefix_tags: list[str]
    prefix_roles: list[str]
    check_updates: bool
    update_source: str  # "github" or "manual"
    update_target: str | None
    update_cache_ttl: int
    debug: bool


def parse_arguments(argv: list[str]) -> Args:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", required=True,
                        help="Base URL of the NetBox server, e.g. https://netbox.example.com "
                             "(without a trailing slash and without /api)")
    tok = parser.add_mutually_exclusive_group()
    tok.add_argument("--token", help="NetBox API token (read-only is sufficient)")
    tok.add_argument("--token-file",
                     help="Read the API token from this file (alternative to --token)")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT,
                        help="HTTP timeout in seconds (default: %(default)s)")
    parser.add_argument("--no-cert-check", action="store_true",
                        help="Do not verify the TLS certificate (self-signed setups)")
    parser.add_argument("--max-workers", type=int, default=8,
                        help="Number of parallel API workers (default: %(default)s, "
                             "use 1 to disable parallelism)")

    parser.add_argument("--health", dest="collect_health", action="store_true",
                        help="Query the netbox-healthcheck-plugin endpoint "
                             "(PostgreSQL, Django cache, both Redis instances).")
    parser.add_argument("--health-path", default=HEALTHCHECK_PATH,
                        help="Path of the healthcheck endpoint (default: %(default)s)")

    parser.add_argument("--no-queues", dest="collect_queues", action="store_false", default=True,
                        help="Do not query /api/core/background-queues/ (needs a superuser token).")
    parser.add_argument("--no-workers", dest="collect_workers", action="store_false", default=True,
                        help="Do not query /api/core/background-workers/ (needs a superuser "
                             "token).")

    parser.add_argument("--no-jobs", dest="collect_jobs", action="store_false", default=True,
                        help="Do not collect NetBox background jobs (/api/core/jobs/).")
    parser.add_argument("--job-window", type=int, default=86400,
                        help="Only consider jobs created within this many seconds for the "
                             "aggregate 'NetBox Jobs' service (default: %(default)s = 24h). "
                             "Recurring and named jobs are always reported with their latest "
                             "run, regardless of this window.")
    parser.add_argument("--job-name", dest="job_names", action="append", default=[],
                        help="Create a dedicated service for jobs with this name. Glob patterns "
                             "are allowed (e.g. 'Sync *'). Repeatable. Jobs with a scheduling "
                             "interval get a dedicated service automatically.")

    parser.add_argument("--no-datasources", dest="collect_datasources", action="store_false",
                        default=True, help="Do not collect data sources (/api/core/data-sources/).")

    parser.add_argument("--no-changelog", dest="collect_changelog", action="store_false",
                        default=True, help="Do not collect the change-log rate.")
    parser.add_argument("--changelog-window", type=int, default=86400,
                        help="Time window for the change-log rate in seconds "
                             "(default: %(default)s = 24h)")

    parser.add_argument("--tokens", dest="collect_tokens", action="store_true",
                        help="Collect the API token overview (expiry/usage; needs a token that "
                             "may list /api/users/tokens/, i.e. an administrator).")

    parser.add_argument("--object", dest="collect_objects", action="append", default=[],
                        choices=sorted(OBJECT_ENDPOINTS), metavar="TYPE",
                        help="Report the number of objects of this type as its own service. "
                             "Repeatable. Available: " + ", ".join(sorted(OBJECT_ENDPOINTS)))
    parser.add_argument("--inventory", dest="collect_inventory", action="store_true",
                        help="Collect data-hygiene counters (devices in status failed/offline, "
                             "objects without a primary IP, deprecated IPs/prefixes, ...).")

    parser.add_argument("--prefix", dest="prefixes", action="append", default=[],
                        help="Monitor the utilization of this prefix, e.g. 10.0.0.0/22. "
                             "Repeatable.")
    parser.add_argument("--prefix-tag", dest="prefix_tags", action="append", default=[],
                        help="Monitor the utilization of every prefix carrying this tag (slug). "
                             "Repeatable.")
    parser.add_argument("--prefix-role", dest="prefix_roles", action="append", default=[],
                        help="Monitor the utilization of every prefix with this role (slug). "
                             "Repeatable.")

    parser.add_argument("--check-updates", action="store_true",
                        help="Compare the installed version against the latest NetBox release "
                             "and report whether an update is available.")
    parser.add_argument("--update-source", choices=("github", "manual"), default="github",
                        help="Where to obtain the latest version: 'github' queries the "
                             "netbox-community/netbox releases API (needs internet on the "
                             "Checkmk server); 'manual' uses --update-target (air-gapped).")
    parser.add_argument("--update-target", default=None,
                        help="Desired/latest version for --update-source manual, e.g. 4.4.2")
    parser.add_argument("--update-cache-ttl", type=int, default=28800,
                        help="Cache the looked-up latest version this many seconds "
                             "(default: %(default)s = 8h) to avoid frequent GitHub calls.")

    parser.add_argument("--debug", action="store_true",
                        help="Show tracebacks and raise on API errors.")

    args = parser.parse_args(argv)

    token = args.token
    if args.token_file:
        with open(args.token_file, encoding="utf-8") as handle:
            token = handle.read().strip()
    token = normalize_token(token)

    return Args(
        url=args.url.rstrip("/"),
        token=token,
        token_problem=token_problem(token),
        timeout=args.timeout,
        no_cert_check=args.no_cert_check,
        max_workers=max(1, args.max_workers),
        collect_health=args.collect_health,
        health_path=args.health_path,
        collect_queues=args.collect_queues,
        collect_workers=args.collect_workers,
        collect_jobs=args.collect_jobs,
        job_window=max(0, args.job_window),
        job_names=args.job_names,
        collect_datasources=args.collect_datasources,
        collect_changelog=args.collect_changelog,
        changelog_window=max(60, args.changelog_window),
        collect_tokens=args.collect_tokens,
        collect_objects=args.collect_objects,
        collect_inventory=args.collect_inventory,
        prefixes=args.prefixes,
        prefix_tags=args.prefix_tags,
        prefix_roles=args.prefix_roles,
        check_updates=args.check_updates,
        update_source=args.update_source,
        update_target=args.update_target,
        update_cache_ttl=max(0, args.update_cache_ttl),
        debug=args.debug,
    )


class APIError(RuntimeError):
    """An API request failed. Carries the HTTP status so callers can tell a
    missing permission (403) from a missing endpoint (404) or a real outage."""

    def __init__(self, message: str, status: int | None = None) -> None:
        super().__init__(message)
        self.status = status


class NetBoxAPI:
    def __init__(self, args: Args) -> None:
        self._base = args.url
        self._api = f"{args.url}/api"
        self._timeout = args.timeout
        self._headers = {"User-Agent": USER_AGENT, "Accept": "application/json"}
        if args.token:
            # NetBox 4.5 introduced v2 tokens, which carry an "nbt_" prefix and are
            # documented with the "Bearer" keyword; v1 tokens use "Token". NetBox
            # actually infers the version from the prefix and accepts either keyword,
            # but sending the documented one keeps this working if that ever tightens.
            keyword = "Bearer" if args.token.startswith(TOKEN_V2_PREFIX) else "Token"
            self._headers["Authorization"] = f"{keyword} {args.token}"
        self._ctx = None
        if args.no_cert_check:
            import ssl
            self._ctx = ssl.create_default_context()
            self._ctx.check_hostname = False
            self._ctx.verify_mode = ssl.CERT_NONE

    def _open(self, url: str) -> tuple[int, dict[str, str], bytes]:
        request = urllib.request.Request(url, headers=self._headers)
        try:
            with urllib.request.urlopen(request, timeout=self._timeout, context=self._ctx) as resp:
                return resp.status, dict(resp.headers), resp.read()
        except urllib.error.HTTPError as exc:
            return exc.code, dict(exc.headers or {}), exc.read()
        except (urllib.error.URLError, OSError) as exc:
            # A connection-level failure (host down, DNS, TLS, timeout) must not kill
            # the whole agent run: it is turned into an APIError like any HTTP error,
            # so the affected collector degrades and the instance check can report a
            # proper CRIT with the reason instead of the agent just exiting.
            raise APIError(f"{url.split('/api', 1)[0]} unreachable: {exc}", None) from exc

    def get_json(self, path: str, params: dict[str, Any] | None = None) -> Any:
        query = f"?{urllib.parse.urlencode(params, doseq=True)}" if params else ""
        status, headers, body = self._open(f"{self._api}{path}{query}")
        if status >= 400:
            raise APIError(
                f"NetBox API {path} returned HTTP {status}{_error_reason(headers, body)}",
                status)
        return json.loads(body) if body else None

    def get_count(self, path: str, params: dict[str, Any] | None = None) -> int:
        """Return the number of objects matching the query.

        Uses the paginated response's "count" field with limit=1, so this is a
        single cheap request no matter how many objects exist. "brief" keeps the
        one returned object small.
        """
        query_params = dict(params or {})
        query_params.update({"limit": 1, "brief": "true"})
        data = self.get_json(path, query_params)
        count = (data or {}).get("count")
        if not isinstance(count, int):
            raise APIError(f"NetBox API {path} returned no usable 'count'")
        return count

    def get_paginated(self, path: str, params: dict[str, Any] | None = None,
                      limit: int = 100) -> Iterator[Any]:
        query_params = dict(params or {})
        query_params["limit"] = limit
        query_params["offset"] = 0
        for _page in range(MAX_PAGES):
            data = self.get_json(path, query_params)
            if not isinstance(data, dict):
                break
            results = data.get("results") or []
            yield from results
            if not data.get("next") or not results:
                break
            query_params["offset"] = int(query_params["offset"]) + limit

    def get_raw(self, path: str) -> tuple[int | None, str, str | None]:
        """GET an absolute path below the base URL (not below /api).

        Returns (http_status, body_text, error). Used for the healthcheck plugin,
        which lives under /plugins/... and is not part of the REST API.
        """
        request = urllib.request.Request(f"{self._base}{path}", headers=self._headers)
        try:
            with urllib.request.urlopen(request, timeout=self._timeout, context=self._ctx) as resp:
                return resp.status, resp.read().decode("utf-8", "replace"), None
        except urllib.error.HTTPError as exc:
            return exc.code, (exc.read() or b"").decode("utf-8", "replace"), None
        except (urllib.error.URLError, OSError) as exc:
            return None, "", str(exc)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _parallel_map(func, items: list, max_workers: int) -> list:
    """Run func over items, in parallel when max_workers > 1, preserving order."""
    items = list(items)
    if not items:
        return []
    if max_workers <= 1 or len(items) == 1:
        return [func(it) for it in items]
    with ThreadPoolExecutor(max_workers=min(max_workers, len(items))) as pool:
        return list(pool.map(func, items))


def _parse_ts(value: Any) -> datetime | None:
    if not value or not isinstance(value, str):
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def _error_reason(headers: dict[str, str], body: bytes) -> str:
    """A short, safe tail for an APIError message.

    NetBox's REST framework answers a 4xx with a small JSON body such as
    {"detail": "You do not have permission ..."} - surface that text. But a
    request that misses the API router entirely (a stale endpoint path, an
    auth redirect to the HTML login page) gets NetBox's whole SPA page back;
    that must never be spliced into a service summary, so it is collapsed to a
    single note instead of dumping markup.
    """
    text = (body or b"").decode("utf-8", "replace").strip()
    ctype = next((v.lower() for k, v in (headers or {}).items()
                  if k.lower() == "content-type"), "")
    if text[:1] in "{[" or "json" in ctype:
        try:
            data = json.loads(text)
        except ValueError:
            data = None
        if isinstance(data, dict):
            detail = data.get("detail") or data.get("error") or data.get("message")
            return f": {str(detail)[:200]}" if detail else f": {json.dumps(data)[:200]}"
        if isinstance(data, list) and data:
            return f": {str(data[0])[:200]}"
    if "html" in ctype or text[:1] == "<":
        return " (non-JSON response - this path is probably not a REST API endpoint here)"
    return f": {text[:150]}" if text else ""


def _emit(section: str, payload: Any) -> None:
    sys.stdout.write(f"<<<{section}:sep(0)>>>\n")
    sys.stdout.write(json.dumps(payload, default=str))
    sys.stdout.write("\n")


# ---------------------------------------------------------------------------
# Instance / status
# ---------------------------------------------------------------------------


def collect_instance(api: NetBoxAPI, args: Args) -> dict[str, Any]:
    """GET /api/status/ plus the wall-clock time the request took.

    /api/status/ is the cheapest authenticated endpoint NetBox has, which makes
    its round-trip time a reasonable proxy for "how responsive is the API".
    """
    if args.token_problem:
        # Report the reason instead of firing a request that is guaranteed to 403,
        # so the NetBox Instance service says what to fix.
        return {"url": args.url, "response_time": None,
                "error": f"API token unusable: {args.token_problem}"}

    started = time.monotonic()
    try:
        status = api.get_json("/status/")
        elapsed = time.monotonic() - started
        error = None
    except APIError as exc:
        status, elapsed, error = None, time.monotonic() - started, str(exc)

    instance: dict[str, Any] = {
        "url": args.url,
        "response_time": round(elapsed, 4),
        "error": error,
    }
    if isinstance(status, dict):
        # NetBox renamed "installed-apps" to "installed_apps"; accept both.
        instance["status"] = {
            "netbox_version": status.get("netbox-version"),
            "netbox_full_version": status.get("netbox-full-version"),
            "django_version": status.get("django-version"),
            "python_version": status.get("python-version"),
            "hostname": status.get("hostname"),
            "rq_workers_running": status.get("rq-workers-running"),
            "plugins": status.get("plugins") or {},
            "installed_apps": status.get("installed_apps") or status.get("installed-apps") or {},
        }
    if args.check_updates:
        version = (instance.get("status") or {}).get("netbox_version")
        instance["update"] = build_update_info(args, version)
    return instance


# ---------------------------------------------------------------------------
# Health (netbox-healthcheck-plugin)
# ---------------------------------------------------------------------------


def collect_health(api: NetBoxAPI, args: Args) -> dict[str, Any]:
    """Query the netbox-healthcheck-plugin endpoint and normalize its result.

    The plugin (via django-health-check) returns
    {"<check name>": "working"|"OK"|"<error text>", ...} as JSON when asked with
    format=json - "working" on django-health-check 3.x, "OK" on the 4.x rewrite.
    Anything else is the component's error text; the check plugin flags it.
    """
    separator = "&" if "?" in args.health_path else "?"
    status, body, error = api.get_raw(f"{args.health_path}{separator}format=json")
    result: dict[str, Any] = {
        "http_status": status,
        "error": error,
        "components": {},
        "installed": status is not None and status != 404,
    }
    if not body or status == 404:
        # On 404 the body is DRF's own {"detail": "Not found."}, which must not be
        # mistaken for a health component.
        return result
    try:
        parsed = json.loads(body)
    except ValueError:
        # The endpoint answered with HTML - most likely because format=json was
        # not honoured or a login page was returned instead.
        result["error"] = result["error"] or "endpoint did not return JSON"
        return result
    if isinstance(parsed, dict):
        result["components"] = {str(name): str(value) for name, value in parsed.items()}
    return result


# ---------------------------------------------------------------------------
# Background queues / workers (superuser only)
# ---------------------------------------------------------------------------


def collect_queues(api: NetBoxAPI) -> dict[str, Any]:
    try:
        queues = list(api.get_paginated("/core/background-queues/"))
    except APIError as exc:
        return {"error": str(exc), "status": exc.status, "queues": []}
    return {"error": None, "status": None, "queues": [
        {
            "name": queue.get("name"),
            "jobs": queue.get("jobs"),
            "workers": queue.get("workers"),
            "finished_jobs": queue.get("finished_jobs"),
            "started_jobs": queue.get("started_jobs"),
            "deferred_jobs": queue.get("deferred_jobs"),
            "failed_jobs": queue.get("failed_jobs"),
            "scheduled_jobs": queue.get("scheduled_jobs"),
            "oldest_job_timestamp": queue.get("oldest_job_timestamp"),
            "scheduler_pid": queue.get("scheduler_pid"),
        }
        for queue in queues
    ]}


def collect_workers(api: NetBoxAPI) -> dict[str, Any]:
    try:
        workers = list(api.get_paginated("/core/background-workers/"))
    except APIError as exc:
        return {"error": str(exc), "status": exc.status, "workers": []}
    def _state(raw: Any) -> str:
        # RQ reports "?" when a worker's state is not set. That cannot be used as a
        # key in the check's status map, because Checkmk form-spec element names must
        # be valid Python identifiers - so normalize it here.
        text = str(raw or "").strip()
        return text if text and text != "?" else "unknown"

    return {"error": None, "status": None, "workers": [
        {
            "name": worker.get("name"),
            "state": _state(worker.get("state")),
            "birth_date": worker.get("birth_date"),
            "queue_names": worker.get("queue_names") or [],
            "pid": worker.get("pid"),
            "successful_job_count": worker.get("successful_job_count"),
            "failed_job_count": worker.get("failed_job_count"),
            "total_working_time": worker.get("total_working_time"),
        }
        for worker in workers
    ]}


# ---------------------------------------------------------------------------
# NetBox jobs (/api/core/jobs/)
# ---------------------------------------------------------------------------

# NetBox job statuses that mean "this run is over and it went wrong".
JOB_BAD_STATUSES = ("errored", "failed")
JOB_ACTIVE_STATUSES = ("pending", "scheduled", "running")


def collect_jobs(api: NetBoxAPI, args: Args) -> dict[str, Any]:
    """Collect background jobs, ordered newest first.

    Two different things are derived from the same data, so it is fetched once:
      * an aggregate view over the configured time window (how many jobs errored,
        how many are running, is anything stuck), and
      * the latest run per job name, for the jobs that deserve their own service
        (recurring jobs, or names selected via --job-name).
    """
    try:
        # Newest first: the aggregate only cares about the recent window, and the
        # per-name view only about the most recent run of each name.
        raw = list(api.get_paginated("/core/jobs/", {"ordering": "-created"}))
    except APIError as exc:
        return {"error": str(exc), "status": exc.status, "jobs": [], "items": {},
                "window": args.job_window}

    now = datetime.now(timezone.utc)
    cutoff = now - timedelta(seconds=args.job_window)

    def _record(job: dict[str, Any]) -> dict[str, Any]:
        return {
            "id": job.get("id"),
            "name": job.get("name"),
            "status": (job.get("status") or {}).get("value") if isinstance(
                job.get("status"), dict) else job.get("status"),
            "created": job.get("created"),
            "scheduled": job.get("scheduled"),
            "started": job.get("started"),
            "completed": job.get("completed"),
            "interval": job.get("interval"),
            "error": job.get("error"),
            "queue_name": job.get("queue_name"),
            "object_type": job.get("object_type"),
            "user": (job.get("user") or {}).get("username") if isinstance(
                job.get("user"), dict) else None,
        }

    records = [_record(job) for job in raw]

    in_window: list[dict[str, Any]] = []
    for record in records:
        created = _parse_ts(record.get("created"))
        if created is None or created >= cutoff:
            in_window.append(record)

    # Latest run per name (records are newest first, so the first wins).
    latest_by_name: dict[str, dict[str, Any]] = {}
    for record in records:
        name = record.get("name")
        if name and name not in latest_by_name:
            latest_by_name[name] = record

    items: dict[str, dict[str, Any]] = {}
    for name, record in latest_by_name.items():
        # A scheduling interval means the job is expected to run again, which is
        # exactly the kind of job worth its own service. Explicitly named jobs
        # are included regardless.
        recurring = bool(record.get("interval"))
        selected = any(fnmatch(name, pattern) for pattern in args.job_names)
        if recurring or selected:
            items[name] = {**record, "recurring": recurring}

    # Duration of each finished job, so the check can apply levels without
    # re-parsing timestamps for every service.
    for record in in_window + list(items.values()):
        started = _parse_ts(record.get("started"))
        completed = _parse_ts(record.get("completed"))
        if started and completed:
            record["duration"] = (completed - started).total_seconds()
        elif started:
            record["running_for"] = (now - started).total_seconds()

    return {
        "error": None,
        "status": None,
        "window": args.job_window,
        "jobs": in_window,
        "items": items,
        "truncated": len(raw) >= MAX_PAGES * 100,
    }


# ---------------------------------------------------------------------------
# Data sources (/api/core/data-sources/)
# ---------------------------------------------------------------------------


def collect_datasources(api: NetBoxAPI) -> dict[str, Any]:
    try:
        raw = list(api.get_paginated("/core/data-sources/"))
    except APIError as exc:
        return {"error": str(exc), "status": exc.status, "sources": []}

    def _value(field: Any) -> Any:
        return field.get("value") if isinstance(field, dict) else field

    return {"error": None, "status": None, "sources": [
        {
            "id": source.get("id"),
            "name": source.get("name"),
            "type": _value(source.get("type")),
            "source_url": source.get("source_url"),
            "enabled": bool(source.get("enabled")),
            "status": _value(source.get("status")),
            "last_synced": source.get("last_synced"),
            "sync_interval": source.get("sync_interval"),
            "file_count": source.get("file_count"),
            "description": source.get("description") or "",
        }
        for source in raw
    ]}


# ---------------------------------------------------------------------------
# Change log (/api/core/object-changes/)
# ---------------------------------------------------------------------------


def collect_changelog(api: NetBoxAPI, args: Args) -> dict[str, Any]:
    """Number of change records in the window, plus the most recent change.

    The endpoint moved from /api/extras/object-changes/ to
    /api/core/object-changes/ in NetBox 4.1, so both are tried.
    """
    cutoff = datetime.now(timezone.utc) - timedelta(seconds=args.changelog_window)
    # NetBox's filter is inclusive on the "after" bound and accepts ISO 8601.
    filters = {"time_after": cutoff.isoformat()}

    last_error: str | None = None
    last_status: int | None = None
    for path in ("/core/object-changes/", "/extras/object-changes/"):
        try:
            count = api.get_count(path, filters)
        except APIError as exc:
            last_error, last_status = str(exc), exc.status
            # /extras/object-changes/ is only the pre-4.1 location. A 403 (or any
            # non-404) on /core/object-changes/ is already the real answer - don't
            # chase the legacy path and end up reporting its "not an API route"
            # HTML 404 in place of the permission error.
            if exc.status != 404:
                break
            continue
        last_change_at = None
        total = None
        try:
            newest = api.get_json(path, {"limit": 1, "ordering": "-time"})
            if isinstance(newest, dict):
                total = newest.get("count")
                results = newest.get("results") or []
                if results:
                    last_change_at = results[0].get("time")
        except APIError:
            pass
        return {"error": None, "status": None, "endpoint": path,
                "window": args.changelog_window, "count": count,
                "total": total, "last_change_at": last_change_at}

    return {"error": last_error, "status": last_status, "endpoint": None,
            "window": args.changelog_window, "count": None,
            "total": None, "last_change_at": None}


# ---------------------------------------------------------------------------
# API tokens (/api/users/tokens/)
# ---------------------------------------------------------------------------


def collect_tokens(api: NetBoxAPI) -> dict[str, Any]:
    """Overview of the instance's API tokens: expiry and last use.

    Only metadata is collected. NetBox never returns the full key of another
    user's token, and this agent does not store any key material either.
    """
    try:
        raw = list(api.get_paginated("/users/tokens/"))
    except APIError as exc:
        return {"error": str(exc), "status": exc.status, "tokens": []}

    now = datetime.now(timezone.utc)
    tokens: list[dict[str, Any]] = []
    for token in raw:
        expires = _parse_ts(token.get("expires"))
        last_used = _parse_ts(token.get("last_used"))
        user = token.get("user")
        tokens.append({
            "id": token.get("id"),
            "user": user.get("username") if isinstance(user, dict) else None,
            "description": token.get("description") or "",
            "enabled": token.get("enabled", True),
            "write_enabled": token.get("write_enabled"),
            "expires": token.get("expires"),
            "expires_in": (expires - now).total_seconds() if expires else None,
            "last_used": token.get("last_used"),
            "unused_for": (now - last_used).total_seconds() if last_used else None,
            "never_used": last_used is None,
        })
    return {"error": None, "status": None, "tokens": tokens}


# ---------------------------------------------------------------------------
# Object counts and hygiene counters
# ---------------------------------------------------------------------------


def collect_objects(api: NetBoxAPI, args: Args) -> dict[str, Any]:
    def _one(name: str) -> tuple[str, dict[str, Any]]:
        path, label = OBJECT_ENDPOINTS[name]
        try:
            return name, {"label": label, "count": api.get_count(path), "error": None}
        except APIError as exc:
            return name, {"label": label, "count": None, "error": str(exc)}

    selected = [name for name in args.collect_objects if name in OBJECT_ENDPOINTS]
    return dict(_parallel_map(_one, selected, args.max_workers))


def collect_inventory(api: NetBoxAPI, args: Args) -> dict[str, Any]:
    def _one(name: str) -> tuple[str, dict[str, Any]]:
        path, params, label = INVENTORY_COUNTERS[name]
        try:
            return name, {"label": label, "count": api.get_count(path, params), "error": None}
        except APIError as exc:
            return name, {"label": label, "count": None, "error": str(exc)}

    counters = dict(_parallel_map(_one, list(INVENTORY_COUNTERS), args.max_workers))
    return {"counters": counters}


# ---------------------------------------------------------------------------
# Prefix utilization
# ---------------------------------------------------------------------------


def _prefix_size(network: ipaddress.IPv4Network | ipaddress.IPv6Network,
                 is_pool: bool) -> int:
    """The denominator NetBox measures address utilization against.

    Mirrors Prefix._get_utilization_denominator(): IPv4 prefixes shorter than /31
    that are not marked as a pool exclude the network and broadcast address.
    """
    size = network.num_addresses
    if network.version == 4 and network.prefixlen < 31 and not is_pool:
        return size - 2
    return size


def _resolve_prefixes(api: NetBoxAPI, args: Args) -> list[dict[str, Any]]:
    """Look up the prefix objects selected by --prefix / --prefix-tag / --prefix-role."""
    found: dict[int, dict[str, Any]] = {}

    def _add(entries: list[dict[str, Any]]) -> None:
        for entry in entries:
            if entry.get("id") is not None:
                found[entry["id"]] = entry

    for prefix in args.prefixes:
        try:
            _add(list(api.get_paginated("/ipam/prefixes/", {"prefix": prefix})))
        except APIError:
            continue
    for tag in args.prefix_tags:
        try:
            _add(list(api.get_paginated("/ipam/prefixes/", {"tag": tag})))
        except APIError:
            continue
    for role in args.prefix_roles:
        try:
            _add(list(api.get_paginated("/ipam/prefixes/", {"role": role})))
        except APIError:
            continue
    return list(found.values())


def _container_utilization(api: NetBoxAPI, network, vrf_id: int | None) -> float | None:
    """Utilization of a container prefix: covered address space of its children.

    NetBox sums the child prefixes as a set (so overlapping children are not
    counted twice); ipaddress.collapse_addresses() gives the same result without
    needing netaddr.
    """
    params: dict[str, Any] = {"within": str(network)}
    params["vrf_id"] = vrf_id if vrf_id is not None else "null"
    try:
        children = list(api.get_paginated("/ipam/prefixes/", params))
    except APIError:
        return None
    networks = []
    for child in children:
        try:
            networks.append(ipaddress.ip_network(child.get("prefix"), strict=False))
        except ValueError:
            continue
    if not networks:
        return 0.0
    covered = sum(net.num_addresses
                  for net in ipaddress.collapse_addresses(
                      [n for n in networks if n.version == network.version]))
    return min(covered / network.num_addresses * 100.0, 100.0)


def _address_utilization(api: NetBoxAPI, prefix: dict[str, Any], network,
                         vrf_id: int | None) -> tuple[float | None, int | None, int]:
    """Utilization of a non-container prefix: child IPs plus utilized IP ranges.

    Returns (percentage, child IP count, denominator).
    """
    denominator = _prefix_size(network, bool(prefix.get("is_pool")))
    if denominator <= 0:
        return None, None, denominator

    params: dict[str, Any] = {"parent": str(network)}
    if vrf_id is not None:
        params["vrf_id"] = vrf_id
    try:
        ip_count = api.get_count("/ipam/ip-addresses/", params)
    except APIError:
        return None, None, denominator

    # IP ranges flagged "mark utilized" count as fully used in NetBox's own
    # calculation, so include their size to match what the web UI shows.
    utilized_range_size = 0
    try:
        ranges = list(api.get_paginated(
            "/ipam/ip-ranges/", {"parent": str(network), "mark_utilized": "true"}))
        for entry in ranges:
            size = entry.get("size")
            if isinstance(size, int):
                utilized_range_size += size
    except APIError:
        pass

    used = min(ip_count + utilized_range_size, denominator)
    return min(used / denominator * 100.0, 100.0), ip_count, denominator


def collect_prefixes(api: NetBoxAPI, args: Args) -> list[dict[str, Any]]:
    """Utilization per selected prefix.

    The REST API does not expose a prefix's utilization, so it is recomputed the
    same way NetBox does internally (see Prefix.get_utilization).
    """
    prefixes = _resolve_prefixes(api, args)

    def _one(prefix: dict[str, Any]) -> dict[str, Any] | None:
        try:
            network = ipaddress.ip_network(prefix.get("prefix"), strict=False)
        except (ValueError, TypeError):
            return None
        status = prefix.get("status")
        status = status.get("value") if isinstance(status, dict) else status
        vrf = prefix.get("vrf")
        vrf_id = vrf.get("id") if isinstance(vrf, dict) else None
        vrf_name = vrf.get("name") if isinstance(vrf, dict) else None

        record: dict[str, Any] = {
            "prefix": str(network),
            "id": prefix.get("id"),
            "status": status,
            "is_pool": bool(prefix.get("is_pool")),
            "mark_utilized": bool(prefix.get("mark_utilized")),
            "vrf": vrf_name,
            "description": prefix.get("description") or "",
            "family": network.version,
            "size": network.num_addresses,
        }
        role = prefix.get("role")
        if isinstance(role, dict):
            record["role"] = role.get("name")

        if prefix.get("mark_utilized"):
            record.update({"utilization": 100.0, "kind": "marked",
                           "ip_count": None, "denominator": None})
        elif status == "container":
            record.update({"utilization": _container_utilization(api, network, vrf_id),
                           "kind": "container", "ip_count": None,
                           "denominator": network.num_addresses})
        else:
            utilization, ip_count, denominator = _address_utilization(
                api, prefix, network, vrf_id)
            record.update({"utilization": utilization, "kind": "addresses",
                           "ip_count": ip_count, "denominator": denominator})
        return record

    return [record for record in _parallel_map(_one, prefixes, args.max_workers)
            if record is not None]


# ---------------------------------------------------------------------------
# Update check
# ---------------------------------------------------------------------------

_SEMVER_RE = re.compile(r"(\d+)\.(\d+)(?:\.(\d+))?")


def _parse_version(value: str | None) -> tuple[int, int, int] | None:
    if not value:
        return None
    match = _SEMVER_RE.search(str(value))
    if not match:
        return None
    return (int(match.group(1)), int(match.group(2)), int(match.group(3) or 0))


def _cache_path() -> str:
    base = os.environ.get("OMD_ROOT")
    base = os.path.join(base, "tmp") if base else tempfile.gettempdir()
    return os.path.join(base, "agent_netbox_latest.json")


def _read_cache(ttl: int) -> dict[str, Any] | None:
    path = _cache_path()
    try:
        if ttl > 0 and (time.time() - os.path.getmtime(path)) <= ttl:
            with open(path, encoding="utf-8") as handle:
                return json.load(handle)
    except (OSError, ValueError):
        return None
    return None


def _write_cache(payload: dict[str, Any]) -> None:
    try:
        with open(_cache_path(), "w", encoding="utf-8") as handle:
            json.dump(payload, handle)
    except OSError:
        pass


def _github_latest(timeout: int) -> str | None:
    """Latest stable NetBox release tag (e.g. "v4.4.2" -> "4.4.2")."""
    request = urllib.request.Request(
        GITHUB_RELEASES_URL,
        headers={"User-Agent": USER_AGENT, "Accept": "application/vnd.github+json"})
    with urllib.request.urlopen(request, timeout=timeout) as resp:
        data = json.loads(resp.read())
    tag = data.get("tag_name") or ""
    parsed = _parse_version(tag)
    return f"{parsed[0]}.{parsed[1]}.{parsed[2]}" if parsed else None


def build_update_info(args: Args, version_str: str | None) -> dict[str, Any]:
    installed = _parse_version(version_str)
    info: dict[str, Any] = {
        "enabled": True,
        "source": args.update_source,
        "installed": version_str,
        "latest": None,
        "update_available": None,
        "error": None,
    }

    if args.update_source == "manual":
        info["latest"] = (args.update_target or "").strip() or None
        if info["latest"] is None:
            info["error"] = "manual update source selected but no target version given"
    else:
        cached = _read_cache(args.update_cache_ttl)
        if cached and cached.get("latest"):
            info["latest"] = cached["latest"]
            info["cached"] = True
        else:
            try:
                latest = _github_latest(args.timeout)
                info["latest"] = latest
                if latest:
                    _write_cache({"latest": latest})
                else:
                    info["error"] = "could not parse a version from the latest GitHub release"
            except Exception as exc:  # noqa: BLE001 - never break the agent on this
                info["error"] = f"GitHub lookup failed: {exc}"

    latest = _parse_version(info["latest"])
    if installed is not None and latest is not None:
        info["update_available"] = latest > installed
    return info


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def agent_netbox_main(args: Args) -> int:
    api = NetBoxAPI(args)

    instance = collect_instance(api, args)
    _emit("netbox_instance", instance)

    # If /api/status/ - the cheapest endpoint there is - could not be read, NetBox is
    # down, unreachable or the token is invalid. Every other collector would just run
    # into the same wall one timeout at a time, so stop here and let the NetBox
    # Instance service report the reason.
    if not instance.get("status"):
        sys.stderr.write(f"agent_netbox: /api/status/ unavailable, skipping all other "
                         f"collectors: {instance.get('error')}\n")
        return 0

    if args.collect_health:
        _emit("netbox_health", collect_health(api, args))
    if args.collect_queues:
        _emit("netbox_queues", collect_queues(api))
    if args.collect_workers:
        _emit("netbox_workers", collect_workers(api))
    if args.collect_jobs:
        _emit("netbox_jobs", collect_jobs(api, args))
    if args.collect_datasources:
        _emit("netbox_datasources", collect_datasources(api))
    if args.collect_changelog:
        _emit("netbox_changelog", collect_changelog(api, args))
    if args.collect_tokens:
        _emit("netbox_tokens", collect_tokens(api))
    if args.collect_objects:
        _emit("netbox_objects", collect_objects(api, args))
    if args.collect_inventory:
        _emit("netbox_inventory", collect_inventory(api, args))
    if args.prefixes or args.prefix_tags or args.prefix_roles:
        _emit("netbox_prefixes", collect_prefixes(api, args))

    return 0


def main(argv: list[str] | None = None) -> int:
    args = parse_arguments(sys.argv[1:] if argv is None else argv)
    try:
        return agent_netbox_main(args)
    except Exception as exc:  # noqa: BLE001 - agents must fail with a clean message
        if args.debug:
            raise
        sys.stderr.write(f"agent_netbox: {exc}\n")
        return 1


if __name__ == "__main__":
    sys.exit(main())
