netbox-zabbix-sync/netbox_zabbix_sync/modules/cli.py

import argparse
import logging
from os import environ

from netbox_zabbix_sync.modules.core import Sync
from netbox_zabbix_sync.modules.exceptions import EnvironmentVarError
from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels, setup_logger
from netbox_zabbix_sync.modules.settings import load_config

# Boolean settings that can be toggled via --flag / --no-flag
_BOOL_ARGS = [
    ("clustering", "Enable clustering of devices with virtual chassis setup."),
    ("create_hostgroups", "Enable hostgroup generation (requires Zabbix permissions)."),
    ("create_journal", "Create NetBox journal entries on changes."),
    ("sync_vms", "Enable virtual machine sync."),
    (
        "full_proxy_sync",
        "Enable full proxy sync (removes proxies not in config context).",
    ),
    (
        "templates_config_context",
        "Use config context as the template source instead of a custom field.",
    ),
    (
        "templates_config_context_overrule",
        "Give config context templates higher priority than custom field templates.",
    ),
    ("traverse_regions", "Use the full parent-region path in hostgroup names."),
    ("traverse_site_groups", "Use the full parent-site-group path in hostgroup names."),
    (
        "extended_site_properties",
        "Fetch additional site info from NetBox (increases API queries).",
    ),
    ("inventory_sync", "Sync NetBox device properties to Zabbix inventory."),
    ("usermacro_sync", "Sync usermacros from NetBox to Zabbix."),
    ("tag_sync", "Sync host tags to Zabbix."),
    ("tag_lower", "Lowercase tag names and values before syncing."),
]

# String settings that can be set via --option VALUE
_STR_ARGS = [
    ("template_cf", "NetBox custom field name for the Zabbix template.", "FIELD"),
    ("device_cf", "NetBox custom field name for the Zabbix host ID.", "FIELD"),
    (
        "hostgroup_format",
        "Hostgroup path pattern for physical devices (e.g. site/manufacturer/role).",
        "PATTERN",
    ),
    (
        "vm_hostgroup_format",
        "Hostgroup path pattern for virtual machines (e.g. cluster_type/cluster/role).",
        "PATTERN",
    ),
    (
        "inventory_mode",
        "Zabbix inventory mode: disabled, manual, or automatic.",
        "MODE",
    ),
    ("tag_name", "Zabbix tag name used when syncing NetBox tags.", "NAME"),
    (
        "tag_value",
        "NetBox tag property to use as the Zabbix tag value (name, slug, or display).",
        "PROPERTY",
    ),
]


def _apply_cli_overrides(config: dict, arguments: argparse.Namespace) -> dict:
    """Override loaded config with any values explicitly provided on the CLI."""
    for key, _help in _BOOL_ARGS:
        cli_val = getattr(arguments, key, None)
        if cli_val is not None:
            config[key] = cli_val
    for key, _help, _meta in _STR_ARGS:
        cli_val = getattr(arguments, key, None)
        if cli_val is not None:
            config[key] = cli_val
    return config


def main(arguments):
    """Run the sync process."""
    # Set logging
    setup_logger()
    logger = get_logger()
    # Set log levels based on verbosity flags
    if arguments.verbose:
        set_log_levels(logging.WARNING, logging.INFO)
    if arguments.debug:
        set_log_levels(logging.WARNING, logging.DEBUG)
    if arguments.debug_all:
        set_log_levels(logging.DEBUG, logging.DEBUG)
    if arguments.quiet:
        set_log_levels(logging.ERROR, logging.ERROR)

    # Gather environment variables for Zabbix and Netbox communication
    env_vars = ["ZABBIX_HOST", "NETBOX_HOST", "NETBOX_TOKEN"]
    if "ZABBIX_TOKEN" in environ:
        env_vars.append("ZABBIX_TOKEN")
    else:
        env_vars.append("ZABBIX_USER")
        env_vars.append("ZABBIX_PASS")
    for var in env_vars:
        if var not in environ:
            e = f"Environment variable {var} has not been defined."
            logger.error(e)
            raise EnvironmentVarError(e)
    # Get all virtual environment variables
    if "ZABBIX_TOKEN" in env_vars:
        zabbix_user = None
        zabbix_pass = None
        zabbix_token = environ.get("ZABBIX_TOKEN")
    else:
        zabbix_user = environ.get("ZABBIX_USER")
        zabbix_pass = environ.get("ZABBIX_PASS")
        zabbix_token = None
    zabbix_host = environ.get("ZABBIX_HOST")
    netbox_host = environ.get("NETBOX_HOST")
    netbox_token = environ.get("NETBOX_TOKEN")

    # Load config (defaults → config.py → env vars), then apply CLI overrides
    config = load_config(config_file=arguments.config)
    config = _apply_cli_overrides(config, arguments)

    # Run main sync process
    syncer = Sync(config=config)
    syncer.connect(
        nb_host=netbox_host,
        nb_token=netbox_token,
        zbx_host=zabbix_host,
        zbx_user=zabbix_user,
        zbx_pass=zabbix_pass,
        zbx_token=zabbix_token,
    )
    syncer.start()


def parse_cli():
    """
    Parse command-line arguments and run the main function.
    """
    parser = argparse.ArgumentParser(
        description="Synchronise NetBox device data to Zabbix."
    )

    # ── Verbosity ──────────────────────────────────────────────────────────────
    parser.add_argument(
        "-v", "--verbose", help="Turn on verbose logging.", action="store_true"
    )
    parser.add_argument(
        "-vv", "--debug", help="Turn on debugging.", action="store_true"
    )
    parser.add_argument(
        "-vvv",
        "--debug-all",
        help="Turn on debugging for all modules.",
        action="store_true",
    )
    parser.add_argument("-q", "--quiet", help="Turn off warnings.", action="store_true")
    parser.add_argument(
        "-c",
        "--config",
        help="Path to the config file (default: config.py next to the script or in the current directory).",
        metavar="FILE",
        default=None,
    )

    # ── Boolean config overrides ───────────────────────────────────────────────
    bool_group = parser.add_argument_group(
        "config overrides (boolean)",
        "Override boolean settings from config.py. "
        "Use --flag to enable or --no-flag to disable. "
        "When omitted, the value from config.py (or the built-in default) is used.",
    )
    for key, help_text in _BOOL_ARGS:
        flag = key.replace("_", "-")
        bool_group.add_argument(
            f"--{flag}",
            dest=key,
            help=help_text,
            action=argparse.BooleanOptionalAction,
            default=None,
        )

    # ── String config overrides ────────────────────────────────────────────────
    str_group = parser.add_argument_group(
        "config overrides (string)",
        "Override string settings from config.py. "
        "When omitted, the value from config.py (or the built-in default) is used.",
    )
    for key, help_text, metavar in _STR_ARGS:
        flag = key.replace("_", "-")
        str_group.add_argument(
            f"--{flag}",
            dest=key,
            help=help_text,
            metavar=metavar,
            default=None,
        )

    args = parser.parse_args()
    main(args)