76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
import re
|
||
|
||
from dcim.models.devices import Device
|
||
from extras.models.tags import Tag
|
||
from extras.scripts import ObjectVar, Script
|
||
from dcim.models import VirtualChassis, Interface
|
||
from netbox.choices import ColorChoices
|
||
from utilities.exceptions import AbortScript
|
||
|
||
TAGS = "sync:vchas_interface"
|
||
|
||
|
||
class Vchassis_checker(Script):
|
||
IFACE_RE = re.compile(
|
||
r"^(?P<prefix>[A-Za-z]+)\s?(?:(?P<slot>\d+)/)?(?P<rest>\d+/\d+)$"
|
||
)
|
||
|
||
class Meta(Script.Meta):
|
||
commit_default = False
|
||
scheduling_enabled = False
|
||
name = "Переопределение интерфейсов"
|
||
description = "Переопределение интерфейсов у стекированных коммутаторов"
|
||
|
||
virtual_chassis = ObjectVar(
|
||
model=VirtualChassis,
|
||
description="Виртуальное шасси без тегов",
|
||
query_params={"tag__n": [TAGS.replace(":", "_")]},
|
||
)
|
||
|
||
def run(self, data, commit):
|
||
if Tag.objects.filter(name=TAGS).exists():
|
||
tag = Tag.objects.get(name=TAGS)
|
||
else:
|
||
self.log_warning(f"Отсутствует tag {TAGS}")
|
||
tag_data = {
|
||
"color": ColorChoices.COLOR_LIGHT_BLUE,
|
||
"name": TAGS,
|
||
"slug": TAGS.replace(":", "_"),
|
||
"description": "тег для автоматического назначения на синхронизированные интерфейсы vchassis",
|
||
}
|
||
tag = Tag(**tag_data)
|
||
if commit:
|
||
tag.full_clean()
|
||
tag.save()
|
||
self.log_success(
|
||
f"Создан тег: {TAGS}",
|
||
obj=tag,
|
||
)
|
||
vc = data["virtual_chassis"]
|
||
devices: list[Device] = list(vc.members.all().order_by("vc_position"))
|
||
if not vc._state.adding and vc.master and vc.master not in devices:
|
||
raise AbortScript(f"В стеке {vc.name} отсутствует мастер")
|
||
device_ids = [d.pk for d in devices]
|
||
interfaces_qs = Interface.objects.filter(
|
||
device_id__in=device_ids
|
||
).select_related("device")
|
||
|
||
to_update = []
|
||
for iface in interfaces_qs:
|
||
m = self.IFACE_RE.match(iface.name)
|
||
if m:
|
||
prefix = m.group("prefix")
|
||
slot = m.group("slot")
|
||
rest = m.group("rest")
|
||
iface.name = f"{prefix}{iface.device.vc_position}/{rest}"
|
||
to_update.append(iface)
|
||
if commit:
|
||
Interface.objects.bulk_update(to_update, fields=["name"], batch_size=100)
|
||
vc.tags.add(tag)
|
||
for iface in to_update:
|
||
iface.tags.add(tag)
|
||
|
||
self.log_success(f"Обновлено интерфейсов: {len(to_update)}")
|
||
else:
|
||
self.log_success(f"Будет обновлено интерфейсов: {len(to_update)}")
|