fix: match patchpanel ports by identifier

netbox_utilities/__init__.py

@@ -1,6 +1,6 @@
 from netbox.plugins import PluginConfig, get_plugin_config
 
-__version__ = "0.7.0"
+__version__ = "0.7.1"
 
 
 class NetBoxUtilitiesConfig(PluginConfig):

netbox_utilities/migrations/0005_fix_patchpanel_port_number_mappings.py

@@ -0,0 +1,81 @@
+import re
+from collections import defaultdict
+
+from django.db import migrations
+
+
+def natural_port_key(port):
+    parts = tuple(
+        int(part) if part.isdigit() else part.casefold()
+        for part in re.split(r"(\d+)", str(port.name))
+    )
+    return parts, port.pk
+
+
+def port_identifier(port):
+    name = str(port.name).casefold()
+    name = re.sub(r"\b(?:front|rear)(?:[\s_-]*port)?(?=\b|\d)", " ", name)
+    name = re.sub(r"^(?:fp|rp|f|r)(?=\s*[-_.:/]?\s*\d)", "", name)
+    tokens = re.findall(r"\d+|[^\W\d_]+", name)
+    identifier = tuple(int(token) if token.isdigit() else token for token in tokens)
+    return identifier or None
+
+
+def pair_ports(front_ports, rear_ports):
+    front_by_identifier = defaultdict(list)
+    rear_by_identifier = defaultdict(list)
+    for port in front_ports:
+        if identifier := port_identifier(port):
+            front_by_identifier[identifier].append(port)
+    for port in rear_ports:
+        if identifier := port_identifier(port):
+            rear_by_identifier[identifier].append(port)
+
+    pairs = []
+    for identifier, matching_front_ports in front_by_identifier.items():
+        pairs.extend(
+            zip(
+                sorted(matching_front_ports, key=natural_port_key),
+                sorted(rear_by_identifier.get(identifier, []), key=natural_port_key),
+            )
+        )
+    return pairs
+
+
+def fix_existing_patchpanel_mappings(apps, schema_editor):
+    Device = apps.get_model("dcim", "Device")
+    FrontPort = apps.get_model("dcim", "FrontPort")
+    PortMapping = apps.get_model("dcim", "PortMapping")
+    RearPort = apps.get_model("dcim", "RearPort")
+    database = schema_editor.connection.alias
+
+    device_ids = Device.objects.using(database).filter(role__name__iexact="Patchpanel").values_list("pk", flat=True)
+    for device_id in device_ids.iterator():
+        front_ports = FrontPort.objects.using(database).filter(device_id=device_id).only("pk", "name")
+        rear_ports = RearPort.objects.using(database).filter(device_id=device_id).only("pk", "name")
+        pairs = pair_ports(front_ports, rear_ports)
+
+        PortMapping.objects.using(database).filter(device_id=device_id).delete()
+        PortMapping.objects.using(database).bulk_create(
+            [
+                PortMapping(
+                    device_id=device_id,
+                    front_port_id=front_port.pk,
+                    rear_port_id=rear_port.pk,
+                    front_port_position=1,
+                    rear_port_position=1,
+                )
+                for front_port, rear_port in pairs
+            ],
+            batch_size=1000,
+        )
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("netbox_utilities", "0004_patchpanel_port_mappings"),
+    ]
+
+    operations = [
+        migrations.RunPython(fix_existing_patchpanel_mappings, migrations.RunPython.noop),
+    ]

netbox_utilities/patchpanel.py

@@ -1,4 +1,5 @@
 import re
+from collections import defaultdict
 from contextvars import ContextVar
 from functools import wraps
 
@@ -20,14 +21,42 @@ def natural_port_key(port):
     return parts, getattr(port, "pk", 0) or 0
 
 
+def port_identifier(port):
+    """Normalize a front/rear port name to its side-independent identifier."""
+    name = str(getattr(port, "name", "")).casefold()
+    name = re.sub(r"\b(?:front|rear)(?:[\s_-]*port)?(?=\b|\d)", " ", name)
+    name = re.sub(r"^(?:fp|rp|f|r)(?=\s*[-_.:/]?\s*\d)", "", name)
+    tokens = re.findall(r"\d+|[^\W\d_]+", name)
+    identifier = tuple(int(token) if token.isdigit() else token for token in tokens)
+    return identifier or None
+
+
 def is_patchpanel(device):
     role = getattr(device, "role", None)
     return bool(role and str(getattr(role, "name", "")).casefold() == PATCHPANEL_ROLE_NAME.casefold())
 
 
 def pair_ports(front_ports, rear_ports):
-    """Pair naturally sorted front and rear ports one-to-one."""
-    return list(zip(sorted(front_ports, key=natural_port_key), sorted(rear_ports, key=natural_port_key)))
+    """Pair only front and rear ports with the same normalized identifier."""
+    front_by_identifier = defaultdict(list)
+    rear_by_identifier = defaultdict(list)
+    for port in front_ports:
+        if identifier := port_identifier(port):
+            front_by_identifier[identifier].append(port)
+    for port in rear_ports:
+        if identifier := port_identifier(port):
+            rear_by_identifier[identifier].append(port)
+
+    pairs = []
+    for identifier, matching_front_ports in front_by_identifier.items():
+        matching_rear_ports = rear_by_identifier.get(identifier, [])
+        pairs.extend(
+            zip(
+                sorted(matching_front_ports, key=natural_port_key),
+                sorted(matching_rear_ports, key=natural_port_key),
+            )
+        )
+    return sorted(pairs, key=lambda pair: natural_port_key(pair[0]))
 
 
 def synchronize_patchpanel(device_id):

netbox_utilities/tests/test_patchpanel.py

@@ -2,7 +2,7 @@ from types import SimpleNamespace
 
 from django.test import SimpleTestCase
 
-from netbox_utilities.patchpanel import is_patchpanel, pair_ports
+from netbox_utilities.patchpanel import is_patchpanel, pair_ports, port_identifier
 
 
 def port(pk, name):
@@ -15,8 +15,8 @@ class PatchpanelPairingTest(SimpleTestCase):
         self.assertTrue(is_patchpanel(SimpleNamespace(role=SimpleNamespace(name="PATCHPANEL"))))
         self.assertFalse(is_patchpanel(SimpleNamespace(role=SimpleNamespace(name="Switch"))))
 
-    def test_pairs_front_and_rear_ports_in_natural_order(self):
-        front_ports = [port(10, "10"), port(2, "2"), port(1, "1")]
+    def test_pairs_front_and_rear_ports_by_matching_number(self):
+        front_ports = [port(10, "Front 10"), port(2, "Front 2"), port(1, "Front 1")]
         rear_ports = [port(102, "Rear 2"), port(110, "Rear 10"), port(101, "Rear 1")]
 
         pairs = pair_ports(front_ports, rear_ports)
@@ -24,6 +24,16 @@ class PatchpanelPairingTest(SimpleTestCase):
         self.assertEqual([(front.pk, rear.pk) for front, rear in pairs], [(1, 101), (2, 102), (10, 110)])
 
     def test_leaves_unmatched_ports_without_a_mapping(self):
-        pairs = pair_ports([port(1, "1"), port(2, "2")], [port(101, "1")])
+        pairs = pair_ports([port(1, "Front 1"), port(2, "Front 2")], [port(102, "Rear 2")])
 
-        self.assertEqual([(front.pk, rear.pk) for front, rear in pairs], [(1, 101)])
+        self.assertEqual([(front.pk, rear.pk) for front, rear in pairs], [(2, 102)])
+
+    def test_normalizes_common_side_prefixes_and_leading_zeroes(self):
+        self.assertEqual(port_identifier(port(1, "F01")), (1,))
+        self.assertEqual(port_identifier(port(2, "Rear Port 1")), (1,))
+        self.assertEqual(port_identifier(port(3, "FrontPort01")), (1,))
+
+    def test_does_not_pair_different_identifiers(self):
+        pairs = pair_ports([port(1, "Front 1")], [port(102, "Rear 2")])
+
+        self.assertEqual(pairs, [])