Migrate IOS to folder

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))

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

cogs/ios.py → 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,15 +123,15 @@ 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):
             out.append(line)
     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))

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"

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" \