Add playitstatus cog - monitor playit.gg status

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))

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."
+}

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")