#!/usr/bin/env python3 """ NetBox Device Type Library Import Tool Compatible with NetBox 4.4+ Part 1 of 2 """ import sys import argparse from pathlib import Path from typing import List, Optional from config.settings import settings from core.netbox_client import NetBoxClient from core.exceptions import ( NetBoxImportException, NetBoxConnectionError, RepositoryError ) from utils.git_handler import GitHandler from utils.logger import logger from importers.device_type import DeviceTypeImporter def parse_arguments(): """Parse command line arguments""" parser = argparse.ArgumentParser( description='Import NetBox Device Types from Community Library', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Import all vendors python main.py # Import specific vendors python main.py --vendors cisco juniper arista # Import specific device types python main.py --vendors cisco --slugs catalyst-9300 catalyst-9500 # Dry run (no changes) python main.py --vendors mikrotik --dry-run # Verbose output python main.py --vendors cisco --verbose """ ) parser.add_argument( '--vendors', nargs='+', help='List of vendors to import (case-insensitive, comma or space separated)' ) parser.add_argument( '--slugs', nargs='+', help='List of device type slugs to import' ) parser.add_argument( '--url', '--git', help='Git repository URL (default: from .env or official repo)' ) parser.add_argument( '--branch', help='Git branch to use (default: from .env or master)' ) parser.add_argument( '--dry-run', action='store_true', help='Simulate import without making changes' ) parser.add_argument( '--verbose', action='store_true', help='Enable verbose output' ) parser.add_argument( '--no-update', action='store_true', help='Skip repository update if already cloned' ) return parser.parse_args() def setup_logging(verbose: bool): """Configure logging based on verbosity""" log_level = 'DEBUG' if verbose else settings.LOG_LEVEL logger.configure(log_level, verbose) def prepare_vendors_list(args_vendors: Optional[List[str]]) -> List[str]: """ Prepare list of vendors to import Args: args_vendors: Vendors from command line arguments Returns: List of vendor names (lowercase) """ # Use command line vendors if provided, otherwise use settings vendors = args_vendors if args_vendors else settings.VENDORS # Normalize vendor names (lowercase, handle comma-separated) normalized = [] for vendor in vendors: # Split by comma in case user provided comma-separated list parts = [v.strip().lower() for v in vendor.split(',') if v.strip()] normalized.extend(parts) return normalized def prepare_slugs_list(args_slugs: Optional[List[str]]) -> List[str]: """ Prepare list of device slugs to filter Args: args_slugs: Slugs from command line arguments Returns: List of slugs (lowercase) """ # Use command line slugs if provided, otherwise use settings slugs = args_slugs if args_slugs else settings.SLUGS # Normalize slugs (lowercase) normalized = [] for slug in slugs: # Split by comma in case user provided comma-separated list parts = [s.strip().lower() for s in slug.split(',') if s.strip()] normalized.extend(parts) return normalized def main(): """Main entry point""" try: # Parse arguments args = parse_arguments() # Setup logging setup_logging(args.verbose or settings.VERBOSE) # Show configuration logger.info("=" * 60) logger.info("NetBox Device Type Library Import") logger.info("=" * 60) logger.info(f"NetBox URL: {settings.NETBOX_URL}") # Check for dry run mode dry_run = args.dry_run or settings.DRY_RUN if dry_run: logger.warning("DRY RUN MODE - No changes will be made") # Validate settings settings.validate() # Connect to NetBox logger.info("Connecting to NetBox...") netbox_client = NetBoxClient( **settings.get_pynetbox_kwargs() ) netbox_client.test_connection() # Setup Git handler repo_url = args.url if args.url else settings.REPO_URL repo_branch = args.branch if args.branch else settings.REPO_BRANCH git_handler = GitHandler( repo_url=repo_url, repo_path=settings.REPO_PATH, branch=repo_branch ) # Clone or update repository if args.no_update and settings.REPO_PATH.exists(): logger.info("Skipping repository update (--no-update)") else: git_handler.clone_or_update() # Get list of manufacturers from repository all_manufacturers = git_handler.get_manufacturers() logger.info(f"Found {len(all_manufacturers)} manufacturers in repository") # Filter manufacturers if specified filter_vendors = prepare_vendors_list(args.vendors) if filter_vendors: manufacturers_to_import = [ m for m in all_manufacturers if m.lower() in filter_vendors ] logger.info(f"Filtered to {len(manufacturers_to_import)} manufacturers") else: manufacturers_to_import = all_manufacturers logger.info("Importing all manufacturers") if not manufacturers_to_import: logger.error("No manufacturers to import!") return 1 # Filter slugs if specified filter_slugs = prepare_slugs_list(args.slugs) if filter_slugs: logger.info(f"Filtering device types by slugs: {', '.join(filter_slugs)}") # Create device type importer importer = DeviceTypeImporter(netbox_client, dry_run=dry_run) # Process each manufacturer logger.info("=" * 60) logger.info("Starting import process...") logger.info("=" * 60) for manufacturer_name in manufacturers_to_import: try: # Create or get manufacturer in NetBox manufacturer_slug = manufacturer_name.lower().replace(' ', '-') if dry_run: logger.info(f"[DRY RUN] Processing manufacturer: {manufacturer_name}") # Create a mock manufacturer object class MockManufacturer: def __init__(self, name, slug): self.id = 0 self.name = name self.slug = slug manufacturer_obj = MockManufacturer(manufacturer_name, manufacturer_slug) else: manufacturer_obj = netbox_client.get_or_create_manufacturer( name=manufacturer_name, slug=manufacturer_slug ) # Get device type files for this manufacturer device_files = git_handler.get_manufacturer_device_files(manufacturer_name) if not device_files: logger.warning(f"No device files found for {manufacturer_name}") continue # Import device types importer.import_manufacturer_devices( manufacturer_name=manufacturer_name, manufacturer_obj=manufacturer_obj, device_files=device_files, filter_slugs=filter_slugs if filter_slugs else None ) except Exception as e: logger.error(f"Error processing manufacturer {manufacturer_name}: {e}") continue # Print summary importer.print_summary() return 0 except NetBoxConnectionError as e: logger.error(f"NetBox connection error: {e}") logger.error("Please check your NETBOX_URL and NETBOX_TOKEN in .env file") return 1 except RepositoryError as e: logger.error(f"Repository error: {e}") logger.error("Please check your internet connection and REPO_URL") return 1 except NetBoxImportException as e: logger.error(f"Import error: {e}") return 1 except KeyboardInterrupt: logger.warning("Import interrupted by user") return 130 except Exception as e: logger.critical(f"Unexpected error: {e}") logger.debug("Exception details:", exc_info=True) return 1 if __name__ == '__main__': sys.exit(main())