#!/usr/bin/env python3 """ sync_to_netbox_full.py Full sync - clears all cables in NetBox and re-creates from lldp-mapper DB. Use this after topology changes (moved switches, new connections, removed links). For a safe additive-only sync, use sync_to_netbox.py instead. """ import sqlite3 import requests import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') logger = logging.getLogger(__name__) # --- Configuration --- DB_PATH = "/opt/lldp-mapper/data/network.db" NETBOX_URL = "http://192.168.16.130:8001" NETBOX_TOKEN = "nbt_T6aq9XpwNFQG.HtZziXuSATgabbeagWKk3vEhc2Ask1EMV210PWMM" SITE_NAME = "Field Sites" DEVICE_ROLE = "Access Switch" MANUFACTURER = "FS" DEVICE_TYPE = "FS Switch" # --- Helpers --- NB_HEADERS = { "Authorization": f"Bearer {NETBOX_TOKEN}", "Content-Type": "application/json", "Accept": "application/json", } def nb_get(path, params=None): r = requests.get(f"{NETBOX_URL}/api/{path}", headers=NB_HEADERS, params=params) r.raise_for_status() return r.json() def nb_post(path, data): r = requests.post(f"{NETBOX_URL}/api/{path}", headers=NB_HEADERS, json=data) if r.status_code not in (200, 201): logger.error(f"NetBox POST {path} failed: {r.status_code} {r.text}") return None return r.json() def nb_delete(path): r = requests.delete(f"{NETBOX_URL}/api/{path}", headers=NB_HEADERS) return r.status_code == 204 def nb_get_or_create(path, lookup_params, create_data): results = nb_get(path, params=lookup_params).get("results", []) if results: return results[0] logger.info(f"Creating {path}: {create_data.get('name', create_data)}") return nb_post(path, create_data) # --- Read from SQLite --- def load_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row switches = [dict(r) for r in conn.execute("SELECT * FROM switches ORDER BY hostname").fetchall()] links = [dict(r) for r in conn.execute(""" SELECT l.*, sa.hostname as hn_a, sb.hostname as hn_b FROM links l LEFT JOIN switches sa ON l.chassis_a = sa.chassis_id LEFT JOIN switches sb ON l.chassis_b = sb.chassis_id """).fetchall()] conn.close() return switches, links # --- NetBox Sync --- def ensure_site(): return nb_get_or_create( "dcim/sites/", {"name": SITE_NAME}, {"name": SITE_NAME, "slug": SITE_NAME.lower().replace(" ", "-")} ) def ensure_manufacturer(): return nb_get_or_create( "dcim/manufacturers/", {"name": MANUFACTURER}, {"name": MANUFACTURER, "slug": MANUFACTURER.lower()} ) def ensure_device_type(manufacturer_id): return nb_get_or_create( "dcim/device-types/", {"slug": "fs-switch"}, {"model": DEVICE_TYPE, "slug": "fs-switch", "manufacturer": manufacturer_id} ) def ensure_device_role(): return nb_get_or_create( "dcim/device-roles/", {"name": DEVICE_ROLE}, {"name": DEVICE_ROLE, "slug": "access-switch", "color": "2196f3"} ) def ensure_device(switch, site_id, device_type_id, role_id): hostname = switch["hostname"] or switch["mgmt_ip"] results = nb_get("dcim/devices/", params={"name": hostname}).get("results", []) if results: logger.info(f" Device exists: {hostname}") return results[0] logger.info(f" Creating device: {hostname}") return nb_post("dcim/devices/", { "name": hostname, "site": site_id, "device_type": device_type_id, "role": role_id, "status": "active", "comments": f"Chassis ID: {switch['chassis_id']}\nDiscovered by lldp-mapper", }) def ensure_interface(device_id, port_name): results = nb_get("dcim/interfaces/", params={ "device_id": device_id, "name": port_name }).get("results", []) if results: return results[0] return nb_post("dcim/interfaces/", { "device": device_id, "name": port_name, "type": "1000base-t", }) def ensure_ip(switch, device_id): if not switch.get("mgmt_ip"): return ip_addr = f"{switch['mgmt_ip']}/24" results = nb_get("ipam/ip-addresses/", params={"address": ip_addr}).get("results", []) if results: ip_obj = results[0] else: logger.info(f" Creating IP: {ip_addr}") ip_obj = nb_post("ipam/ip-addresses/", { "address": ip_addr, "status": "active", }) if not ip_obj: return requests.patch( f"{NETBOX_URL}/api/dcim/devices/{device_id}/", headers=NB_HEADERS, json={"primary_ip4": ip_obj["id"]} ) def clear_all_cables(): logger.info("Clearing all existing cables from NetBox...") count = 0 while True: results = nb_get("dcim/cables/", params={"limit": 50}).get("results", []) if not results: break for cable in results: nb_delete(f"dcim/cables/{cable['id']}/") count += 1 logger.info(f"Deleted {count} cables.") def create_cable(device_map, link): hn_a = link["hn_a"] hn_b = link["hn_b"] port_a = link["port_a"] port_b = link["port_b"] if hn_a not in device_map or hn_b not in device_map: logger.warning(f" Skipping cable {hn_a}:{port_a} <-> {hn_b}:{port_b} — device not in NetBox") return dev_a = device_map[hn_a] dev_b = device_map[hn_b] iface_a = ensure_interface(dev_a["id"], port_a) iface_b = ensure_interface(dev_b["id"], port_b) if not iface_a or not iface_b: return logger.info(f" Creating cable: {hn_a}:{port_a} <-> {hn_b}:{port_b}") nb_post("dcim/cables/", { "a_terminations": [{"object_type": "dcim.interface", "object_id": iface_a["id"]}], "b_terminations": [{"object_type": "dcim.interface", "object_id": iface_b["id"]}], "status": "connected", }) def sync_netbox(switches, links): logger.info("=== Syncing to NetBox (FULL - cables will be cleared and re-created) ===") site = ensure_site() manufacturer = ensure_manufacturer() device_type = ensure_device_type(manufacturer["id"]) role = ensure_device_role() site_id = site["id"] device_type_id = device_type["id"] role_id = role["id"] device_map = {} for sw in switches: hostname = sw["hostname"] or sw["mgmt_ip"] logger.info(f"Processing device: {hostname}") device = ensure_device(sw, site_id, device_type_id, role_id) if device: device_map[hostname] = device ensure_ip(sw, device["id"]) logger.info(f"Devices synced: {len(device_map)}") clear_all_cables() logger.info("Re-creating cables from current scan data...") for link in links: create_cable(device_map, link) logger.info("NetBox sync complete.") # --- Main --- if __name__ == "__main__": logger.info("*** FULL SYNC — existing cables will be deleted and re-created ***") switches, links = load_db() logger.info(f"Loaded {len(switches)} switches and {len(links)} links from DB") sync_netbox(switches, links) logger.info("=== All done ===")