Feat: Twitch Message Context handler

Author: DirectiveAthena

src/AthenaTwitchBot/functions/twitch_message_constructors.py

@@ -84,6 +84,7 @@ def twitch_message_constructor_tags(message_bytes:bytearray, bot_name:str) -> Tw
         message=message,
         message_type=irc_message,
         channel=channel,
+        user=user,
         text=" ".join(text).replace(":", "", 1) #only remove the first ":"
     )
 

src/AthenaTwitchBot/models/twitch_bot.py

@@ -42,15 +42,17 @@ class TwitchBot:
     def __new__(cls, *args, **kwargs):
         # Loop over own functions to see if any is decorated with the command setup
         cls.commands = {}
+
+        # create the actual instance
+        #   Which is to be used in the commands tuple
+        obj = super(TwitchBot, cls).__new__(cls,*args,**kwargs)
         for k,v in cls.__dict__.items():
             if inspect.isfunction(v) and getattr(v, "command_name", False):
-                cls.commands[v.command_name] = v
+                cls.commands[v.command_name] = (v, obj)
 
-        # create the actual instance
-        return super(TwitchBot, cls).__new__(cls,*args,**kwargs)
+        return obj
 
     def __post_init__(self, predefined_commands: dict[str: Callable]=None):
         if predefined_commands is not None:
-            self.commands |= predefined_commands
-
-
+            # the self instance isn't assigned on the predefined_commands input
+            self.commands |= {k:(v,self) for k, v in predefined_commands.items()}

src/AthenaTwitchBot/models/twitch_bot_protocol.py

@@ -16,6 +16,7 @@
 
 from AthenaTwitchBot.models.twitch_message import TwitchMessage, TwitchMessagePing
 from AthenaTwitchBot.models.twitch_bot import TwitchBot
+from AthenaTwitchBot.models.twitch_message_context import TwitchMessageContext
 
 # ----------------------------------------------------------------------------------------------------------------------
 # - Code -
@@ -36,7 +37,6 @@ def __post_init__(self):
             raise NotImplementedError("This needs to be created")
 
     def connection_made(self, transport: asyncio.transports.Transport) -> None:
-        # todo make some sort of connector to make t
         self.transport = transport
         # first write the password then the nickname else the connection will fail
         self.transport.write(messages.password(oauth_token=self.bot.oauth_token))
@@ -59,7 +59,18 @@ def data_received(self, data: bytearray) -> None:
                     print(ForeNest.ForestGreen("COMMAND CAUGHT"))
                     try:
                         user_cmd = user_message.replace(f"{self.bot.prefix}", "")
-                        self.bot.commands[user_cmd](self=self.bot,message=twitch_message,transport=self.transport)
+                        # tuple unpacking because we have a callback
+                        #   and the object instance where the command is placed in
+                        callback, orign_obj = self.bot.commands[user_cmd]
+                        callback(
+                            self=orign_obj,
+                            # Assign a context so the user doesn't need to write the transport messages themselves
+                            #   A user opnly has to write the text
+                            context=TwitchMessageContext(
+                                message=twitch_message,
+                                transport=self.transport
+                            )
+                        )
                     except KeyError:
                         pass
 

src/AthenaTwitchBot/models/twitch_message.py

@@ -14,30 +14,38 @@
 # ----------------------------------------------------------------------------------------------------------------------
 # - Code -
 # ----------------------------------------------------------------------------------------------------------------------
+EMPTY_STR = ""
+
 @dataclass(slots=True, eq=True, match_args=True)
 class TwitchMessage:
-    message:str="" # complete message without the sufix: "\r\n"
-    message_type:str=""
-    channel:str=""
-    text:str=""
+    message:str=EMPTY_STR # complete message without the sufix: "\r\n"
+    message_type:str=EMPTY_STR
+    channel:str=EMPTY_STR
+    text:str=EMPTY_STR
+    user:str=EMPTY_STR
+    username:str=field(init=False) # decoupled after init
 
     # optional info if tags is enabled
-    badge_info:str=""
+    badge_info:str=EMPTY_STR
     badges:list[str]=field(default_factory=list)
-    client_nonce:str=""
+    client_nonce:str=EMPTY_STR
     color:HEX=field(default_factory=HEX)
-    display_name:str=""
+    display_name:str=EMPTY_STR
     first_msg:bool=False
-    message_id:str=""
+    message_id:str=EMPTY_STR
     mod:bool=False
-    room_id:str=""
+    room_id:str=EMPTY_STR
     subscriber:bool=False
     tmi_sent_ts:int=0
     turbo:bool=False
     user_id:int=0
-    emotes:str=""
-    flags:str=""
-    user_type:str=""
+    emotes:str=EMPTY_STR
+    flags:str=EMPTY_STR
+    user_type:str=EMPTY_STR
+
+    def __post_init__(self):
+        self.username = self.user.split("!")[0][1:]
+
 # ----------------------------------------------------------------------------------------------------------------------
 # - Special message types -
 # ----------------------------------------------------------------------------------------------------------------------

src/AthenaTwitchBot/models/twitch_message_context.py

@@ -0,0 +1,42 @@
+# ----------------------------------------------------------------------------------------------------------------------
+# - Package Imports -
+# ----------------------------------------------------------------------------------------------------------------------
+# General Packages
+from __future__ import annotations
+from dataclasses import dataclass, field
+import asyncio
+
+# Custom Library
+
+# Custom Packages
+from AthenaTwitchBot.models.twitch_message import TwitchMessage
+
+# ----------------------------------------------------------------------------------------------------------------------
+# - Code -
+# ----------------------------------------------------------------------------------------------------------------------
+@dataclass(slots=True,eq=False,order=False,match_args=True)
+class TwitchMessageContext:
+    message:TwitchMessage
+    transport:asyncio.Transport
+
+    # things not to be defined on init
+    username:str=field(init=False)
+
+    def __post_init__(self):
+        self.username = f"@{self.message.username}"
+
+    def reply(self, text:str):
+        """
+        a "reply" method does reply to the user's message that invoked the command
+        """
+        self.transport.write(
+            f"@reply-parent-msg-id={self.message.message_id} PRIVMSG {self.message.channel} :{text}\r\n".encode("UTF_8")
+        )
+
+    def write(self, text:str):
+        """
+        a "write" method does not reply to the message that invoked the command, but simply writes the text to the channel
+        """
+        self.transport.write(
+            f"PRIVMSG {self.message.channel} :{text}\r\n".encode("UTF_8")
+        )