Compare commits

...

1 Commits

Author SHA1 Message Date
Metatron
a5dabb2a03 feat: portal health check — auto-disable broken portals (closes #1539)
Background health check that:
- Checks each portal's destination URL
- Unreachable portal → status set to 'offline' (dim + tooltip)
- Auto-re-enable when reachable again
- Updates portals.json in-place

Usage:
  python3 scripts/portal-health-check.py              # check + update
  python3 scripts/portal-health-check.py --dry-run     # check only
  python3 scripts/portal-health-check.py --json        # JSON output

Handles 6 URL-based portals from portals.json.
2026-04-16 00:38:59 -04:00

143
scripts/portal-health-check.py Executable file
View File

@@ -0,0 +1,143 @@
#!/usr/bin/env python3
"""
Portal Health Check — Auto-disable broken portals in the Nexus.
Checks each portal's destination URL and updates status:
- online: URL reachable
- offline: URL unreachable → dim portal + tooltip "Offline"
- online again: auto-re-enable
Usage:
python3 scripts/portal-health-check.py # check and update
python3 scripts/portal-health-check.py --dry-run # check only
python3 scripts/portal-health-check.py --json # JSON output
Ref: #1539
"""
import json
import os
import sys
import urllib.request
from datetime import datetime, timezone
from typing import Any, Dict, List
PORTALS_FILE = os.environ.get("PORTALS_FILE", "portals.json")
CHECK_TIMEOUT = int(os.environ.get("PORTAL_CHECK_TIMEOUT", "5"))
def load_portals(path: str = PORTALS_FILE) -> List[dict]:
with open(path) as f:
return json.load(f)
def save_portals(portals: List[dict], path: str = PORTALS_FILE):
with open(path, "w") as f:
json.dump(portals, f, indent=2, ensure_ascii=False)
def check_portal_url(url: str, timeout: int = CHECK_TIMEOUT) -> dict:
"""Check if a portal URL is reachable."""
if not url or url.startswith("./") or url.startswith("../"):
return {"reachable": True, "status": "local", "latency_ms": 0}
try:
import time
start = time.time()
req = urllib.request.Request(url, method="HEAD")
with urllib.request.urlopen(req, timeout=timeout) as resp:
latency = int((time.time() - start) * 1000)
return {"reachable": True, "status": resp.status, "latency_ms": latency}
except urllib.error.HTTPError as e:
# 4xx/5xx means the server responded — portal is reachable
return {"reachable": True, "status": e.code, "latency_ms": 0}
except Exception as e:
return {"reachable": False, "status": "unreachable", "error": str(e)[:100]}
def check_all_portals(dry_run: bool = False) -> List[dict]:
"""Check all portals and update status."""
portals = load_portals()
results = []
changes = 0
for portal in portals:
pid = portal["id"]
dest = portal.get("destination") or {}
url = dest.get("url")
old_status = portal.get("status", "unknown")
# Check health
health = check_portal_url(url) if url else {"reachable": True, "status": "no_url"}
is_reachable = health.get("reachable", True)
# Determine new status
if old_status == "offline" and is_reachable:
new_status = "online"
action = "RE-ENABLED"
changes += 1
elif old_status == "online" and not is_reachable:
new_status = "offline"
action = "DISABLED"
changes += 1
elif old_status == "offline" and not is_reachable:
new_status = "offline"
action = "still offline"
else:
new_status = old_status
action = "ok"
if not dry_run and new_status != old_status:
portal["status"] = new_status
results.append({
"id": pid,
"name": portal.get("name", pid),
"old_status": old_status,
"new_status": new_status,
"action": action,
"reachable": is_reachable,
"latency_ms": health.get("latency_ms", 0),
"url": url or "local",
})
if not dry_run and changes > 0:
save_portals(portals)
return results, changes
def print_report(results: List[dict], changes: int):
print(f"\n{'='*70}")
print(f" PORTAL HEALTH CHECK")
print(f" {datetime.now().isoformat()}")
print(f"{'='*70}\n")
print(f" {'Portal':20} {'Status':10} {'Action':15} {'Latency':10} {'URL'}")
print(f" {'-'*20} {'-'*10} {'-'*15} {'-'*10} {'-'*30}")
for r in results:
latency = f"{r['latency_ms']}ms" if r['latency_ms'] else ""
url = (r["url"] or "local")[:30]
print(f" {r['name']:20} {r['new_status']:10} {r['action']:15} {latency:10} {url}")
print(f"\n Changes: {changes} portal(s) updated")
def main():
import argparse
parser = argparse.ArgumentParser(description="Portal Health Check")
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
results, changes = check_all_portals(dry_run=args.dry_run)
if args.json:
print(json.dumps({"results": results, "changes": changes}, indent=2))
else:
print_report(results, changes)
if __name__ == "__main__":
main()