Jlk fix oui

.github/workflows/oui_filepull.yml

@@ -45,6 +45,16 @@ jobs:
       - name: "Generate oui_mappings.py"
         run: "python ./flat_postprocess/oui_postprocess.py ./netutils/data_files/oui_mappings.py"
 
+      - name: "Setup environment"
+        uses: "networktocode/gh-action-setup-poetry-environment@v7"
+        with:
+          poetry-version: "2.1.3"
+
+      - name: "Linting: ruff format"
+        env:
+          INVOKE_NETUTILS_LOCAL: "True"
+        run: "poetry run invoke a"
+
       - name: "Commit changes"
         run: |
           git config user.name "github-actions[bot]"

changes/oui-automation.housekeeping

@@ -0,0 +1 @@
+Fix OUI automation and CI workflow to auto format.

flat_postprocess/oui_postprocess.py

@@ -1,19 +1,35 @@
 """Python code used to postprocess Flat github action data related to OUI mappings."""
 
+import json
 import re
 import subprocess
 import sys
 
-HEX_RE = r"^MA-L,(?P<hex>[0-9A-Fa-f]{6}),\"(?P<company>[^\"]+)\",.*$"
+HEX_RE = re.compile(
+    r"^[^,]*,(?P<hex>[0-9A-Fa-f]{6})," + r'(?:"(?P<company_q>(?:[^"]|"")*)"|(?P<company_u>[^,]*))(?:,|$)'
+)
 
 OUI_MAPPINGS = {}
-URL = "https://standards-oui.ieee.org/oui/oui.txt"
+URL = "https://standards-oui.ieee.org/oui/oui.csv"
 
 
 def download_csv_text(url: str = URL) -> str:
     """Download the CSV text from the given URL."""
     proc = subprocess.run(  # noqa: S603
-        ["curl", "-fsSL", url],  # noqa: S607
+        [  # noqa: S607
+            "curl",
+            "-fsSL",
+            "--retry",
+            "5",
+            "--retry-all-errors",
+            "--retry-max-time",
+            "300",
+            "--connect-timeout",
+            "15",
+            "--max-time",
+            "120",
+            url,
+        ],
         check=True,
         capture_output=True,
         text=True,
@@ -23,28 +39,33 @@ def download_csv_text(url: str = URL) -> str:
 
 if __name__ == "__main__":
     if len(sys.argv) < 2:
-        raise SystemExit("Usage: python oui_postprocess.py <output_file> [<download_latest>]")
+        raise SystemExit("Usage: python oui_postprocess.py <output_file> [--stdin]")
 
     output_path = sys.argv[1]
-    download = "--download" in sys.argv[2:]
 
-    if download:
+    if "--stdin" in sys.argv[2:]:
+        csv_text = sys.stdin.read()
+    else:
         csv_text = download_csv_text(URL)
-        with open(output_path, "w", encoding="utf-8", newline="") as oui_textfile:
-            oui_textfile.write(csv_text)
 
-    with open(output_path, "r", encoding="utf-8", newline="") as oui_file:
-        for line in oui_file:
-            if re.search(HEX_RE, line):
-                group_regex_values = re.search(HEX_RE, line).groupdict()
-                if group_regex_values.get("hex") and group_regex_values.get("company"):
-                    OUI_MAPPINGS.update({group_regex_values.get("hex").lower(): group_regex_values.get("company")})
+    for line in csv_text.splitlines():
+        match = HEX_RE.match(line)
+        if not match:
+            continue
+        company = match.group("company_q") or match.group("company_u") or ""
+        company = company.replace('""', '"').strip()
+        hex_value = match.group("hex")
+        if hex_value and company:
+            OUI_MAPPINGS[hex_value.lower()] = company
+
+    if not OUI_MAPPINGS:
+        raise SystemExit("No OUI records parsed; refusing to write empty mappings.")
 
     with open(output_path, "w", encoding="utf-8") as oui_mappings:
         oui_mappings.write('"""Dictionary object to store OUI information."""\n')
         oui_mappings.write("# pylint: disable=too-many-lines\n")
         oui_mappings.write("import typing\n\n")
         oui_mappings.write("OUI_MAPPINGS: typing.Dict[str, str] = {\n")
         for mac, company in sorted(OUI_MAPPINGS.items()):
-            oui_mappings.write(f'    "{mac}": "{company}",\n')
+            oui_mappings.write(f"    {json.dumps(mac)}: {json.dumps(company)},\n")
         oui_mappings.write("}\n")