#!/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.command(name="ip") @root_required async def get_ip(ctx): r = requests.get("https://ifconfig.me") await ctx.reply("Current IP: `"+ r.content.decode("utf-8")+"`") @bot.command(name="reboot") @root_required async def reboot(ctx): await ctx.message.add_reaction("🆗") os.system("reboot") bot.run(config.TOKEN)