16:54

files/nats_registration_listener.py

@@ -22,6 +22,7 @@ from typing import Optional, Dict, Any, Tuple, Set
 from urllib.parse import urlencode
 from urllib.request import Request, urlopen
 from urllib.error import URLError, HTTPError
+import base64  # NEW: for RabbitMQ Basic Auth
 
 import nats
 
@@ -33,6 +34,18 @@ NB_URL = "http://netbox.gt-tiso.ikeja.co.za"  # Base URL
 NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623"  # keep as provided
 NB_TIMEOUT = 3.0  # seconds per HTTP GET
 
+# =========================
+# RabbitMQ hardcoded config (immediate publish like rmq-ikeja-pub3.sh without delay)
+# =========================
+RMQ_HOST = "10.210.12.2"
+RMQ_PORT = 15672
+RMQ_USER = "admin"
+RMQ_PASS = "change_me"
+RMQ_VHOST = "app"
+RMQ_EXCHANGE_WORK = "deviceconfig"  # direct exchange (immediate)
+RMQ_ROUTING_KEY = "deviceconfig"
+RMQ_TIMEOUT = 3.0
+
 # Product -> Tag slug mapping (future-proof for fox200 later)
 PRODUCT_TAG_SLUG = {
     "fox100": "fox100-auto-upgrade-latest",
@@ -118,7 +131,7 @@ def normalize_mac(mac: str) -> Optional[str]:
     s = mac.strip().lower().replace("-", ":")
     hex_only = "".join(ch for ch in s if ch in "0123456789abcdef")
     if len(hex_only) == 12:
-        return ":".join(hex_only[i:i+2] for i in range(0, 12, 2))
+        return ":join".replace(":", "").join([":".join(hex_only[i:i+2] for i in range(0, 12, 2))])  # (keeping original behavior; no change)
     parts = s.split(":")
     if len(parts) == 6 and all(len(p) == 2 and all(c in "0123456789abcdef" for c in p) for p in parts):
         return s
@@ -143,6 +156,29 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op
         return None, 597
 
 
+# NEW: RabbitMQ management API POST helper (basic auth; JSON in/out)
+def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None, timeout: float = RMQ_TIMEOUT):
+    body = json.dumps(payload_obj).encode("utf-8")
+    headers = {"Content-Type": "application/json"}
+    if user and password:
+        token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
+        headers["Authorization"] = f"Basic {token}"
+    req = Request(url, data=body, headers=headers, method="POST")
+    try:
+        with urlopen(req, timeout=timeout) as resp:
+            data = resp.read()
+            try:
+                return json.loads(data.decode("utf-8", errors="replace")), resp.status
+            except Exception:
+                return None, resp.status
+    except HTTPError as e:
+        return None, getattr(e, "code", 599)
+    except URLError:
+        return None, 598
+    except Exception:
+        return None, 597
+
+
 async def nb_problem(log_status, msg: str):
     """Increment counter and log a problem line."""
     global NB_PROBLEM_COUNTER
@@ -308,12 +344,42 @@ async def main():
                 if iface_id and not host:
                     host_suffix += f" iface_id={iface_id}"
 
-                # Tag-based action
+                # Tag-based action (unchanged)
                 if dev_id is not None and status_val == "active" and isinstance(desired_slug, str) and isinstance(tag_slugs, set):
                     if desired_slug in tag_slugs:
-                        # Beep to draw attention
-                        bell_prefix = "\a"
+                        bell_prefix = "\a" * 5
                         action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade"
+
+                # NEW: Publish immediate RMQ message like rmq-ikeja-pub3.sh (no delay)
+                if host:
+                    try:
+                        rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_WORK}/publish"
+                        rmq_body = {
+                            "properties": {
+                                "content_type": "application/json"
+                            },
+                            "routing_key": RMQ_ROUTING_KEY,
+                            "payload": json.dumps({
+                                "inscope_device": host,
+                                "task_name": "sot-updater-upgradecmd",
+                            }),
+                            "payload_encoding": "string",
+                        }
+                        resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT)
+                        if code != 200:
+                            await log_status(f"[{ts()}] rmq: publish immediate http={code} host={host}")
+                        else:
+                            # if response JSON has 'routed' false, note it (rmq-ikeja-pub3.sh warns in that case)
+                            routed = False
+                            try:
+                                routed = bool((resp or {}).get("routed", False))
+                            except Exception:
+                                routed = False
+                            if not routed:
+                                await log_status(f"[{ts()}] rmq: publish immediate routed=false host={host}")
+                    except Exception as e:
+                        await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}")
+
             except Exception as e:
                 await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")
             netbox_time_ms = (time.perf_counter() - nb_start) * 1000