From b61f196d0b868ea53759363ef3452f622450a8b3 Mon Sep 17 00:00:00 2001 From: g0ldyy <153996346+g0ldyy@users.noreply.github.com> Date: Fri, 26 Dec 2025 02:13:46 +0100 Subject: [PATCH] feat: introduce a Python script for Comet instance uptime monitoring and Discord notifications --- scripts/uptime_monitor.py | 235 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 235 insertions(+) create mode 100644 scripts/uptime_monitor.py diff --git a/scripts/uptime_monitor.py b/scripts/uptime_monitor.py new file mode 100644 index 0000000..c3f76f3 --- /dev/null +++ b/scripts/uptime_monitor.py @@ -0,0 +1,235 @@ +import asyncio +import sys +from datetime import datetime, timezone + +import aiohttp + +# ============== CONFIGURATION ============== + +INSTANCES = [ + "https://your-instance-1.example.com", + "https://your-instance-2.example.com", +] + +IMDB_ID = "tt30472557" +CHECK_INTERVAL = 300 +TIMEOUT = 30 + +WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_WEBHOOK_TOKEN" +WEBHOOK_USERNAME = "Comet Monitor" +WEBHOOK_AVATAR = "https://i.imgur.com/jmVoVMu.jpeg" + +# =========================================== + + +C_RESET = "\033[0m" +C_DIM = "\033[2m" +C_BOLD = "\033[1m" +C_GREEN = "\033[38;5;114m" +C_YELLOW = "\033[38;5;221m" +C_RED = "\033[38;5;203m" +C_CYAN = "\033[38;5;117m" +C_GRAY = "\033[38;5;245m" + + +class InstanceStatus: + def __init__( + self, + url: str, + is_online: bool, + manifest_ok: bool, + search_ok: bool, + response_time: float, + error: str | None, + ): + self.url = url + self.is_online = is_online + self.manifest_ok = manifest_ok + self.search_ok = search_ok + self.response_time = response_time + self.error = error + + +async def check_instance(session: aiohttp.ClientSession, url: str) -> InstanceStatus: + start = asyncio.get_event_loop().time() + manifest_ok = False + search_ok = False + error = None + + try: + async with session.get(f"{url}/manifest.json") as resp: + if resp.status == 200: + data = await resp.json() + manifest_ok = "id" in data and "resources" in data + except Exception as e: + error = str(e)[:60] + + if manifest_ok: + try: + async with session.get(f"{url}/e30=/stream/movie/{IMDB_ID}.json") as resp: + if resp.status == 200: + data = await resp.json() + streams = data.get("streams", []) + search_ok = len(streams) > 0 or resp.status == 200 + except Exception as e: + if not error: + error = str(e)[:60] + + response_time = asyncio.get_event_loop().time() - start + + return InstanceStatus( + url=url, + is_online=manifest_ok, + manifest_ok=manifest_ok, + search_ok=search_ok, + response_time=response_time, + error=error, + ) + + +def build_embed(status: InstanceStatus) -> dict: + instance_name = status.url.replace("https://", "").replace("http://", "") + timestamp = datetime.now(timezone.utc).isoformat() + + if status.is_online and status.search_ok: + color = 0x43B581 + description = "● Instance is healthy and responding" + elif status.is_online: + color = 0xFAA61A + description = "◐ Instance is online but search may be slow" + else: + color = 0xF04747 + description = "○ Instance is unreachable" + + manifest_value = "✓ Valid" if status.manifest_ok else "✗ Failed" + search_value = "✓ Working" if status.search_ok else "✗ Failed" + + fields = [ + { + "name": "Response Time", + "value": f"```\n{status.response_time:.2f}s\n```", + "inline": True, + }, + { + "name": "Manifest", + "value": f"```\n{manifest_value}\n```", + "inline": True, + }, + { + "name": "Search API", + "value": f"```\n{search_value}\n```", + "inline": True, + }, + ] + + if status.error: + fields.append( + { + "name": "Error", + "value": f"```\n{status.error}\n```", + "inline": False, + } + ) + + return { + "embeds": [ + { + "title": f"☄️ {instance_name}", + "url": status.url, + "description": description, + "color": color, + "fields": fields, + "footer": { + "icon_url": WEBHOOK_AVATAR, + }, + "timestamp": timestamp, + } + ], + "username": WEBHOOK_USERNAME, + "avatar_url": WEBHOOK_AVATAR, + } + + +async def send_webhook(session: aiohttp.ClientSession, payload: dict) -> bool: + try: + async with session.post(WEBHOOK_URL, json=payload) as resp: + return resp.status in (200, 204) + except Exception: + return False + + +def print_status(status: InstanceStatus) -> None: + timestamp = datetime.now().strftime("%H:%M:%S") + instance_name = status.url.replace("https://", "").replace("http://", "") + + if status.is_online and status.search_ok: + indicator = f"{C_GREEN}●{C_RESET}" + state = "operational" + state_color = C_GREEN + elif status.is_online: + indicator = f"{C_YELLOW}◐{C_RESET}" + state = "degraded" + state_color = C_YELLOW + else: + indicator = f"{C_RED}○{C_RESET}" + state = "offline" + state_color = C_RED + + latency = f"{status.response_time:>6.2f}s" + + manifest_check = ( + f"{C_GREEN}✓{C_RESET}" if status.manifest_ok else f"{C_RED}✗{C_RESET}" + ) + search_check = f"{C_GREEN}✓{C_RESET}" if status.search_ok else f"{C_RED}✗{C_RESET}" + + print( + f" {C_DIM}{timestamp}{C_RESET} {indicator} {C_BOLD}{instance_name:<40}{C_RESET} " + f"{state_color}{state:<11}{C_RESET} {C_CYAN}{latency}{C_RESET} {C_GRAY}[{manifest_check}{C_GRAY}/{search_check}{C_GRAY}]{C_RESET}" + ) + + if status.error: + print(f" {C_DIM}└─ {status.error}{C_RESET}") + + +async def monitor_instance(session: aiohttp.ClientSession, url: str) -> None: + while True: + status = await check_instance(session, url) + print_status(status) + + payload = build_embed(status) + await send_webhook(session, payload) + + await asyncio.sleep(CHECK_INTERVAL) + + +def print_header() -> None: + print() + print(f" {C_BOLD}☄️ Comet Uptime Monitor{C_RESET}") + print(f" {C_DIM}{'─' * 50}{C_RESET}") + print( + f" {C_GRAY}Instances: {C_RESET}{len(INSTANCES)} {C_GRAY}│ Interval: {C_RESET}{CHECK_INTERVAL}s" + ) + print() + for inst in INSTANCES: + name = inst.replace("https://", "").replace("http://", "") + print(f" {C_DIM}•{C_RESET} {name}") + print() + print(f" {C_DIM}{'─' * 50}{C_RESET}") + print() + + +async def main() -> None: + print_header() + + timeout = aiohttp.ClientTimeout(total=TIMEOUT) + async with aiohttp.ClientSession(timeout=timeout) as session: + tasks = [monitor_instance(session, url) for url in INSTANCES] + await asyncio.gather(*tasks) + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print(f"\n {C_GREEN}✓{C_RESET} Stopped\n") + sys.exit(0)