#!/usr/bin/env python3 """ Recursive single-target sync: Subsystem -> NetBox (Source of Truth), with depth. Usage: ./netbox_subsystem_ikejanum [--deep N] Behavior: - Start with . - Sync from Subsystem into NetBox (create device if missing). - If device missing in Subsystem but present in NetBox -> set NetBox status=failed. - While syncing: * If an IP is MOVED from another NetBox device -> enqueue that old device for sync (depth+1) * If a stale IP is being DELETED from a device -> find Subsystem device that owns that IP and enqueue it (depth+1) Depth: - Default depth: 3 - --deep N allowed, but N is hard-limited to <= 10. NEW: - --info-only: Print the raw Subsystem record (JSON) for the device and exit. No NetBox changes, no recursion. Exit codes: 0 success 1 runtime error / not found (info-only) 2 config/arg error """ # ========================= # USER CONFIG — EDIT HERE # ========================= NB_URL = "http://netbox.gt-tiso.ikeja.co.za" NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623" SUBSYSTEM_BASE = "https://subsystem.ikeja.co.za" SUBSYSTEM_TOKEN = "HJL+&XRCoeHwh5?13@gxvg86qD#kQfgc" ROLE_CPE = 1 TYPE_FOX100_CPE = 1 TYPE_FOX200 = 2 SITE_ID = 1 DEFAULT_MIN_LAST_DETECTED_MINUTES = 0 # 0 = accept any age DEFAULT_DEEP = 3 MAX_DEEP = 10 # ========================= import os import sys import json import time import logging import argparse from typing import Optional, Dict, Any, Tuple, List, Set, Deque from collections import deque import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from datetime import datetime, timezone from email.utils import parsedate_to_datetime # ------------- Logging ------------- def setup_logging(troubleshoot: bool) -> None: level = logging.DEBUG if troubleshoot else logging.INFO logging.basicConfig( level=level, format='%(asctime)s %(levelname)s %(message)s', datefmt='%H:%M:%S' ) def summarize_body(body: Any, limit: int = 500) -> str: if body is None: return "" if isinstance(body, (dict, list)): s = json.dumps(body) else: s = str(body) if len(s) > limit: return s[:limit] + f"... (+{len(s)-limit}B)" return s def is_ascii_or_die(label: str, value: str) -> None: try: value.encode('latin-1') except UnicodeEncodeError: bad = ''.join(ch for ch in value if ord(ch) > 127) logging.error( "%s contains non-ASCII characters (e.g. %r). Please paste the exact token without smart punctuation.", label, bad, ) raise SystemExit(2) # ------------- HTTP Session helpers ------------- class Http: def __init__(self, troubleshoot: bool = False): self.s_nb = self._new_session() self.s_subsystem = self._new_session() self.troubleshoot = troubleshoot def _new_session(self) -> requests.Session: s = requests.Session() retries = Retry( total=3, connect=3, read=3, status=3, backoff_factor=0.75, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=("GET", "POST", "PATCH", "PUT", "DELETE"), respect_retry_after_header=True, ) adapter = HTTPAdapter(max_retries=retries, pool_connections=16, pool_maxsize=32) s.mount('http://', adapter) s.mount('https://', adapter) s.headers['Accept'] = 'application/json' s.headers['Content-Type'] = 'application/json' return s def nb(self, method: str, url: str, token: str, **kw) -> requests.Response: headers = kw.pop('headers', {}) headers['Authorization'] = f'Token {token}' t0 = time.time() resp = self.s_nb.request(method, url, headers=headers, timeout=30, **kw) dt = (time.time() - t0) * 1000 if self.troubleshoot or resp.status_code >= 400: logging.debug( "NB %s %s [%s] %.1fms\n req:%s\n resp:%s", method, url, resp.status_code, dt, summarize_body(kw.get('data') or kw.get('json')), summarize_body(resp.text), ) return resp def subsystem(self, method: str, url: str, token: str, **kw) -> requests.Response: headers = kw.pop('headers', {}) headers['token'] = token t0 = time.time() resp = self.s_subsystem.request(method, url, headers=headers, timeout=30, **kw) dt = (time.time() - t0) * 1000 if self.troubleshoot or resp.status_code >= 400: logging.debug( "SUBSYS %s %s [%s] %.1fms\n resp:%s", method, url, resp.status_code, dt, summarize_body(resp.text), ) return resp # ------------- Utility ------------- def normalize_mac_from_subsystem(mac_raw: str) -> Optional[str]: if not mac_raw: return None s = mac_raw.strip().replace(":", "").replace("-", "") s = s.upper() try: chunks = [s[i:i+2] for i in range(0, len(s), 2)] return ":".join(chunks) except Exception: logging.warning(" !! failed to normalize MAC from subsystem: %r", mac_raw) return None def parse_last_detected(ts: Optional[str]) -> Optional[datetime]: if not ts: return None try: dt = parsedate_to_datetime(ts) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) else: dt = dt.astimezone(timezone.utc) return dt except Exception: logging.warning(" !! cannot parse last_detected: %r", ts) return None def minutes_age(now_utc: datetime, past_utc: datetime) -> float: return (now_utc - past_utc).total_seconds() / 60.0 def ip_strip_prefix(ip_with_prefix: str) -> str: # "10.0.0.1/32" -> "10.0.0.1" return (ip_with_prefix or "").split("/")[0].strip() # ------------- Subsystem client (with caching) ------------- class Subsystem: def __init__(self, base: str, token: str, http: Http): self.base = base.rstrip('/') self.token = token self.http = http self._cache_devices: Optional[list[Dict[str, Any]]] = None def fetch_outdoor_devices(self) -> list[Dict[str, Any]]: url = f"{self.base}/customers/wave-devices/get-outdoor-devices" r = self.http.subsystem("POST", url, self.token) r.raise_for_status() data = r.json() return data.get("device_list") or [] def _ensure_cache(self) -> None: if self._cache_devices is None: self._cache_devices = self.fetch_outdoor_devices() def find_device_by_name(self, target_name: str) -> Optional[Dict[str, Any]]: self._ensure_cache() assert self._cache_devices is not None for rec in self._cache_devices: name = (rec.get("name") or "").strip() if name.lower() == target_name.lower(): return rec return None def find_device_by_ip(self, ip: str) -> Optional[Dict[str, Any]]: if not ip: return None self._ensure_cache() assert self._cache_devices is not None ip = ip.strip() for rec in self._cache_devices: rip = (rec.get("ip") or "").strip() if rip == ip: return rec return None # ------------- NetBox API wrappers ------------- class NetBox: def __init__( self, base: str, token: str, role_cpe: int, type_fox100: int, type_fox200: int, site_id: int, http: Http, dry_run: bool = False, ): self.base = base.rstrip('/') self.token = token self.role_cpe = role_cpe self.type_fox100 = type_fox100 self.type_fox200 = type_fox200 self.site_id = site_id self.http = http self.dry = dry_run self.allow_moves = True def _url(self, path: str) -> str: return f"{self.base}{path}" def _req(self, method: str, path: str, **kw) -> requests.Response: if self.dry and method in ("POST", "PATCH", "PUT", "DELETE"): logging.info("DRY %s %s", method, path) r = requests.Response() if method == "POST": r.status_code = 201 r._content = b'{"id": 0}' else: r.status_code = 200 r._content = b'{}' r.headers['Content-Type'] = 'application/json' return r return self.http.nb(method, self._url(path), self.token, **kw) # --- devices --- def get_device_id_by_name(self, name: str) -> Optional[int]: r = self._req("GET", "/api/dcim/devices/", params={"name": name}) r.raise_for_status() data = r.json() if data.get('results'): return data['results'][0]['id'] return None def get_device(self, dev_id: int) -> Dict[str, Any]: r = self._req("GET", f"/api/dcim/devices/{dev_id}/") r.raise_for_status() return r.json() def get_device_name(self, dev_id: int) -> Optional[str]: r = self._req("GET", f"/api/dcim/devices/{dev_id}/") if r.status_code == 200: return r.json().get("name") return None def create_device(self, name: str, dtype_id: int) -> Optional[int]: payload = { "name": name, "role": self.role_cpe, "device_type": dtype_id, "site": self.site_id, "status": "active", } r = self._req("POST", "/api/dcim/devices/", json=payload) if r.status_code == 201: dev_id = r.json().get('id') logging.info(" -> created NetBox device name=%s id=%s", name, dev_id) return dev_id logging.error("device create failed (%s) HTTP=%s body=%s", name, r.status_code, summarize_body(r.text)) return None def patch_device(self, dev_id: int, patch: Dict[str, Any]) -> bool: if not patch: return True r = self._req("PATCH", f"/api/dcim/devices/{dev_id}/", json=patch) if 200 <= r.status_code < 300: return True logging.error("device patch failed dev=%s HTTP=%s body=%s", dev_id, r.status_code, summarize_body(r.text)) return False def set_primary_ip4_if_changed(self, dev: Dict[str, Any], ip_id: int) -> bool: curr = (dev.get('primary_ip4') or {}).get('id') if curr == ip_id: return True r = self._req("PATCH", f"/api/dcim/devices/{dev['id']}/", json={"primary_ip4": ip_id}) ok = 200 <= r.status_code < 300 if not ok: logging.error("set primary_ip4 failed dev=%s HTTP=%s body=%s", dev['id'], r.status_code, summarize_body(r.text)) return ok def set_status(self, dev_id: int, status: str) -> bool: r = self._req("PATCH", f"/api/dcim/devices/{dev_id}/", json={"status": status}) if 200 <= r.status_code < 300: return True logging.error("status patch failed dev=%s status=%s HTTP=%s body=%s", dev_id, status, r.status_code, summarize_body(r.text)) return False def add_journal_entry(self, dev_id: int, comments: str, kind: str = "info") -> bool: payload = { "assigned_object_type": "dcim.device", "assigned_object_id": dev_id, "kind": kind, "comments": comments, } r = self._req("POST", "/api/extras/journal-entries/", json=payload) if 200 <= r.status_code < 300: return True logging.error( "journal entry create failed dev=%s kind=%s HTTP=%s body=%s", dev_id, kind, r.status_code, summarize_body(r.text) ) return False # --- interfaces --- def get_iface_id(self, dev_id: int, name: str) -> Optional[int]: r = self._req("GET", "/api/dcim/interfaces/", params={"device_id": dev_id, "name": name}) r.raise_for_status() data = r.json() if data.get('results'): return data['results'][0]['id'] return None def ensure_eth0(self, dev_id: int) -> Optional[int]: ifid = self.get_iface_id(dev_id, 'eth0') if ifid: return ifid payload = {"device": dev_id, "name": "eth0", "type": "1000base-t"} r = self._req("POST", "/api/dcim/interfaces/", json=payload) if r.status_code == 201: return r.json()['id'] if r.status_code == 400: return self.get_iface_id(dev_id, 'eth0') logging.error("interface create failed device=%s HTTP=%s body=%s", dev_id, r.status_code, summarize_body(r.text)) return None def get_device_id_of_interface(self, iface_id: int) -> Optional[int]: r = self._req("GET", f"/api/dcim/interfaces/{iface_id}/") if r.status_code == 200: return (r.json().get('device') or {}).get('id') return None def get_interface_detail(self, iface_id: int) -> Optional[Dict[str, Any]]: r = self._req("GET", f"/api/dcim/interfaces/{iface_id}/") if 200 <= r.status_code < 300: return r.json() logging.error("interface fetch failed iface=%s HTTP=%s body=%s", iface_id, r.status_code, summarize_body(r.text)) return None # --- IP inventory helpers --- def list_device_ips(self, dev_id: int) -> list[Dict[str, Any]]: r = self._req("GET", "/api/ipam/ip-addresses/", params={"device_id": dev_id, "limit": 1000}) if r.status_code == 200: return r.json().get('results') or [] # fallback by interface ips: list[Dict[str, Any]] = [] r2 = self._req("GET", "/api/dcim/interfaces/", params={"device_id": dev_id, "limit": 1000}) if r2.status_code == 200: for iface in (r2.json().get('results') or []): ifid = iface.get('id') r3 = self._req("GET", "/api/ipam/ip-addresses/", params={ "assigned_object_type": "dcim.interface", "assigned_object_id": ifid, "limit": 1000, }) if r3.status_code == 200: ips.extend(r3.json().get('results') or []) return ips def prune_other_ips_with_details(self, dev_id: int, keep_ip_id: int) -> List[Tuple[int, str]]: """ Delete all other IPs assigned to this device, keeping only keep_ip_id. Returns list of (ip_id, address) that were removed. """ removed: List[Tuple[int, str]] = [] all_ips = self.list_device_ips(dev_id) for rec in all_ips: rid = rec.get('id') if rid == keep_ip_id: continue addr = rec.get('address') logging.warning(" -> removing stale IP %s (id=%s) from device %s", addr, rid, dev_id) self._req("DELETE", f"/api/ipam/ip-addresses/{rid}/") if rid is not None and addr: removed.append((int(rid), str(addr))) return removed # --- MAC helpers --- def ensure_single_mac_on_iface(self, iface_id: int, mac_norm: str) -> Tuple[int, int]: created = 0 deleted = 0 iface = self.get_interface_detail(iface_id) if not iface: logging.error(" !! cannot fetch interface detail for MAC sync (iface=%s)", iface_id) return (0, 0) macs = iface.get("mac_addresses") or [] matching_ids: list[int] = [] bad_ids: list[int] = [] for m in macs: mid = m.get("id") mval = (m.get("mac_address") or "").upper() if mval == mac_norm: if mid is not None: matching_ids.append(mid) else: if mid is not None: bad_ids.append(mid) if not matching_ids: payload = { "mac_address": mac_norm, "assigned_object_type": "dcim.interface", "assigned_object_id": iface_id, } logging.info(" -> create MAC %s on iface %s", mac_norm, iface_id) r = self._req("POST", "/api/dcim/mac-addresses/", json=payload) if r.status_code == 201: created += 1 else: logging.error(" !! MAC create failed iface=%s mac=%s HTTP=%s body=%s", iface_id, mac_norm, r.status_code, summarize_body(r.text)) else: extra_ids = matching_ids[1:] bad_ids.extend([i for i in extra_ids if i is not None]) for mid in bad_ids: logging.warning(" -> removing stale/duplicate MAC entry id=%s from iface=%s", mid, iface_id) self._req("DELETE", f"/api/dcim/mac-addresses/{mid}/") deleted += 1 return (created, deleted) # --- IP addresses --- def get_ip_by_address(self, addr: str) -> Dict[str, Any] | None: r = self._req("GET", "/api/ipam/ip-addresses/", params={"address": f"{addr}/32"}) r.raise_for_status() res = r.json().get('results') or [] return res[0] if res else None def assign_ip_to_iface(self, ip_id: int, iface_id: int) -> bool: r = self._req("PATCH", f"/api/ipam/ip-addresses/{ip_id}/", json={ "assigned_object_type": "dcim.interface", "assigned_object_id": iface_id, }) return r.status_code == 200 def ensure_ip_for_device(self, dev_id: int, iface_id: int, addr: str) -> Tuple[Optional[int], str, Optional[int]]: """ Ensure addr/32 exists and is assigned to iface_id on dev_id. Returns (ip_id, action, old_device_id_if_moved) action: created, reused, assigned, moved, skipped, error """ ip_rec = self.get_ip_by_address(addr) if not ip_rec: payload = { "address": f"{addr}/32", "status": "active", "assigned_object_type": "dcim.interface", "assigned_object_id": iface_id, } r = self._req("POST", "/api/ipam/ip-addresses/", json=payload) if r.status_code == 201: return (r.json()['id'], "created", None) logging.error("ip create failed iface=%s addr=%s HTTP=%s body=%s", iface_id, addr, r.status_code, summarize_body(r.text)) return (None, "error", None) ip_id = ip_rec['id'] aot = ip_rec.get('assigned_object_type') or '' if not aot: ok = self.assign_ip_to_iface(ip_id, iface_id) return (ip_id if ok else None, "assigned" if ok else "error", None) if aot == 'dcim.interface': assigned_ifid = ip_rec.get('assigned_object_id') if str(assigned_ifid) == str(iface_id): return (ip_id, "reused", None) assigned_dev = self.get_device_id_of_interface(assigned_ifid) if assigned_ifid else None if assigned_dev and str(assigned_dev) == str(dev_id): logging.info(" -> IP %s already on same device (iface=%s); reusing.", addr, assigned_ifid) return (ip_id, "reused", None) if self.allow_moves: old_dev_id = assigned_dev # clear primary_ip4 on old device if needed if old_dev_id: r_old = self._req("GET", f"/api/dcim/devices/{old_dev_id}/") if r_old.status_code == 200: old_primary_id = (r_old.json().get("primary_ip4") or {}).get("id") if str(old_primary_id) == str(ip_id): r_clr = self._req("PATCH", f"/api/dcim/devices/{old_dev_id}/", json={"primary_ip4": None}) if not (200 <= r_clr.status_code < 300): logging.error(" -> cannot clear old device %s primary_ip4 for IP %s; aborting move", old_dev_id, addr) return (None, "error", None) logging.warning( " -> IP %s currently belongs to device %s (iface %s); moving to this device (iface %s)", addr, old_dev_id, assigned_ifid, iface_id, ) r2 = self._req("PATCH", f"/api/ipam/ip-addresses/{ip_id}/", json={ "assigned_object_type": "dcim.interface", "assigned_object_id": iface_id, }) if r2.status_code == 200: return (ip_id, "moved", old_dev_id) logging.error(" -> move failed for IP %s HTTP=%s body=%s", addr, r2.status_code, summarize_body(r2.text)) return (None, "error", None) return (None, "skipped", None) logging.warning(" -> IP %s assigned to %s; skipping.", addr, aot) return (None, "skipped", None) # ------------- Core sync functions ------------- def sync_from_subsystem_record( target: str, rec: Dict[str, Any], nb: NetBox, subsystem: Subsystem, now_utc: datetime, min_last_detected_minutes: int, enqueue_fn, ) -> Dict[str, Any]: """ enqueue_fn(name: str, reason: str) -> None """ summary = { "target": target, "subsystem_found": True, "netbox_created": False, "status_set_active": False, "node_updated": False, "mac_created": 0, "mac_deleted": 0, "ip_action": "none", "ips_pruned": 0, "primary_set": False, "skipped_stale": False, "errors": 0, } name = rec.get("name") or target ip = rec.get("ip") mac_raw = rec.get("mac") connected_node = rec.get("connected_node") last_detected = rec.get("last_detected") dt_last = parse_last_detected(last_detected) if min_last_detected_minutes > 0: if dt_last is None: logging.info(" -> skip: last_detected unavailable; requires <= %s minutes", min_last_detected_minutes) summary["skipped_stale"] = True return summary age_min = minutes_age(now_utc, dt_last) if age_min > min_last_detected_minutes: logging.info(" -> skip: last_detected age %.1f min > allowed %s min", age_min, min_last_detected_minutes) summary["skipped_stale"] = True return summary dev_id = nb.get_device_id_by_name(name) if not dev_id: logging.info(" -> device not found in NetBox; creating") dev_id = nb.create_device(name, nb.type_fox100) if not dev_id: summary["errors"] += 1 return summary summary["netbox_created"] = True dev = nb.get_device(dev_id) curr_status = dev.get("status", {}).get("value") if isinstance(dev.get("status"), dict) else dev.get("status") if curr_status != "active": active_reason = "device present in subsystem" if min_last_detected_minutes > 0: active_reason += f" and last_detected within {min_last_detected_minutes} minutes" else: active_reason += " and freshness filter accepts it" if nb.set_status(dev_id, "active"): nb.add_journal_entry(dev_id, f"setting to active because {active_reason}", kind="info") summary["status_set_active"] = True dev = nb.get_device(dev_id) else: summary["errors"] += 1 cf = dev.get("custom_fields") or {} current_node = (cf.get("nodeName") if isinstance(cf, dict) else None) patch: Dict[str, Any] = {} if connected_node and connected_node != current_node: patch.setdefault("custom_fields", {})["nodeName"] = connected_node logging.info(" -> patch custom_fields.nodeName: %r -> %r", current_node, connected_node) if nb.patch_device(dev_id, patch): summary["node_updated"] = True dev = nb.get_device(dev_id) iface_id = nb.ensure_eth0(dev_id) if not iface_id: logging.error(" !! interface ensure/create failed for device=%s", dev_id) summary["errors"] += 1 return summary mac_norm = normalize_mac_from_subsystem(mac_raw) if mac_raw else None if mac_norm: logging.info(" -> ensure single MAC %s on iface %s", mac_norm, iface_id) c, d = nb.ensure_single_mac_on_iface(iface_id, mac_norm) summary["mac_created"] += c summary["mac_deleted"] += d else: logging.info(" -> MAC missing/unusable from subsystem; skipping MAC sync") if ip and str(ip).lower() != "null" and ip != "0.0.0.0": logging.info(" -> ensure IP %s for device %s (iface %s)", ip, dev_id, iface_id) ip_id, ip_action, old_dev_id = nb.ensure_ip_for_device(dev_id, iface_id, ip) summary["ip_action"] = ip_action # If we moved from some other NetBox device, enqueue it for repair if ip_action == "moved" and old_dev_id: old_name = nb.get_device_name(old_dev_id) if old_name and old_name.lower() != name.lower(): enqueue_fn(old_name, f"ip_moved_away:{ip}") if ip_id: removed = nb.prune_other_ips_with_details(dev_id, ip_id) summary["ips_pruned"] = len(removed) # For each deleted stale IP, see who owns that IP in Subsystem and enqueue them. for _rid, addr_pref in removed: stale_ip = ip_strip_prefix(addr_pref) rec2 = subsystem.find_device_by_ip(stale_ip) if rec2: n2 = rec2.get("name") if n2 and n2.lower() != name.lower(): enqueue_fn(n2, f"stale_ip_deleted:{stale_ip}") dev = nb.get_device(dev_id) before = (dev.get("primary_ip4") or {}).get("id") if nb.set_primary_ip4_if_changed(dev, ip_id): if str(before) != str(ip_id): summary["primary_set"] = True else: summary["errors"] += 1 else: logging.info(" -> IP missing or 0.0.0.0; skipping IP sync") return summary def mark_failed_if_in_netbox_only(target: str, nb: NetBox) -> Dict[str, Any]: summary = { "target": target, "subsystem_found": False, "netbox_exists": False, "status_set_failed": False, "errors": 0, } dev_id = nb.get_device_id_by_name(target) if not dev_id: return summary summary["netbox_exists"] = True dev = nb.get_device(dev_id) curr_status = dev.get("status", {}).get("value") if isinstance(dev.get("status"), dict) else dev.get("status") if curr_status == "failed": return summary failed_reason = "device missing in subsystem" if nb.set_status(dev_id, "failed"): nb.add_journal_entry(dev_id, f"setting to failed, because {failed_reason}", kind="warning") summary["status_set_failed"] = True else: summary["errors"] += 1 return summary def one_liner_single(result: Dict[str, Any]) -> str: t = result.get("target", "?") if result.get("subsystem_found") is False: if result.get("netbox_exists"): if result.get("status_set_failed"): return f"{t}: not in subsystem -> NetBox status set to failed" return f"{t}: not in subsystem -> NetBox status NOT changed (error)" return f"{t}: not in subsystem and not in NetBox -> nothing to do" if result.get("skipped_stale"): return f"{t}: subsystem record skipped due to last_detected freshness filter" parts = [] if result.get("netbox_created"): parts.append("created in NetBox") if result.get("status_set_active"): parts.append("status set to active") if result.get("node_updated"): parts.append("nodeName updated") if (result.get("mac_created", 0) or 0) > 0 or (result.get("mac_deleted", 0) or 0) > 0: parts.append(f"mac c{result.get('mac_created',0)}/d{result.get('mac_deleted',0)}") ia = result.get("ip_action") if ia and ia != "none": parts.append(f"ip {ia}") if (result.get("ips_pruned", 0) or 0) > 0: parts.append(f"ips pruned={result.get('ips_pruned')}") if result.get("primary_set"): parts.append("primary_ip4 set") if (result.get("errors") or 0) > 0: parts.append(f"errors={result.get('errors')}") if not parts: return f"{t}: already in sync (no changes)" return f"{t}: " + ", ".join(parts) # ------------- Depth driver ------------- def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("device_name", help="Exact device name to sync (e.g. ikeja12345)") ap.add_argument( "--info-only", action="store_true", default=False, help="Print Subsystem record for the device and exit (no NetBox changes)", ) ap.add_argument( "--min-last-detected-minutes", type=int, default=DEFAULT_MIN_LAST_DETECTED_MINUTES, help="Require subsystem last_detected within N minutes (0 = accept any age)", ) ap.add_argument( "--deep", type=int, default=DEFAULT_DEEP, help=f"Recursive depth (default {DEFAULT_DEEP}, max {MAX_DEEP})", ) ap.add_argument("--troubleshoot", action='store_true', default=bool(os.getenv('TROUBLESHOOT'))) ap.add_argument("--dry-run", action='store_true', default=False) args = ap.parse_args() setup_logging(args.troubleshoot) if not NB_TOKEN: logging.error("Set NB_TOKEN in script") return 2 if not SUBSYSTEM_TOKEN: logging.error("Set SUBSYSTEM_TOKEN in script") return 2 is_ascii_or_die("NB_TOKEN", NB_TOKEN) is_ascii_or_die("SUBSYSTEM_TOKEN", SUBSYSTEM_TOKEN) min_min = max(0, int(args.min_last_detected_minutes)) deep = int(args.deep) if deep < 1: deep = 1 if deep > MAX_DEEP: logging.warning("[WARN] --deep %s requested, capping to %s", deep, MAX_DEEP) deep = MAX_DEEP http = Http(troubleshoot=args.troubleshoot) nb = NetBox( NB_URL, NB_TOKEN, ROLE_CPE, TYPE_FOX100_CPE, TYPE_FOX200, SITE_ID, http, dry_run=args.dry_run, ) subsystem = Subsystem(SUBSYSTEM_BASE, SUBSYSTEM_TOKEN, http) # --info-only mode (no NetBox changes, no recursion) if args.info_only: rec = subsystem.find_device_by_name(args.device_name.strip()) if not rec: print(f"NOT_FOUND {args.device_name.strip()} in Subsystem") return 1 print(json.dumps(rec, indent=2, sort_keys=True)) return 0 now_utc = datetime.now(timezone.utc) # BFS queue: (name, depth, reason) q: Deque[Tuple[str, int, str]] = deque() seen: Set[str] = set() root = args.device_name.strip() q.append((root, 0, "root")) seen.add(root.lower()) totals = { "processed": 0, "subsystem_found": 0, "netbox_failed_set": 0, "created": 0, "errors": 0, "enqueued": 0, } def enqueue(name: str, reason: str, parent_depth: int = 0) -> None: nonlocal q, seen, totals, deep if not name: return k = name.lower() if k in seen: return next_depth = parent_depth + 1 if next_depth > deep: return seen.add(k) q.append((name, next_depth, reason)) totals["enqueued"] += 1 logging.info(" -> enqueue depth=%s name=%s reason=%s", next_depth, name, reason) per_device_one_liners: List[str] = [] try: while q: name, d, reason = q.popleft() totals["processed"] += 1 logging.info("[INFO] depth=%s name=%s reason=%s", d, name, reason) rec = subsystem.find_device_by_name(name) if rec: totals["subsystem_found"] += 1 def enq_child(child_name: str, child_reason: str) -> None: enqueue(child_name, child_reason, parent_depth=d) res = sync_from_subsystem_record( name, rec, nb, subsystem, now_utc, min_min, enq_child ) if res.get("netbox_created"): totals["created"] += 1 if (res.get("errors") or 0) > 0: totals["errors"] += 1 per_device_one_liners.append(one_liner_single(res)) else: res2 = mark_failed_if_in_netbox_only(name, nb) if res2.get("status_set_failed"): totals["netbox_failed_set"] += 1 if (res2.get("errors") or 0) > 0: totals["errors"] += 1 per_device_one_liners.append(one_liner_single(res2)) except Exception as e: logging.exception("Unhandled exception in deep sync: %s", e) print(f"{root}: error (exception during deep sync)") return 1 for line in per_device_one_liners: print(line) print( f"SUMMARY: root={root} processed={totals['processed']} " f"subsys_found={totals['subsystem_found']} created={totals['created']} " f"failed_set={totals['netbox_failed_set']} enqueued={totals['enqueued']} errors={totals['errors']}" ) return 0 if __name__ == "__main__": sys.exit(main())