Netbox/topology/export_topology.py

"""
Run INSIDE the NetBox Django shell to export the field-switch topology
(switches + every patch panel / splice module they physically pass through)
as JSON, for generate_drawio.py to render.

Usage (from the NetBox host, e.g. /opt/netbox):

    docker exec netbox-netbox-1 python3 /opt/netbox/netbox/manage.py shell \
        -c "$(cat topology/export_topology.py)" 2>/dev/null > topology/topo_full.json

The script only reads data -- it never modifies NetBox. If a switch's
connected_endpoints comes back empty when it clearly shouldn't, that is a
NetBox cable-path bug, not a bug in this script -- see the "gotchas" section
of the repo README before assuming the physical link doesn't exist.

Two passes:
  1. Walk every switch's cable path (Interface.trace()) to find every
     device -- patch panel, splice module, other switch -- it physically
     touches. This is just to build the device set; NetBox's cached path
     is what can go stale (see README), which is why pass 2 doesn't rely
     on it for the edges themselves.
  2. Pull every real Cable among those devices directly, so each edge
     carries the exact port name on both ends (what generate_drawio.py
     labels connectors with) instead of just "these two are connected."

Also exports every ipam.VLAN (with its prefixes) and, for each edge whose
endpoint is a real Interface (not a passive patch-panel port), that
interface's 802.1Q mode/untagged/tagged VLANs. As of this writing NetBox
has zero VLANs defined here, so `vlans` comes back `[]` and every edge's
`a_vlan`/`b_vlan` comes back `null` -- generate_drawio.py renders that as
an explicit "not configured yet" placeholder rather than silently
omitting the VLAN legend. The moment VLANs and interface assignments get
added in NetBox, rerunning this (no code changes needed) starts filling
the diagram in for real.
"""
from dcim.models import Device, Interface, Cable
from ipam.models import VLAN
import json

def vlan_info(vlan):
    if not vlan:
        return None
    prefixes = [str(p.prefix) for p in vlan.prefixes.all()]
    return {"vid": vlan.vid, "name": vlan.name, "prefix": prefixes[0] if prefixes else None}

def port_vlans(termination):
    """VLAN mode/tagging for a real Interface termination; None for a
    passive patch-panel front/rear port, which has no 802.1Q concept."""
    if not isinstance(termination, Interface):
        return None
    return {
        "mode": termination.mode,
        "untagged": vlan_info(termination.untagged_vlan),
        "tagged": [vlan_info(v) for v in termination.tagged_vlans.all()],
    }

switch_ids = set(
    Device.objects.filter(
        role__slug__in=["core-switch", "dist-switch", "field-switch"]
    ).values_list("id", flat=True)
)

ifaces = Interface.objects.filter(
    device_id__in=switch_ids, cable__isnull=False
).select_related("device")

touched_devices = set(switch_ids)
for i in ifaces:
    path = i._path
    if not path or not path.is_active:
        continue
    for hop in i.trace():
        for item in hop:
            for t in item:
                dev = getattr(t, "device", None)
                if dev:
                    touched_devices.add(dev.id)

cables = Cable.objects.filter(
    terminations__termination_type__model__in=["interface", "frontport", "rearport"],
).distinct()

edges = {}
for c in cables:
    terms = list(c.terminations.all())
    a_side = [t for t in terms if t.cable_end == "A"]
    b_side = [t for t in terms if t.cable_end == "B"]
    for ta in a_side:
        for tb in b_side:
            da = getattr(ta.termination, "device", None)
            db = getattr(tb.termination, "device", None)
            if not da or not db:
                continue
            if da.id in touched_devices and db.id in touched_devices and da.id != db.id:
                key = tuple(sorted((da.id, db.id)))
                edges[key] = {
                    "a": da.id, "a_port": ta.termination.name, "a_vlan": port_vlans(ta.termination),
                    "b": db.id, "b_port": tb.termination.name, "b_vlan": port_vlans(tb.termination),
                }

devs = Device.objects.filter(id__in=touched_devices).select_related(
    "site", "role", "device_type"
)

out = {
    "devices": [
        {
            "id": d.id,
            "name": d.name,
            "role": d.role.slug,
            "role_name": d.role.name,
            "site": d.site.name if d.site else "",
            "status": str(d.status),
            "device_type": d.device_type.model,
        }
        for d in devs
    ],
    "edges": list(edges.values()),
    "vlans": [
        {"vid": v.vid, "name": v.name, "description": v.description,
         "prefixes": [str(p.prefix) for p in v.prefixes.all()],
         "site": v.site.name if v.site else ""}
        for v in VLAN.objects.all().select_related("site").order_by("vid")
    ],
}
print(json.dumps(out))