diff --git a/cogs/ios/__init__.py b/cogs/ios/__init__.py new file mode 100644 index 000000000..f28142937 --- /dev/null +++ b/cogs/ios/__init__.py @@ -0,0 +1,7 @@ +from disnake.ext.commands import Bot + +from .cog import IOS + + +def setup(bot: Bot): + bot.add_cog(IOS(bot)) diff --git a/cogs/ios/cog.py b/cogs/ios/cog.py new file mode 100644 index 000000000..7cdec9a52 --- /dev/null +++ b/cogs/ios/cog.py @@ -0,0 +1,116 @@ +""" +Cog for the IOS subject. Get users on merlin/eva server which have blocking processes running. +""" + +import subprocess + +import disnake +from disnake.ext import commands, tasks + +from cogs.base import Base +from config import cooldowns +from permissions import permission_check + +from . import features +from .messages_cz import MessagesCZ + + +class IOS(Base, commands.Cog): + def __init__(self, bot: commands.Bot): + super().__init__() + self.bot = bot + self.tasks = [self.ios_task] + + @cooldowns.default_cooldown + @commands.check(permission_check.helper_plus) + @commands.slash_command(name="ios", description=MessagesCZ.ios_brief, guild_ids=[Base.config.guild_id]) + async def ios(self, inter: disnake.ApplicationCommandInteraction): + await inter.response.defer() + await self.ios_task(inter) + + @commands.slash_command(name="ios_task", guild_ids=[Base.config.guild_id]) + async def _ios(self, inter: disnake.ApplicationCommandInteraction): + pass + + @commands.check(permission_check.is_bot_admin) + @_ios.sub_command(name="start", description=MessagesCZ.task_start_brief) + async def ios_task_start(self, inter: disnake.ApplicationCommandInteraction): + try: + self.ios_task.start() + await inter.send(MessagesCZ.task_start_success) + except RuntimeError: + await inter.send(MessagesCZ.task_start_already_set) + + @commands.check(permission_check.is_bot_admin) + @_ios.sub_command(name="stop", description=MessagesCZ.task_stop_brief) + async def ios_task_stop(self, inter: disnake.ApplicationCommandInteraction): + if self.ios_task.is_running(): + self.ios_task.stop() + await inter.send(MessagesCZ.task_stop_success) + else: + await inter.send(MessagesCZ.task_nothing_to_stop) + + @commands.check(permission_check.is_bot_admin) + @_ios.sub_command(name="cancel", description=MessagesCZ.task_cancel_brief) + async def ios_task_cancel(self, inter: disnake.ApplicationCommandInteraction): + if self.ios_task.is_running(): + self.ios_task.cancel() + await inter.send(MessagesCZ.task_stop_success) + else: + await inter.send(MessagesCZ.task_nothing_to_stop) + + @tasks.loop(minutes=Base.config.ios_looptime_minutes) + async def ios_task(self, inter: disnake.ApplicationCommandInteraction = None): + # Respond to interaction if any, else print everything to #ios-private + channel = inter.channel if inter is not None else self.bot.get_channel(self.config.ios_channel_id) + if inter is not None: + await inter.edit_original_response(MessagesCZ.howto_clean) + else: + await channel.send(MessagesCZ.howto_clean) + + process = subprocess.Popen( + ["ssh", "-i", self.config.ios_leakcheck_key_path, "merlin"], stdout=subprocess.PIPE + ) + output, _ = process.communicate() + memory, rest = output.decode("utf-8").split("semafory:\n") + semaphores, processes = rest.split("procesy:\n") + try: + parsed_memory = features.parse_memory(memory) + parsed_semaphores, parsed_files = features.parse_semaphores(semaphores) + parsed_processes = features.parse_processes(processes) + parsed_resources = { + features.RESOURCE_TYPE.MEMORY: parsed_memory, + features.RESOURCE_TYPE.SEMAPHORE: parsed_semaphores, + features.RESOURCE_TYPE.FILE: parsed_files, + features.RESOURCE_TYPE.PROCESS: parsed_processes, + } + await features.print_output(channel, "merlinovi", features.filter_year(parsed_resources)) + except (IndexError, ValueError) as e: + await channel.send(MessagesCZ.parsing_error) + # Send it to bot-dev channel anyway + raise e + + process = subprocess.Popen( + ["ssh", "-i", self.config.ios_leakcheck_key_path, "eva"], stdout=subprocess.PIPE + ) + output, _ = process.communicate() + + memory, rest = output.decode("utf-8").split("semafory:\n") + semaphores, processes = rest.split("procesy:\n") + # remove unwanted processes + processes = features.filter_processes(processes) + try: + parsed_memory = features.parse_memory(memory) + parsed_semaphores, _ = features.parse_semaphores(semaphores) + parsed_processes = features.parse_processes(processes) + parsed_resources = { + features.RESOURCE_TYPE.MEMORY: parsed_memory, + features.RESOURCE_TYPE.SEMAPHORE: parsed_semaphores, + features.RESOURCE_TYPE.PROCESS: parsed_processes, + } + await features.print_output(channel, "evě", features.filter_year(parsed_resources)) + except (IndexError, ValueError) as e: + await channel.send(MessagesCZ.parsing_error) + # Send it to bot-dev channel anyway + raise e + # eva doesn't seem to have /dev/shm diff --git a/cogs/ios.py b/cogs/ios/features.py similarity index 57% rename from cogs/ios.py rename to cogs/ios/features.py index 785eb7897..d9346801d 100644 --- a/cogs/ios.py +++ b/cogs/ios/features.py @@ -1,25 +1,17 @@ -""" -Cog for the IOS subject. Get users on merlin/eva server which have blocking processes running. -""" +from __future__ import annotations import datetime import re -import subprocess import disnake -from disnake.ext import commands, tasks import utils -from cogs.base import Base -from config import cooldowns -from config.messages import Messages from database import session from database.verification import PermitDB, ValidPersonDB from features.list_message_sender import send_list_of_messages -from permissions import permission_check -def running_for(time): +def running_for(time: str) -> int: now = datetime.datetime.now() time = time.split(":") if len(time) == 2: @@ -37,14 +29,14 @@ def running_for(time): return minutes - 1440 -def unchanged_for(date, format_str): +def unchanged_for(date: str, format_str: str) -> int: now = datetime.datetime.now() date = datetime.datetime.strptime(date, format_str) return (now - date.replace(year=now.year)).total_seconds() // 60 # filter people and keep only those containing "BIT" or "FEKT" in person.year -def filter_year(resources): +def filter_year(resources: dict[RESOURCE_TYPE, dict]): # get unique logins and people objects from db logins = set(login for res_data in resources.values() for login in res_data.keys()) people = { @@ -61,7 +53,7 @@ def filter_year(resources): return out_res -def parse_memory(memory): +def parse_memory(memory: str) -> dict: parsed = {} for line in memory.strip().splitlines(): line = line.split() @@ -77,7 +69,7 @@ def parse_memory(memory): return parsed -def parse_semaphores(semaphores): +def parse_semaphores(semaphores: str) -> tuple[dict, dict]: parsed = {} parsed_files = {} if "soubory semaforu" in semaphores: @@ -115,7 +107,7 @@ def parse_semaphores(semaphores): return parsed, parsed_files -def parse_processes(processes): +def parse_processes(processes: str) -> dict: parsed = {} for line in processes.strip().splitlines(): line = line.split() @@ -131,7 +123,7 @@ def parse_processes(processes): return parsed -def filter_processes(processes): +def filter_processes(processes: str) -> str: out = [] for line in processes.strip().splitlines(): if re.search(r"/[a-zA-Z0-9.]+ \d+ \d+ \d+ \d+ \d+$", line): @@ -139,7 +131,7 @@ def filter_processes(processes): return "\n".join(out) -def format_time(minutes): +def format_time(minutes: int) -> str: hours = minutes / 60 days = hours / 24 weeks = days / 7 @@ -174,7 +166,7 @@ class RESOURCE_TYPE: } -def insult_login(parsed_items, system, res_type): +def insult_login(parsed_items: dict, system: str, res_type: RESOURCE_TYPE) -> list[str]: output_array = [] for login, array in parsed_items.items(): user = session.query(PermitDB).filter(PermitDB.login == login).one_or_none() @@ -197,7 +189,7 @@ def insult_login(parsed_items, system, res_type): return output_array -def insult_login_shm(parsed_files, system): +def insult_login_shm(parsed_files: dict, system: str) -> list[str]: output_array = [] for login, data in parsed_files.items(): user = session.query(PermitDB).filter(PermitDB.login == login).one_or_none() @@ -221,7 +213,9 @@ def insult_login_shm(parsed_files, system): return output_array -async def print_output(bot, channel, system, resources): +async def print_output( + channel: disnake.TextChannel, system: str, resources: dict[RESOURCE_TYPE, dict] +) -> None: out_arr = [] for res_type in [RESOURCE_TYPE.MEMORY, RESOURCE_TYPE.SEMAPHORE, RESOURCE_TYPE.PROCESS]: if resources.get(res_type): @@ -233,108 +227,3 @@ async def print_output(bot, channel, system, resources): await channel.send(f"Na {system} uklizeno <:HYPERS:493154327318233088>") else: await send_list_of_messages(channel, out_arr) - - -class IOS(Base, commands.Cog): - def __init__(self, bot): - super().__init__() - self.bot = bot - self.tasks = [self.ios_task] - - @cooldowns.default_cooldown - @commands.check(permission_check.helper_plus) - @commands.slash_command(name="ios", description=Messages.ios_brief, guild_ids=[Base.config.guild_id]) - async def ios(self, inter: disnake.ApplicationCommandInteraction): - await inter.response.defer() - await self.ios_task(inter) - - @commands.slash_command(name="ios_task", guild_ids=[Base.config.guild_id]) - async def _ios(self, inter): - pass - - @commands.check(permission_check.is_bot_admin) - @_ios.sub_command(name="start", description=Messages.ios_task_start_brief) - async def ios_task_start(self, inter: disnake.ApplicationCommandInteraction): - try: - self.ios_task.start() - await inter.send(Messages.ios_task_start_success) - except RuntimeError: - await inter.send(Messages.ios_task_start_already_set) - - @commands.check(permission_check.is_bot_admin) - @_ios.sub_command(name="stop", description=Messages.ios_task_stop_brief) - async def ios_task_stop(self, inter: disnake.ApplicationCommandInteraction): - if self.ios_task.is_running(): - self.ios_task.stop() - await inter.send(Messages.ios_task_stop_success) - else: - await inter.send(Messages.ios_task_stop_nothing_to_stop) - - @commands.check(permission_check.is_bot_admin) - @_ios.sub_command(name="cancel", description=Messages.ios_task_cancel_brief) - async def ios_task_cancel(self, inter: disnake.ApplicationCommandInteraction): - if self.ios_task.is_running(): - self.ios_task.cancel() - await inter.send(Messages.ios_task_stop_success) - else: - await inter.send(Messages.ios_task_stop_nothing_to_stop) - - @tasks.loop(minutes=Base.config.ios_looptime_minutes) - async def ios_task(self, inter: disnake.ApplicationCommandInteraction = None): - # Respond to interaction if any, else print everything to #ios-private - channel = inter.channel if inter is not None else self.bot.get_channel(self.config.ios_channel_id) - if inter is not None: - await inter.edit_original_response(Messages.ios_howto_clean) - else: - await channel.send(Messages.ios_howto_clean) - - process = subprocess.Popen( - ["ssh", "-i", self.config.ios_leakcheck_key_path, "merlin"], stdout=subprocess.PIPE - ) - output, _ = process.communicate() - memory, rest = output.decode("utf-8").split("semafory:\n") - semaphores, processes = rest.split("procesy:\n") - try: - parsed_memory = parse_memory(memory) - parsed_semaphores, parsed_files = parse_semaphores(semaphores) - parsed_processes = parse_processes(processes) - parsed_resources = { - RESOURCE_TYPE.MEMORY: parsed_memory, - RESOURCE_TYPE.SEMAPHORE: parsed_semaphores, - RESOURCE_TYPE.FILE: parsed_files, - RESOURCE_TYPE.PROCESS: parsed_processes, - } - await print_output(self.bot, channel, "merlinovi", filter_year(parsed_resources)) - except (IndexError, ValueError) as e: - await channel.send(Messages.ios_parsing_error) - # Send it to bot-dev channel anyway - raise e - - process = subprocess.Popen( - ["ssh", "-i", self.config.ios_leakcheck_key_path, "eva"], stdout=subprocess.PIPE - ) - output, _ = process.communicate() - - memory, rest = output.decode("utf-8").split("semafory:\n") - semaphores, processes = rest.split("procesy:\n") - # remove unwanted processes - processes = filter_processes(processes) - try: - parsed_memory = parse_memory(memory) - parsed_semaphores, _ = parse_semaphores(semaphores) - parsed_processes = parse_processes(processes) - parsed_resources = { - RESOURCE_TYPE.MEMORY: parsed_memory, - RESOURCE_TYPE.SEMAPHORE: parsed_semaphores, - RESOURCE_TYPE.PROCESS: parsed_processes, - } - await print_output(self.bot, channel, "evě", filter_year(parsed_resources)) - except (IndexError, ValueError) as e: - await channel.send(Messages.ios_parsing_error) - # Send it to bot-dev channel anyway - raise e - # eva doesn't seem to have /dev/shm - - -def setup(bot): - bot.add_cog(IOS(bot)) diff --git a/cogs/ios/messages_cz.py b/cogs/ios/messages_cz.py new file mode 100644 index 000000000..d8d3edae8 --- /dev/null +++ b/cogs/ios/messages_cz.py @@ -0,0 +1,15 @@ +from config.messages import Messages as GlobalMessages +from config.app_config import config + + +class MessagesCZ(GlobalMessages): + ios_brief = "Připomene všem prasatům, že si mají jít po sobě uklidit" + task_start_brief = "Začne pravidelně připomínat všem prasatům, že si mají jít po sobě uklidit" + task_start_success = f"Automatické připomínání úspěšně nastaveno. Bude se od teď provádět každých {config.ios_looptime_minutes} minut." + task_start_already_set = "Automatické připomínání už je nastaveno." + task_stop_brief = "Zastaví automatické připomínání" + task_stop_success = "Automatické připomínání zastaveno." + task_nothing_to_stop = "Automatické připomínání není nastaveno." + task_cancel_brief = "Okamžitě ukončí automatické připomínání (přeruší aktuální běh)" + parsing_error = "Toastere, máš bordel v parsování." + howto_clean = "Pokud nevíte, jak po sobě uklidit, checkněte: https://discordapp.com/channels/461541385204400138/534431057001316362/698701631495340033" diff --git a/config/messages.py b/config/messages.py index 65d16748b..5b7da7911 100644 --- a/config/messages.py +++ b/config/messages.py @@ -76,18 +76,6 @@ class Messages(metaclass=Formatable): bot_room_redirect = "{user} <:sadcat:576171980118687754> 👉 " \ "<#{bot_room}>\n" - # IOS - ios_brief = "Připomene všem prasatům, že si mají jít po sobě uklidit" - ios_task_start_brief = "Začne pravidelně připomínat všem prasatům, že si mají jít po sobě uklidit" - ios_task_start_success = f"Automatické připomínání úspěšně nastaveno. Bude se od teď provádět každých {config.ios_looptime_minutes} minut." - ios_task_start_already_set = "Automatické připomínání už je nastaveno." - ios_task_stop_brief = "Zastaví automatické připomínání" - ios_task_stop_success = "Automatické připomínání zastaveno." - ios_task_stop_nothing_to_stop = "Automatické připomínání není nastaveno." - ios_task_cancel_brief = "Okamžitě ukončí automatické připomínání (přeruší aktuální běh)" - ios_parsing_error = "Toastere, máš bordel v parsování." - ios_howto_clean = "Pokud nevíte, jak po sobě uklidit, checkněte: https://discordapp.com/channels/461541385204400138/534431057001316362/698701631495340033" - # KARMA karma = "{user} Karma uživatele `{target}` je: **{karma}** " \ "(**{order}.**)\nA rozdal:\n" \