Slash commands

fred/cogs/crashes.py

@@ -8,7 +8,18 @@
 from asyncio import Task, TaskGroup
 from os.path import split
 from pathlib import Path
-from typing import AsyncIterator, IO, Type, Coroutine, Generator, Optional, Any, Final, TypedDict, AsyncGenerator
+from typing import (
+    AsyncIterator,
+    IO,
+    Type,
+    Coroutine,
+    Generator,
+    Optional,
+    Any,
+    Final,
+    TypedDict,
+    AsyncGenerator,
+)
 from urllib.parse import urlparse
 from zipfile import ZipFile
 
@@ -197,7 +208,11 @@ def replace_response_value_with_captured(m) -> str:
                             return f"{{Group {group} not captured in crash regex!}}"
                         return match.group(group)
 
-                    response = re2.sub(r"{(\d+)}", replace_response_value_with_captured, str(crash["response"]))
+                    response = re2.sub(
+                        r"{(\d+)}",
+                        replace_response_value_with_captured,
+                        str(crash["response"]),
+                    )
                     yield CrashResponse(name=crash["name"], value=response, inline=True)
 
     async def detect_and_fetch_pastebin_content(self, text: str) -> str:
@@ -219,15 +234,21 @@ async def process_text(self, text: str, filename="") -> list[CrashResponse]:
         responses.extend(await self.process_text(await self.detect_and_fetch_pastebin_content(text)))
 
         if match := await safe_search(
-            r"([^\n]*Critical error:.*Engine exit[^\n]*\))", text, flags=re2.I | re2.M | re2.S
+            r"([^\n]*Critical error:.*Engine exit[^\n]*\))",
+            text,
+            flags=re2.I | re2.M | re2.S,
         ):
             filename = os.path.basename(filename)
             crash = match.group(1)
             responses.append(
                 CrashResponse(
                     name=f"Crash found in {filename}",
                     value="It has been attached to this message.",
-                    attachment=File(io.StringIO(crash), filename="Abridged " + filename, force_close=True),
+                    attachment=File(
+                        io.StringIO(crash),
+                        filename="Abridged " + filename,
+                        force_close=True,
+                    ),
                 )
             )
 
@@ -284,7 +305,8 @@ def _ext_filter(ext: str) -> bool:
 
     async def _obtain_attachments(self, message: Message) -> AsyncGenerator[tuple[str, IO | Exception], None, None]:
         cdn_links = re2.findall(
-            r"(https://(?:cdn.discordapp.com|media.discordapp.net)/attachments/\S+)", message.content
+            r"(https://(?:cdn.discordapp.com|media.discordapp.net)/attachments/\S+)",
+            message.content,
         )
 
         yield bool(cdn_links or message.attachments)
@@ -349,7 +371,8 @@ async def process_message(self, message: Message) -> bool:
                         self.logger.exception(file_or_exc)
                         responses.append(
                             CrashResponse(
-                                name="Download failed", value=f"Could not obtain file '{name}' due to `{file_or_exc}`"
+                                name="Download failed",
+                                value=f"Could not obtain file '{name}' due to `{file_or_exc}`",
                             )
                         )
                         continue
@@ -582,7 +605,10 @@ def _get_fg_log_details(log_file: IO[bytes]):
             logger.info("Didn't find all four pieces of information normally found in a log!")
             logger.debug(json.dumps(info, indent=2))
 
-        mod_loader_logs = filter(lambda l: re2.search("LogSatisfactoryModLoader", l), map(lambda b: b.decode(), lines))
+        mod_loader_logs = filter(
+            lambda l: re2.search("LogSatisfactoryModLoader", l),
+            map(lambda b: b.decode(), lines),
+        )
 
         for line in mod_loader_logs:
             if match := re2.search(r"(?<=v\.)(?P<sml>[\d.]+)", line):

fred/cogs/levelling.py

@@ -71,7 +71,8 @@ async def validate_role(self):
             if not role:
                 logpayload["role_id"] = role_id
                 logger.warning(
-                    "Could not validate someone's level role because the role isn't in the main guild", extra=logpayload
+                    "Could not validate someone's level role because the role isn't in the main guild",
+                    extra=logpayload,
                 )
                 return
             self.DB_user.rank_role_id = role_id
@@ -80,7 +81,10 @@ async def validate_role(self):
                 for member_role in self.member.roles:
                     if config.RankRoles.fetch_by_role(member_role.id) is not None:  # i.e. member_role is a rank role
                         logpayload["role_id"] = member_role.id
-                        logger.info("Removing a mismatched level role from someone", extra=logpayload)
+                        logger.info(
+                            "Removing a mismatched level role from someone",
+                            extra=logpayload,
+                        )
                         await self.member.remove_roles(member_role)
                 logpayload["role_id"] = role.id
                 logger.info("Removing a mismatched level role from someone", logpayload)

fred/cogs/mediaonly.py

@@ -33,7 +33,10 @@ async def _process_message(self, message: Message, *, thread: bool) -> bool:
         ctx = await self.bot.get_context(message)
 
         if await common.l4_only(ctx):
-            self.logger.info("Message doesn't contain media but the author is a T3", extra=common.message_info(message))
+            self.logger.info(
+                "Message doesn't contain media but the author is a T3",
+                extra=common.message_info(message),
+            )
             return False
 
         if thread:

fred/cogs/webhooklistener.py

@@ -97,7 +97,10 @@ def do_POST(self):
 
             # Return error if the payload type is that other weird format instead of a normal json
             if content_type != "application/json":
-                logger.error("POST request has invalid content_type", extra={"content_type": content_type})
+                logger.error(
+                    "POST request has invalid content_type",
+                    extra={"content_type": content_type},
+                )
                 self.send_error(400, "Bad Request", "Expected a JSON request")
                 return
 

fred/cogs/welcome.py

@@ -14,10 +14,16 @@ async def on_member_join(self, member: Member):
 
     async def send_welcome_message(self, member: Member | User):
         if welcome := config.Misc.fetch("welcome_message"):
-            self.logger.info("Sending the welcome message to a new member", extra=common.user_info(member))
+            self.logger.info(
+                "Sending the welcome message to a new member",
+                extra=common.user_info(member),
+            )
             await self.bot.send_safe_direct_message(member, welcome)
 
         if info := config.Misc.fetch("latest_info"):
-            self.logger.info("Sending the latest information to a new member", extra=common.user_info(member))
+            self.logger.info(
+                "Sending the latest information to a new member",
+                extra=common.user_info(member),
+            )
             info = f"Here's the latest information :\n\n{info}"
             await self.bot.send_safe_direct_message(member, info)

fred/config.py

@@ -6,7 +6,16 @@
 from typing import Optional, Any, TypedDict
 
 import nextcord
-from sqlobject import SQLObject, IntCol, BoolCol, JSONCol, BigIntCol, StringCol, FloatCol, sqlhub
+from sqlobject import (
+    SQLObject,
+    IntCol,
+    BoolCol,
+    JSONCol,
+    BigIntCol,
+    StringCol,
+    FloatCol,
+    sqlhub,
+)
 from sqlobject.dberrors import DuplicateEntryError
 
 

fred/fred.py

@@ -227,6 +227,21 @@ async def safe_send(
 
         return await to.send(content=content, files=files, **kwargs)
 
+    async def reply_generic(
+        self,
+        target: nextcord.Message | nextcord.Interaction | commands.Context,
+        content: Optional[str] = None,
+        propagate_reply: bool = True,
+        **kwargs,
+    ) -> nextcord.Message:
+        if isinstance(target, nextcord.Message):
+            return await self.reply_to_msg(target, content, propagate_reply, **kwargs)
+        if isinstance(target, nextcord.Interaction):
+            return await target.send(content, **kwargs)
+        if isinstance(target, commands.Context):
+            return await self.reply_to_msg(target.message, content, propagate_reply, **kwargs)
+        raise TypeError(f"Unsupported type {type(target)}")
+
     async def reply_to_msg(
         self,
         message: nextcord.Message,
@@ -275,7 +290,7 @@ def check(message2: nextcord.Message):
             await self.reply_to_msg(message, "Timed out and aborted after 120 seconds.")
             raise asyncio.TimeoutError
 
-        return response.content, response.attachments[0] if response.attachments else None
+        return response.content, (response.attachments[0] if response.attachments else None)
 
     async def reply_yes_or_no(self, message: nextcord.Message, question: Optional[str] = None, **kwargs) -> bool:
         response, _ = await self.reply_question(message, question, **kwargs)
@@ -300,15 +315,21 @@ async def on_message(self, message: nextcord.Message):
             self.logger.info("Processing a DM", extra=common.message_info(message))
             if message.content.lower() == "start":
                 config.Users.fetch(message.author.id).accepts_dms = True
-                self.logger.info("A user now accepts to receive DMs", extra=common.message_info(message))
+                self.logger.info(
+                    "A user now accepts to receive DMs",
+                    extra=common.message_info(message),
+                )
                 await self.reply_to_msg(
                     message,
                     "You will now receive direct messages from me again! If you change your mind, send a message that says `stop`.",
                 )
                 return
             elif message.content.lower() == "stop":
                 config.Users.fetch(message.author.id).accepts_dms = False
-                self.logger.info("A user now refuses to receive DMs", extra=common.message_info(message))
+                self.logger.info(
+                    "A user now refuses to receive DMs",
+                    extra=common.message_info(message),
+                )
                 await self.reply_to_msg(
                     message,
                     "You will no longer receive direct messages from me! To resume, send a message that says `start`.",