status-bot/main.py

#!/usr/bin/python3
import asyncio
import json
import os
import subprocess
import time

import discord
import requests
from discord.ext import commands

import config


def login_required(fn):
    async def decorated_fn(ctx, *args):
        if str(ctx.author.id) not in config.allowed_users.keys():
            await ctx.reply("Vous n'êtes pas autorisé à exécuter cette commande.")
            return None
        return await fn(ctx, *args)

    return decorated_fn

def root_required(fn):
    async def decorated_fn(ctx, *args):
        if str(ctx.author.id) not in config.allowed_users.keys():
            await ctx.reply("Vous n'êtes pas autorisé à exécuter cette commande.")
            return None
        if config.allowed_users[str(ctx.author.id)] != "*":
            await ctx.reply("Vous n'êtes pas autorisé à exécuter cette commande.")
            return None
        return await fn(ctx, *args)

    return decorated_fn

def check_status(service):
    """Renvoie le statut d'un service"""
    service_status = os.system(f"systemctl is-active --quiet {service}")
    if service_status == 0:
        return "✅ Running"
    return "❌ Stopped"

def get_temperature():
    try:
        return json.loads(subprocess.check_output(["sensors", "-Aj"]))
    except subprocess.CalledProcessError as err:
        return {}

intents = discord.Intents.default()
intents.messages = True
intents.message_content = True

bot = commands.Bot(command_prefix=config.command_prefix, intents=intents)


@bot.event
async def on_ready():
    """On ready"""
    print("Connecté en tant que:")
    print(bot.user.name)
    print(bot.user.id)
    print("------")

    game = discord.Game(f"Disponible ✅ | {bot.command_prefix}help")
    await bot.change_presence(status=discord.Status.idle, activity=game)

    embed = discord.Embed(colour=discord.Colour.blue())
    embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
    embed.add_field(
        name="Redémarrage", value="Le bot est à nouveau en ligne"
    )
    temperature = get_temperature()
    for key1 in temperature.keys():
        for key2 in temperature[key1].keys():
            for key3 in temperature[key1][key2].keys():
                if "input" in key3:
                    embed.add_field(
                        name=f"Temp: {key1} > {key2} > {key3}", value=str(temperature[key1][key2][key3])
                    )

    embed.add_field(
        name="Uptime", value=subprocess.check_output(["uptime", "-p"]).decode("utf-8")
    )
    for discord_id in config.allowed_users.keys():
        if config.allowed_users[discord_id] == "*":
            user = await bot.fetch_user(discord_id)
            await user.send(embed=embed)

    services_status = {service:check_status(service) for service in config.services}
    while True:
        new_services_status = {service:check_status(service) for service in config.services}
        for service in new_services_status.keys():
            if new_services_status[service] != services_status[service]:

                embed = discord.Embed(colour=discord.Colour.blue())
                embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
                embed.add_field(
                    name=service, value=new_services_status[service], inline=True
                )
                for discord_id in config.allowed_users.keys():
                    if config.allowed_users[discord_id] == "*":
                        user = await bot.fetch_user(discord_id)
                        await user.send(embed=embed)

        services_status = new_services_status
        await asyncio.sleep(300)
        


@bot.command(name="status")
@login_required
async def status(ctx, *args):
    """Renvoie le statut des différents services"""
    if config.allowed_users[str(ctx.author.id)] == "*":
        user_services = config.services
    else:
        user_services = config.allowed_users[str(ctx.author.id)]

    if len(args) == 0:
        embed = discord.Embed(colour=discord.Colour.blue())
        embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
        for service in user_services:
            service_status = check_status(service)
            embed.add_field(
                name=service, value=service_status, inline=False
            )
        await ctx.reply(embed=embed)
    else:
        if args[0] not in user_services:
            await ctx.reply(f"Service inconnu ou inaccessible `{args[0]}`")
            return None
        try:
            out = subprocess.check_output(["systemctl", "status", args[0]], stderr=subprocess.STDOUT)
        except Exception as e:
            out = e.output

        out = out.decode("UTF-8")
        embed = discord.Embed(colour=discord.Colour.blue())
        embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
        embed.add_field(
                name=args[0], value=f"```\n{out[:min(len(out)-1, 1017)]}```"
            )
        await ctx.reply(embed=embed)


@bot.command(name="service")
@login_required
async def service(ctx, *args):
    """Exécute une commande systemd"""
    if config.allowed_users[str(ctx.author.id)] == "*":
        user_services = config.services
    else:
        user_services = config.allowed_users[str(ctx.author.id)]

    error = None
    commands = ["start", "stop", "enable", "disable", "restart"]
    if len(args) < 2:
        error = "Paramètres manquants."
    elif args[0] not in commands:
        error = f"Commande invalide `{args[0]}`"
    elif args[1] not in user_services:
        error= f"Service inconnu ou inaccessible `{args[1]}`"

    if error is not None:
        await ctx.reply(error)
        return None
    os.system(f"systemctl {args[0]} {args[1]}")
    await ctx.message.add_reaction("🆗")


@bot.command(name="logs")
@root_required
async def logs(ctx, *args):
    """Renvoie des fichier logs"""
    logs_files = [
        i[:-4] for i in os.listdir(config.logs_dir) if i.endswith(".txt") and os.path.isfile(os.path.join(config.logs_dir, i))
    ]

    if len(args) == 0:
        embed = discord.Embed(colour=discord.Colour.blue())
        embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
        embed.add_field(
            name="Fichiers de logs disponibles", value="\n".join(logs_files)
        )
        await ctx.reply(embed=embed)
        return None
    
    if args[0] not in logs_files:
        print(logs_files)
        await ctx.reply("Ce fichier n'existe pas.")
        return None

    with open(os.path.join(config.logs_dir, f"{args[0]}.txt"), "r") as fp:
        await ctx.reply(file=discord.File(fp, filename=f"{args[0]}.txt"))


@bot.command(name="temperature")
@root_required
async def temperature(ctx):
    embed = discord.Embed(colour=discord.Colour.blue())
    embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
    temperature = get_temperature()
    for key1 in temperature.keys():
        for key2 in temperature[key1].keys():
            for key3 in temperature[key1][key2].keys():
                if "input" in key3:
                    embed.add_field(
                        name=f"Temp: {key1} > {key2} > {key3}", value=str(temperature[key1][key2][key3])
                    )

    await ctx.reply(embed=embed)


@bot.command(name="uptime")
@login_required
async def uptime(ctx):
    embed = discord.Embed(colour=discord.Colour.blue())
    embed.set_author(name=time.strftime("%m-%d-%Y %H:%M"))
    embed.add_field(
        name="Uptime", value=subprocess.check_output(["uptime", "-p"]).decode("utf-8")
    )

    await ctx.reply(embed=embed)

bot.run(config.TOKEN)