Feat: start of message constructor

Author: DirectiveAthena

src/AthenaTwitchBot/__init__.py

@@ -4,3 +4,7 @@
 from AthenaTwitchBot.decorators.command import commandmethod
 
 from AthenaTwitchBot.models.twitch_bot import TwitchBot
+from AthenaTwitchBot.models.twitch_bot_protocol import TwitchBotProtocol
+
+# keep this function to be the last to be imported
+from AthenaTwitchBot.functions.launch import launch

src/AthenaTwitchBot/decorators/command.py

@@ -13,9 +13,11 @@
 # ----------------------------------------------------------------------------------------------------------------------
 def commandmethod(name:str):
     def decorator(fnc):
-        fnc.is_command = True
-        fnc.command_name = name
         def wrapper(*args, **kwargs):
             return fnc(*args, **kwargs)
+
+        # store attributes for later use by the bot
+        wrapper.command_name = name
+
         return wrapper
     return decorator

src/AthenaTwitchBot/functions/launch.py

@@ -0,0 +1,47 @@
+# ----------------------------------------------------------------------------------------------------------------------
+# - Package Imports -
+# ----------------------------------------------------------------------------------------------------------------------
+# General Packages
+from __future__ import annotations
+import asyncio
+from typing import Callable
+
+# Custom Library
+
+# Custom Packages
+from AthenaTwitchBot.models.twitch_bot import TwitchBot
+from AthenaTwitchBot.models.twitch_bot_protocol import TwitchBotProtocol
+
+# ----------------------------------------------------------------------------------------------------------------------
+# - Code -
+# ----------------------------------------------------------------------------------------------------------------------
+def launch(
+        bot:TwitchBot=None,
+        protocol_factory:Callable=None,
+        *,
+        host:str='irc.chat.twitch.tv',
+        port:int=6667 #todo make this into the ssl port
+):
+    # a bot always has to be defined
+    if bot is None or not isinstance(bot, TwitchBot):
+        raise SyntaxError("a proper bot has not been defined")
+
+    loop = asyncio.get_event_loop()
+
+    # assemble the protocol if a custom hasn't been defined
+    if protocol_factory is None:
+        protocol_factory = lambda: TwitchBotProtocol(
+            bot=bot,
+            main_loop=loop,
+        )
+
+    loop.run_until_complete(
+        loop.create_connection(
+            protocol_factory=protocol_factory,
+            host=host,
+            port=port,
+        )
+    )
+    loop.run_forever()
+    loop.close()
+

src/AthenaTwitchBot/functions/twitch_irc_messages.py

@@ -20,4 +20,8 @@ def prep_message(message:str) -> bytes:
 nick = lambda nickname: prep_message(f"NICK {nickname}")
 password = lambda oauth_token: prep_message(f"PASS oauth:{oauth_token}")
 join = lambda channel: prep_message(f"JOIN #{channel}")
-pong = lambda message: prep_message(f"PONG {message}")
+pong = lambda message: prep_message(f"PONG {message}")
+
+request_commands = prep_message("CAP REQ :twitch.tv/commands")
+request_membership = prep_message("CAP REQ :twitch.tv/membership")
+request_tags = prep_message("CAP REQ :twitch.tv/tags")

src/AthenaTwitchBot/functions/twitch_message_constructors.py

@@ -0,0 +1,96 @@
+# ----------------------------------------------------------------------------------------------------------------------
+# - Package Imports -
+# ----------------------------------------------------------------------------------------------------------------------
+# General Packages
+from __future__ import annotations
+from typing import Callable
+from datetime import datetime
+
+# Custom Library
+from AthenaColor import StyleNest, ForeNest, HEX
+
+# Custom Packages
+from AthenaTwitchBot.models.twitch_message import TwitchMessage, TwitchMessagePing, TwitchMessageOnlyForBot
+
+# ----------------------------------------------------------------------------------------------------------------------
+# - Support Code -
+# ----------------------------------------------------------------------------------------------------------------------
+def _find_PING(content:list[str]) -> TwitchMessagePing|False:
+    if content[0] == "PING":
+        # construct the ping message again as this has to be sent to TWITCH again
+        return TwitchMessagePing(text=" ".join(content[1:]))
+    return False
+
+def _find_bot_only(content:list[str],message:str, bot_name:str) -> TwitchMessageOnlyForBot|False:
+    if content[0] == ":tmi.twitch.tv":
+        return TwitchMessageOnlyForBot(text=message)
+    elif content[0] == f":{bot_name}!{bot_name}@{bot_name}.tmi.twitch.tv":
+        return TwitchMessageOnlyForBot(text=message)
+    return False
+
+TAG_MAPPING:dict[str:Callable] = {
+    "@badge-info": lambda tm, tag_value: setattr(tm, "badge_info", tag_value),
+    "badges": lambda tm, tag_value: setattr(tm, "badges", tag_value),
+    "client-nonce": lambda tm, tag_value: setattr(tm, "client_nonce", tag_value),
+    "color": lambda tm, tag_value: setattr(tm, "color", HEX(tag_value)),
+    "display-name": lambda tm, tag_value: setattr(tm, "display_name", tag_value),
+    "emotes": lambda tm, tag_value: setattr(tm, "emotes", tag_value),
+    "first-msg": lambda tm, tag_value: setattr(tm, "first_msg", bool(tag_value)),
+    "flags": lambda tm, tag_value: setattr(tm, "flags", tag_value),
+    "id": lambda tm, tag_value: setattr(tm, "message_id", int(tag_value)),
+    "mod": lambda tm, tag_value: setattr(tm, "mod", bool(tag_value)),
+    "room-id": lambda tm, tag_value: setattr(tm, "room_id", tag_value),
+    "subscriber": lambda tm, tag_value: setattr(tm, "subscriber", bool(tag_value)),
+    "tmi-sent-ts": lambda tm, tag_value: setattr(tm, "tmi_sent_ts", datetime.fromtimestamp(int(tag_value))),
+    "turbo": lambda tm, tag_value: setattr(tm, "turbo", bool(tag_value)),
+    "user-id": lambda tm, tag_value: setattr(tm, "user_id", int(tag_value)),
+    "user-type": lambda tm, tag_value: setattr(tm, "user-type", tag_value),
+
+}
+
+# ----------------------------------------------------------------------------------------------------------------------
+# - Code -
+# ----------------------------------------------------------------------------------------------------------------------
+def twitch_message_constructor_tags(message_bytes:bytearray, bot_name:str) -> TwitchMessage:
+    print(message_bytes)
+    message = message_bytes.decode("UTF_8").replace("\r\n", "")
+    content = message.split(" ")
+
+    # Certain twitch sent messages have a different consistency than a regular user sent message
+    #   If this happens, this has to be caught before the message is parsed further
+    if ping_message := _find_PING(content):
+        return ping_message
+
+    if bot_only_message := _find_bot_only(content, message, bot_name):
+        return bot_only_message
+
+    # with tags enabled, we know that the first element of the list above contains all the user's tags
+    #   This enables us to loop them and assign them to the message
+    #   This is done to make them accessible to the command parsing
+    # The second part of the split message is the user definement. The user id is found in the tags
+    # IRC message string is found next
+    # The channel from which it is sent is also recieved.
+    #   When the bot is only installed in one channel, this isn't usefull, but if a bot is used in multiple channels
+    #   this is part of the usefull known context
+    # Finally all text should be clumped together again, to be searched though for a custom command
+    #   This is to be done by the protocol class, not the message constructir
+
+    tags, user, irc_message, channel, *text = content
+    twitch_message:TwitchMessage = TwitchMessage(
+        message=message,
+        message_type=irc_message,
+        channel=channel,
+        text=" ".join(text)
+    )
+
+
+    for tag in tags.split(";"):
+        # dict mapping is easier than a whole IF-ELSE or match case check
+        tag_name, tag_value = tag.split("=")
+        try:
+            TAG_MAPPING[tag_name](tm=twitch_message,tag_value=tag_value)
+        except KeyError:
+            print(StyleNest.Bold(ForeNest.Maroon(f"Unkown tag of '{tag_name}' found. Please create a bug report on the git repo")))
+            pass
+
+    return twitch_message

src/AthenaTwitchBot/models/twitch_bot.py

@@ -4,16 +4,15 @@
 # General Packages
 from __future__ import annotations
 import asyncio
-from dataclasses import dataclass, field
-import socket
+from dataclasses import dataclass, field, InitVar
 from typing import Callable
+import inspect
 
 # Custom Library
 import AthenaLib
 import AthenaColor
 
 # Custom Packages
-from AthenaTwitchBot.models.twitch_bot_protocol import TwitchBotProtocol
 
 # ----------------------------------------------------------------------------------------------------------------------
 # - Code -
@@ -25,38 +24,33 @@ class TwitchBot:
     channel:str
     prefix:str
 
-    commands:dict[str: Callable]=field(default_factory=dict) # made part of init if someone wants to feel the pain of adding commands manually
+    # Twitch-specific capabilities : https://dev.twitch.tv/docs/irc/capabilities
+    twitch_capibility_commands:bool=False
+    twitch_capibility_membership:bool=False
+    twitch_capibility_tags:bool=True # only one that has the default set to true, as this is required to make reply's work
+
+    predefined_commands:InitVar[dict[str: Callable]]=None # made part of init if someone wants to feel the pain of adding commands manually
+
+    # noinspection PyDataclass
+    commands:dict[str: Callable]=field(init=False)
 
     # non init slots
 
     # ------------------------------------------------------------------------------------------------------------------
     # - Code -
     # ------------------------------------------------------------------------------------------------------------------
     def __new__(cls, *args, **kwargs):
-        # don't set self.commands to a new dictionary
-        #   as this might have been defined due to the field default factory
         # Loop over own functions to see if any is decorated with the command setup
+        cls.commands = {}
+        for k,v in cls.__dict__.items():
+            if inspect.isfunction(v) and getattr(v, "command_name", False):
+                cls.commands[v.command_name] = v
 
-        # surpressed because of pycharm being an ass
-        # noinspection PyTypeChecker
-        return type.__new__(cls, *args, **kwargs)
-
-    def launch(self):
-        loop = asyncio.get_event_loop()
-        coro = loop.create_connection(
-            protocol_factory = lambda: TwitchBotProtocol(
-                bot_nickname=self.nickname,
-                bot_oauth_token=self.oauth_token,
-                bot_channel=self.channel,
-                main_loop=loop,
-                command_prefix=self.prefix
-            ),
-            host='irc.chat.twitch.tv',
-            port=6667,
-        )
-        loop.run_until_complete(coro)
-        loop.run_forever()
-        loop.close()
+        # create the actual instance
+        return super(TwitchBot, cls).__new__(cls,*args,**kwargs)
 
+    def __post_init__(self, predefined_commands: dict[str: Callable]=None):
+        if predefined_commands is not None:
+            self.commands |= predefined_commands
 
 

src/AthenaTwitchBot/models/twitch_bot_protocol.py

@@ -5,62 +5,69 @@
 from __future__ import annotations
 import asyncio
 from dataclasses import dataclass, field
+from typing import Callable
 
 # Custom Library
 from AthenaColor import ForeNest
 
 # Custom Packages
 import AthenaTwitchBot.functions.twitch_irc_messages as messages
-from AthenaTwitchBot.models.twitch_message import TwitchMessage
+from AthenaTwitchBot.functions.twitch_message_constructors import twitch_message_constructor_tags
+
+from AthenaTwitchBot.models.twitch_message import TwitchMessage, TwitchMessagePing
+from AthenaTwitchBot.models.twitch_bot import TwitchBot
 
 # ----------------------------------------------------------------------------------------------------------------------
 # - Code -
 # ----------------------------------------------------------------------------------------------------------------------
 @dataclass(kw_only=True, slots=True, eq=False, order=False)
 class TwitchBotProtocol(asyncio.Protocol):
-    bot_nickname:str
-    bot_oauth_token:str
-    bot_channel:str
-    command_prefix:str
+    bot:TwitchBot
     main_loop:asyncio.AbstractEventLoop
 
     # non init slots
     transport:asyncio.transports.Transport = field(init=False)
+    message_constructor:Callable = field(init=False)
+
+    def __post_init__(self):
+        if self.bot.twitch_capibility_tags:
+            self.message_constructor = twitch_message_constructor_tags
+        else:
+            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))
-        self.transport.write(messages.nick(nickname=self.bot_nickname))
-        self.transport.write(messages.join(channel=self.bot_channel))
+        self.transport.write(messages.password(oauth_token=self.bot.oauth_token))
+        self.transport.write(messages.nick(nickname=self.bot.nickname))
+        self.transport.write(messages.join(channel=self.bot.channel))
+        self.transport.write(messages.request_tags)
 
     def data_received(self, data: bytearray) -> None:
-        match (twitch_message := TwitchMessage(data)):
+        match (twitch_message := self.message_constructor(data, bot_name=self.bot.nickname)):
             # Keepalive messages : https://dev.twitch.tv/docs/irc#keepalive-messages
-            case TwitchMessage(message=["PING", *_]):
+            case TwitchMessagePing():
                 print(ForeNest.ForestGreen("PINGED BY TWITCH"))
-                self.transport.write(pong_message := messages.pong(
-                    message=twitch_message.message[1:-1]
-                ))
-                print(data, pong_message)
+                self.transport.write(pong_message := messages.pong(message=twitch_message.text))
+                print(pong_message)
 
             # catch a message which starts with a command:
-            case TwitchMessage(message=[_,_,_,str(user_message),*user_message_other]) if user_message.startswith(f":{self.command_prefix}"):
+            case TwitchMessage(message=[_,_,_,str(user_message),*user_message_other]) if user_message.startswith(f":{self.bot.prefix}"):
                 user_message:str
                 print(ForeNest.ForestGreen("COMMAND CAUGHT"))
-                print(user_message, user_message_other)
+                try:
+                    user_cmd = user_message.replace(f":{self.bot.prefix}", "")
+                    result = self.bot.commands[user_cmd](self=self.bot,transport=self.transport)
+                    print(result)
+                except KeyError:
+                    pass
 
             # catch a message which has a command within it:
             case TwitchMessage(message=[_,_,_,*messages_parts]):
-                print(messages_parts)
                 for message in messages_parts:
-                    if message.startswith(self.command_prefix):
+                    if message.startswith(self.bot.prefix):
                         print(ForeNest.ForestGreen("COMMAND CAUGHT"))
-                        print(messages_parts)
-                print(data)
-            case _:
-                print(data)
 
     def connection_lost(self, exc: Exception | None) -> None:
         self.main_loop.stop()

src/AthenaTwitchBot/models/twitch_message.py

@@ -3,18 +3,43 @@
 # ----------------------------------------------------------------------------------------------------------------------
 # General Packages
 from __future__ import annotations
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from datetime import datetime
 
 # Custom Library
+from AthenaColor import HEX
 
 # Custom Packages
 
 # ----------------------------------------------------------------------------------------------------------------------
 # - Code -
 # ----------------------------------------------------------------------------------------------------------------------
-@dataclass(init=False, slots=True, eq=True, match_args=True)
+@dataclass(slots=True, eq=True, match_args=True)
 class TwitchMessage:
-    message:list[str]
+    message:str="" # complete message without the sufix: "\r\n"
+    message_type:str=""
+    channel:str=""
+    text:str=""
 
-    def __init__(self, message_bytes:bytearray):
-        self.message = message_bytes.decode("utf_8").replace("\r\n", "").split(" ")
+    # optional info if tags is enabled
+    badge_info:str=""
+    badges:list[str]=field(default_factory=list)
+    client_nonce:str=""
+    color:HEX=field(default_factory=HEX)
+    display_name:str=""
+    first_msg:bool=False
+    message_id:int=0
+    mod:bool=False
+    room_id:str=""
+    subscriber:bool=False
+    tmi_sent_ts:datetime=field(default_factory=datetime.now)
+    turbo:bool=False
+    user_id:int=0
+# ----------------------------------------------------------------------------------------------------------------------
+# - Special message types -
+# ----------------------------------------------------------------------------------------------------------------------
+class TwitchMessagePing(TwitchMessage):
+    pass
+
+class TwitchMessageOnlyForBot(TwitchMessage):
+    pass