radical-bot/ext/system.py

import asyncio

import lightbulb
import hikari

from lib.config import load_config

config = load_config()

plugin = lightbulb.Plugin("SystemPlugin")


async def create_subprocess(*args):
    proc = await asyncio.create_subprocess_exec(*args,
                                                stdout=asyncio.subprocess.PIPE,
                                                stderr=asyncio.subprocess.PIPE
                                                )
    stdout, _ = await proc.communicate()
    return stdout.decode("utf-8").strip()


async def get_git_latest_commit_id():
    return await create_subprocess("git", "rev-parse", "--short", "HEAD")


async def get_git_branch():
    return await create_subprocess("git", "rev-parse", "--abbrev-ref", "HEAD")


async def get_git_remote():
    remote = await create_subprocess("git", "branch", "-vv")
    for line in remote.split("\n"):
        if "*" in line:
            return line.split("[")[1].split("]")[0]


async def get_git_head_diff_branch(branch: str) -> str:
    # diff HEAD to remote
    return await create_subprocess("git", "diff", "head", branch)


async def get_git_index_diff_branch(branch: str) -> str:
    # diff index to remote
    return await create_subprocess("git", "diff", branch)


async def get_git_commits_ahead_behind(branch: str, remote: str) -> tuple[int, int]:
    # commits ahead/behind remote
    output = await create_subprocess("git", "rev-list", "--left-right", "--count", branch+"..."+remote)
    commits_ahead = int(output.split()[0])
    commits_behind = int(output.split()[1])
    return commits_ahead, commits_behind


async def get_git_update(branch, remote) -> bool:
    commits_behind, _ = await get_git_commits_ahead_behind(branch, remote)
    if commits_behind > 0:
        return True
    return False


async def get_git_status() -> dict:
    output = {}
    output["commit_id"] = await get_git_latest_commit_id()
    output["branch"] = await get_git_branch()
    output["remote"] = await get_git_remote()
    output['update'] = await get_git_update(output['branch'], output['remote'])
    return output


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


@plugin.command
@lightbulb.add_checks(lightbulb.owner_only)
@lightbulb.command("update", "update the bot!", ephemeral=True)
@lightbulb.implements(lightbulb.SlashCommand)
async def update(ctx: lightbulb.Context) -> None:
    await create_subprocess("git", "pull")
    embed = hikari.Embed(title="Restarting",
                         description="Restarting to load an update!"
                         )
    await ctx.respond(embed)
    await ctx.bot.close()


@plugin.command
@lightbulb.add_checks(lightbulb.owner_only)
@lightbulb.command("branch", "get or set the working branch the bot uses.")
@lightbulb.implements(lightbulb.SlashCommandGroup)
async def branch(ctx: lightbulb.Context) -> None:
    pass  # SlashCommandGroup body code isn't run


@branch.child
@lightbulb.add_checks(lightbulb.owner_only)
@lightbulb.command("get", "get the current branch", ephemeral=True)
@lightbulb.implements(lightbulb.SlashSubCommand)
async def branch_get(ctx: lightbulb.Context) -> None:
    output = await create_subprocess("git", "rev-parse", "--symbolic-full-name", "--abbrev-ref", "HEAD")
    embed = hikari.Embed(title="Current Branch",
                         description=f"Currently on branch '{output}'")
    await ctx.respond(embed=embed)


@branch.child
@lightbulb.add_checks(lightbulb.owner_only)
@lightbulb.option("name", "name of the branch", type=str, required=True)
@lightbulb.command("switch", "switch branches", ephemeral=True)
@lightbulb.implements(lightbulb.SlashSubCommand)
async def branch_switch(ctx: lightbulb.Context) -> None:
    embed = hikari.Embed(title="Restarting",
                         description="Restarting to switch branches!"
                         )
    output = await create_subprocess("git", "switch", ctx.options.branch)
    if "invalid reference" in output:  # parsing output allows us more specificity.
        embed.title = "Branch does not exist."
        embed.description = "Please check your spelling."
        await ctx.respond(embed)
        return
    elif "Already on" in output:
        embed.title = f"Already on branch '{ctx.options.branch}'"
        embed.description = "Just double checking, I presume?"
        await ctx.respond(embed)
        return
    await ctx.respond(embed)
    await ctx.bot.close()


@plugin.command
@lightbulb.command("info", "get bot information such as the version, and repository link.", 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("Git Repository", config.git_url)
    embed.add_field("Version", git_status['commit_id'], inline=True)
    embed.add_field("Branch", git_status['branch'], inline=True)
    embed.add_field("Remote", git_status['remote'], inline=True)
    embed.add_field("Needs Update?", git_status['update'])
    await ctx.respond(embed)


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


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