Add db gripcheck command, db status command

This commit is contained in:
Zeva Rose 2022-09-01 18:54:43 -06:00
parent fa5f8ae883
commit b1c1f60944

View file

@ -4,6 +4,7 @@ import re
from datetime import datetime, timedelta
import aiohttp
from bs4 import BeautifulSoup
from naff import Client, Extension, InteractionContext
from naff.models.discord.embed import EmbedField
from naff.models.naff.application_commands import (
@ -22,6 +23,50 @@ from jarvis.utils import build_embed
guild_ids = [578757004059738142, 520021794380447745, 862402786116763668]
async def parse_db_status() -> dict:
"""Parse the dbrand status page for a local API"""
async with aiohttp.ClientSession() as session:
async with session.get("https://dbrand.com/status") as response:
response.raise_for_status()
soup = BeautifulSoup(await response.content.read(), features="html.parser")
tables = soup.find_all("table")
data = {}
for table in tables:
data_key = "countries"
rows = table.find_all("tr")
headers = rows.pop(0)
headers = [h.get_text() for h in headers.find_all("th")]
if headers[0] == "Service":
data_key = "operations"
data[data_key] = []
for row in rows:
row_data = []
cells = row.find_all("td")
for cell in cells:
if "column--comment" in cell["class"]:
text = cell.find("span").get_text()
if cell != "Unavailable":
text += ": " + cell.find("div").get_text()
cell = text.strip()
elif "column--status" in cell["class"]:
info = cell.find("span")["class"]
if any("green" in x for x in info):
cell = "🟢"
elif any("yellow" in x for x in info):
cell = "🟡"
elif any("red" in x for x in info):
cell = "🔴"
elif any("black" in x for x in info):
cell = ""
else:
cell = cell.get_text().strip()
row_data.append(cell)
data[data_key].append(
{headers[idx]: value for idx, value in enumerate(row_data)}
)
return data
class DbrandCog(Extension):
"""
dbrand functions for JARVIS
@ -43,6 +88,38 @@ class DbrandCog(Extension):
db = SlashCommand(name="db", description="dbrand commands", scopes=guild_ids)
@db.subcommand(sub_cmd_name="status", sub_cmd_description="Get dbrand operational status")
async def _status(self, ctx: InteractionContext) -> None:
status = self.cache.get("status")
if not status or status["cache_expiry"] <= datetime.utcnow():
status = await parse_db_status()
status["cache_expiry"] = datetime.utcnow() + timedelta(hours=2)
self.cache["status"] = status
status = status.get("operations")
fields = [
EmbedField(name=f'{x["Status"]} {x["Service"]}', value=x["Detail"]) for x in status
]
embed = build_embed(
title="Operational Status",
description="Current dbrand operational status.\n[View online](https://dbrand.com/status)",
fields=fields,
url="https://dbrand.com/status",
color="#FFBB00",
)
embed.set_thumbnail(url="https://dev.zevaryx.com/db_logo.png")
embed.set_footer(
text="dbrand.com",
icon_url="https://dev.zevaryx.com/db_logo.png",
)
await ctx.send(embeds=embed)
@db.subcommand(sub_cmd_name="gripcheck", sub_cmd_description="Watch a dbrand grip get thrown")
async def _gripcheck(self, ctx: InteractionContext) -> None:
video_url = "https://cdn.discordapp.com/attachments/599068193339736096/890679742263623751/video0.mov"
image_url = "https://cdn.discordapp.com/attachments/599068193339736096/890680198306095104/image0.jpg"
await ctx.send(f"Video: {video_url}\nResults: {image_url}")
@db.subcommand(sub_cmd_name="info", sub_cmd_description="Get useful links")
@cooldown(bucket=Buckets.USER, rate=1, interval=30)
async def _info(self, ctx: InteractionContext) -> None: