Refactor Vchassis_checker script for improved readability and performance. Updated regex formatting, enhanced logging messages, and implemented bulk updates for interface names. Added scheduling_enabled flag and adjusted tag handling logic.

check_vchassis_stack_ports.py

@@ -9,69 +9,64 @@ from utilities.exceptions import AbortScript
 
 
 class Vchassis_checker(Script):
-    TAGS = 'sync:vchas_interface'
+    TAGS = "sync:vchas_interface"
-    IFACE_RE = re.compile(r'^(?P<prefix>[A-Za-z]+)\s?(?:(?P<slot>\d+)/)?(?P<rest>\d+/\d+)$')
+    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
-        name = 'Переопределение интерфейсов'
+        scheduling_enabled = False
-        description = 'Переопределение интерфейсов у стекированных коммутаторов'
+        name = "Переопределение интерфейсов"
+        description = "Переопределение интерфейсов у стекированных коммутаторов"
 
     virtual_chassis = ObjectVar(
         model=VirtualChassis,
-        query_params={'tags': None},
+        query_params={"tags": None},
     )
 
     def run(self, data, commit):
         if Tag.objects.filter(name=self.TAGS).exists():
             tag = Tag.objects.get(name=self.TAGS)
         else:
-            self.log_warning(f'Отсутствует tag {self.TAGS}')
+            self.log_warning(f"Отсутствует tag {self.TAGS}")
             tag_data = {
-                'color': ColorChoices.COLOR_LIGHT_BLUE,
+                "color": ColorChoices.COLOR_LIGHT_BLUE,
-                'name': self.TAGS,
+                "name": self.TAGS,
-                'slug': self.TAGS.replace(':', '_'),
+                "slug": self.TAGS.replace(":", "_"),
-                'description': 'тег для автоматического назначения на синхронизированные интерфейсы vchassis',
+                "description": "тег для автоматического назначения на синхронизированные интерфейсы vchassis",
             }
             tag = Tag(**tag_data)
             if commit:
                 tag.full_clean()
                 tag.save()
                 self.log_success(
-                    f'Создан тег: {self.TAGS}',
+                    f"Создан тег: {self.TAGS}",
                     obj=tag,
                 )
-        vc = data['virtual_chassis']
+        vc = data["virtual_chassis"]
-        devices: list[Device] = list(vc.members.all().order_by('vc_position'))
+        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} отсутствует мастер')
+            raise AbortScript(f"В стеке {vc.name} отсутствует мастер")
-        for device in devices:
-            interfaces = Interface.objects.filter(device=device)
-            for iface in interfaces:
-                m = self.IFACE_RE.match(iface)
-                prefix = m.group('prefix')
-                slot = m.group('slot')
-                rest = m.group('rest')
-
-        # Все интерфейсы всех устройств стека одним запросом
         device_ids = [d.pk for d in devices]
-        interfaces_qs = Interface.objects.filter(device_id__in=device_ids)
+        interfaces_qs = Interface.objects.filter(
+            device_id__in=device_ids
+        ).select_related("device")
 
-        if commit:
+        to_update = []
-            # Вариант 1: цикл с .save() — вызываются clean(), save(), сигналы и change log (ObjectChange)
         for iface in interfaces_qs:
-                iface.description = f'{iface.device.name}:{iface.name}'  # пример
+            m = self.IFACE_RE.match(iface.name)
-                iface.save()
+            if m:
+                prefix = m.group("prefix")
+                slot = m.group("slot")
+                rest = m.group("rest")
+                iface.name = f"{prefix}{iface.device.vc_position}/{rest}"
+        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)
 
-            # Вариант 2: массовый UPDATE без change log (быстро, но без аудита):
+            self.log_success(f"Обновлено интерфейсов: {len(to_update)}")
-            # interfaces_qs.update(description=...)  # одно значение для всех
-            # Или bulk_update для разных значений по объектам:
-            # to_update = []
-            # for iface in interfaces_qs:
-            #     iface.description = f'{iface.device.name}:{iface.name}'
-            #     to_update.append(iface)
-            # Interface.objects.bulk_update(to_update, ['description'])
         else:
-            count = interfaces_qs.count()
+            self.log_success(f"Будет обновлено интерфейсов: {len(to_update)}")
-            self.log_success(f'Будет обновлено интерфейсов: {count}')
-