radical-bot/ext/system.py

import subprocess
import asyncio

import lightbulb
import hikari

plugin = lightbulb.Plugin("SystemPlugin")

async def get_git_status() -> dict:
    output = {}
    try:
        id_proc = await asyncio.create_subprocess_exec("git", "rev-parse", "--short", "HEAD",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        br_proc = await asyncio.create_subprocess_exec("git", "rev-parse", "--abbrev-ref", "HEAD",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        dv_proc = await asyncio.create_subprocess_exec("git", "diff", "HEAD",
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        stdout, _ = await id_proc.communicate()
        output["commit_id"] = stdout.decode("utf-8")

        stdout, _ = await br_proc.communicate()
        output["branch"] = stdout.decode("utf-8")

        stdout, _ = await dv_proc.communicate()
        if stdout.decode("utf-8") == "":
            output["dev"] = False
        else:
            output["dev"] = True
    except subprocess.SubprocessError as err:
        print(err)
    return output

@plugin.command
@lightbulb.command("ping", "ping the bot.", ephemeral=True)
@lightbulb.implements(lightbulb.SlashCommand)
async def ping(ctx: lightbulb.Context) -> None:
    embed = hikari.Embed(title="Ping!",
        description=f"latency: {round(ctx.bot.heartbeat_latency * 1000, ndigits=2)}ms"
    )
    await ctx.respond(embed)

@plugin.command
@lightbulb.command("info", "get bot info", ephemeral=True)
@lightbulb.implements(lightbulb.SlashCommand)
async def info(ctx: lightbulb.Context) -> None:
    git_status = await get_git_status()
    embed = hikari.Embed(title="About Me!")
    embed.add_field("GitHub", "https://gitdab.com/InValidFire/radical-bot")
    embed.add_field("Version", git_status['commit_id'], inline=True)
    embed.add_field("Branch", git_status['branch'], inline=True)
    embed.add_field("In Dev-Env?", git_status['dev'])
    await ctx.respond(embed)

def load(bot: lightbulb.BotApp):
    bot.add_plugin(plugin)

def unload(bot: lightbulb.BotApp):
    bot.remove_plugin(plugin)