112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
import subprocess
|
|
import asyncio
|
|
|
|
import lightbulb
|
|
import hikari
|
|
|
|
from lib.config import load_config
|
|
|
|
config = load_config()
|
|
|
|
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
|
|
)
|
|
|
|
stdout, _ = await id_proc.communicate()
|
|
output["commit_id"] = stdout.decode("utf-8")
|
|
|
|
br_proc = await asyncio.create_subprocess_exec("git", "rev-parse", "--abbrev-ref", "HEAD",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
|
|
stdout, _ = await br_proc.communicate()
|
|
output["branch"] = stdout.decode("utf-8").strip()
|
|
|
|
rm_proc = await asyncio.create_subprocess_exec("git", "branch", "-r",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
|
|
stdout, _ = await rm_proc.communicate()
|
|
output['remote'] = "remotes/"+stdout.decode("utf-8").lstrip().split("\n")[0]
|
|
|
|
# diff HEAD to remote
|
|
hd_proc = await asyncio.create_subprocess_exec("git", "diff", "HEAD", output['remote'],
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
|
|
# diff index to remote
|
|
ix_proc = await asyncio.create_subprocess_exec("git", "diff", output['remote'],
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, _ = await ix_proc.communicate()
|
|
ix_diff = stdout.decode("utf-8")
|
|
|
|
# commits ahead/behind remote
|
|
ab_proc = await asyncio.create_subprocess_exec("git", "rev-list", "--left-right", "--count", output['branch']+"..."+output["remote"],
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, _ = await ab_proc.communicate()
|
|
commits_ahead = stdout.decode("utf-8").split()[0]
|
|
commits_behind = stdout.decode("utf-8").split()[1]
|
|
|
|
stdout, _ = await hd_proc.communicate()
|
|
hd_diff = stdout.decode("utf-8")
|
|
output['dev'] = False
|
|
|
|
# dev environment tests
|
|
if hd_diff != "": # difference between HEAD and remote
|
|
output['dev'] = True
|
|
elif config.branch != output["branch"]: # branch has been changed
|
|
output['dev'] = True
|
|
elif ix_diff != "": # difference between index and remote
|
|
output['dev'] = True
|
|
elif int(commits_ahead) > 0: # ahead in commits
|
|
output['dev'] = True
|
|
|
|
output['update'] = False
|
|
if not output['dev'] and int(commits_behind) > 0:
|
|
output['update'] = 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("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("In Dev-Env?", git_status['dev'])
|
|
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)
|