From 3f1701d741144eb1b4e4d619f783fe28dbbd02d1 Mon Sep 17 00:00:00 2001 From: Kingston Date: Fri, 10 Jul 2026 07:12:12 -0500 Subject: [PATCH] Add playitstatus cog - monitor playit.gg status --- playitstatus/__init__.py | 16 +++ playitstatus/info.json | 26 +++++ playitstatus/playitstatus.py | 191 +++++++++++++++++++++++++++++++++++ 3 files changed, 233 insertions(+) create mode 100644 playitstatus/__init__.py create mode 100644 playitstatus/info.json create mode 100644 playitstatus/playitstatus.py diff --git a/playitstatus/__init__.py b/playitstatus/__init__.py new file mode 100644 index 0000000..db735ab --- /dev/null +++ b/playitstatus/__init__.py @@ -0,0 +1,16 @@ +"""Package for PlayitStatus cog.""" + +import json +from pathlib import Path + +from redbot.core.bot import Red + +from .playitstatus import PlayitStatus + +with Path(__file__).parent.joinpath("info.json").open() as fp: + __red_end_user_data_statement__ = json.load(fp)["end_user_data_statement"] + + +async def setup(bot: Red) -> None: + """Load PlayitStatus cog.""" + await bot.add_cog(PlayitStatus(bot)) diff --git a/playitstatus/info.json b/playitstatus/info.json new file mode 100644 index 0000000..9bfaa27 --- /dev/null +++ b/playitstatus/info.json @@ -0,0 +1,26 @@ +{ + "name": "PlayitStatus", + "author": [ + "Scrapyard Cogworks" + ], + "short": "Monitor playit.gg status and alert on downtime.", + "description": "Fetches playit.gg status page every 10 seconds and posts alerts to a channel when services go down or come back up.", + "install_msg": "Thanks for installing PlayitStatus from Scrapyard Cogworks! Visit us at https://homepage.kingstons-scrapyard.net/", + "requirements": [ + "aiohttp" + ], + "tags": [ + "status", + "monitor", + "playit", + "uptime", + "alert" + ], + "min_bot_version": "3.5.0", + "min_python_version": [ + 3, + 11, + 0 + ], + "end_user_data_statement": "This cog does not persistently store data or metadata about users." +} diff --git a/playitstatus/playitstatus.py b/playitstatus/playitstatus.py new file mode 100644 index 0000000..40c6c63 --- /dev/null +++ b/playitstatus/playitstatus.py @@ -0,0 +1,191 @@ +import discord +from redbot.core import commands, Config +import aiohttp +import asyncio +from datetime import datetime + + +class PlayitStatus(commands.Cog): + """Monitor playit.gg status and alert when it goes down.""" + + def __init__(self, bot): + self.bot = bot + self.config = Config.get_conf(self, identifier=1234567890) + self.config.register_guild( + channel_id=None, + monitor_enabled=False, + last_status={}, + ) + self.session = None + self.task = None + self.api_url = "https://status.playit.gg/api/getMonitorList/AxwIw5PSiW" + self.feed_url = "https://status.playit.gg/api/getEventFeed/AxwIw5PSiW" + + async def red_delete_data_for_user(self, **kwargs): + return + + def cog_unload(self): + if self.task: + self.task.cancel() + + async def init_session(self): + if self.session is None or self.session.closed: + self.session = aiohttp.ClientSession() + + async def fetch_status(self): + await self.init_session() + try: + async with self.session.get(self.api_url) as resp: + if resp.status == 200: + data = await resp.json() + return data.get("data", []) + except Exception: + pass + return [] + + async def fetch_events(self): + await self.init_session() + try: + async with self.session.get(self.feed_url) as resp: + if resp.status == 200: + data = await resp.json() + return data.get("results", []) + except Exception: + pass + return [] + + async def monitor_loop(self): + while True: + try: + await asyncio.sleep(10) + guilds = await self.config.all_guilds() + for guild_id, guild_data in guilds.items(): + if not guild_data.get("monitor_enabled"): + continue + channel_id = guild_data.get("channel_id") + if not channel_id: + continue + channel = self.bot.get_channel(channel_id) + if not channel: + continue + monitors = await self.fetch_status() + last_status = guild_data.get("last_status", {}) + for monitor in monitors: + name = monitor.get("name", "Unknown") + status_class = monitor.get("statusClass", "unknown") + is_up = status_class == "success" + was_up = last_status.get(name, True) + if not is_up and was_up: + embed = discord.Embed( + title="Playit.gg Status Alert", + description=f"**{name}** is now **DOWN**", + color=discord.Color.red(), + timestamp=datetime.utcnow(), + ) + embed.add_field( + name="Status", + value=status_class, + inline=True, + ) + embed.set_footer(text="Playit.gg Status Monitor") + await channel.send(embed=embed) + elif is_up and not was_up: + embed = discord.Embed( + title="Playit.gg Status Update", + description=f"**{name}** is now **UP**", + color=discord.Color.green(), + timestamp=datetime.utcnow(), + ) + embed.set_footer(text="Playit.gg Status Monitor") + await channel.send(embed=embed) + last_status[name] = is_up + await self.config.guild( + self.bot.get_guild(guild_id) + ).last_status.set(last_status) + except asyncio.CancelledError: + break + except Exception: + await asyncio.sleep(5) + + @commands.group() + @commands.admin_or_permissions(manage_guild=True) + async def playitstatus(self, ctx): + """Playit.gg status monitor commands.""" + pass + + @playitstatus.command() + async def channel(self, ctx, channel: discord.TextChannel): + """Set the channel for status alerts.""" + await self.config.guild(ctx.guild).channel_id.set(channel.id) + await ctx.send(f"Status alerts will be sent to {channel.mention}") + + @playitstatus.command() + async def start(self, ctx): + """Start monitoring playit.gg status.""" + await self.config.guild(ctx.guild).monitor_enabled.set(True) + await self.config.guild(ctx.guild).last_status.set({}) + if not self.task or self.task.done(): + self.task = self.bot.loop.create_task(self.monitor_loop()) + await ctx.send("Started monitoring playit.gg status (checks every 10 seconds)") + + @playitstatus.command() + async def stop(self, ctx): + """Stop monitoring playit.gg status.""" + await self.config.guild(ctx.guild).monitor_enabled.set(False) + await ctx.send("Stopped monitoring playit.gg status") + + @playitstatus.command() + async def status(self, ctx): + """Check current playit.gg status.""" + monitors = await self.fetch_status() + if not monitors: + await ctx.send("Failed to fetch status") + return + embed = discord.Embed( + title="Playit.gg Current Status", + color=discord.Color.blue(), + timestamp=datetime.utcnow(), + ) + for monitor in monitors: + name = monitor.get("name", "Unknown") + status_class = monitor.get("statusClass", "unknown") + emoji = "🟢" if status_class == "success" else "🔴" + embed.add_field( + name=f"{emoji} {name}", + value=status_class.upper(), + inline=True, + ) + await ctx.send(embed=embed) + + @playitstatus.command() + async def events(self, ctx): + """Check recent status events.""" + events = await self.fetch_events() + if not events: + await ctx.send("No recent events") + return + embed = discord.Embed( + title="Playit.gg Recent Events", + color=discord.Color.orange(), + timestamp=datetime.utcnow(), + ) + for event in events[:5]: + monitor = event.get("monitor", "Unknown") + date = event.get("timeGMT", "Unknown") + duration = event.get("duration", "Unknown") + label = event.get("label", "Unknown") + embed.add_field( + name=monitor, + value=f"{label.upper()} - {date}\nDuration: {duration}", + inline=False, + ) + await ctx.send(embed=embed) + + @playitstatus.command() + async def test(self, ctx): + """Test if the API is reachable.""" + monitors = await self.fetch_status() + if monitors: + await ctx.send(f"API reachable. Found {len(monitors)} monitors.") + else: + await ctx.send("Failed to reach API")