diff --git a/setup.py b/setup.py index ab33603..c285f8b 100644 --- a/setup.py +++ b/setup.py @@ -13,7 +13,7 @@ # - Code - # ---------------------------------------------------------------------------------------------------------------------- def version_handler() -> str: - version = 0,1,1 + version = 0,2,0 version_str = ".".join(str(i) for i in version) with open("src/AthenaTwitchBot/_info/_v.py", "w") as file: diff --git a/src/AthenaTwitchBot/__init__.py b/src/AthenaTwitchBot/__init__.py index 22d1246..d33c6c6 100644 --- a/src/AthenaTwitchBot/__init__.py +++ b/src/AthenaTwitchBot/__init__.py @@ -2,7 +2,7 @@ # - Package Imports - # ---------------------------------------------------------------------------------------------------------------------- from AthenaTwitchBot.decorators.command import command_method -from AthenaTwitchBot.decorators.frequentoutput import frequent_output_method +from AthenaTwitchBot.decorators.scheduled_task import scheduled_task_method from AthenaTwitchBot.models.twitch_bot import TwitchBot from AthenaTwitchBot.models.twitch_bot_protocol import TwitchBotProtocol diff --git a/src/AthenaTwitchBot/_info/_v.py b/src/AthenaTwitchBot/_info/_v.py index 85c1c73..cbe40e8 100644 --- a/src/AthenaTwitchBot/_info/_v.py +++ b/src/AthenaTwitchBot/_info/_v.py @@ -1,2 +1,2 @@ def _version(): - return '0.1.1' \ No newline at end of file + return '0.2.0' \ No newline at end of file diff --git a/src/AthenaTwitchBot/data/emotes.py b/src/AthenaTwitchBot/data/emotes.py new file mode 100644 index 0000000..f552dda --- /dev/null +++ b/src/AthenaTwitchBot/data/emotes.py @@ -0,0 +1,14 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations + +# Custom Library + +# Custom Packages + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- +four_head = "4head" diff --git a/src/AthenaTwitchBot/decorators/command.py b/src/AthenaTwitchBot/decorators/command.py index 7ed2ed7..06e243a 100644 --- a/src/AthenaTwitchBot/decorators/command.py +++ b/src/AthenaTwitchBot/decorators/command.py @@ -7,18 +7,24 @@ # Custom Library # Custom Packages +from AthenaTwitchBot.models.wrapper_helpers.command import Command # ---------------------------------------------------------------------------------------------------------------------- # - Code - # ---------------------------------------------------------------------------------------------------------------------- -def command_method(name:str): +def command_method(name:str, case_sensitive:bool=False): def decorator(fnc): - def wrapper(*args, **kwargs): - return fnc(*args, **kwargs) + def wrapper(*args_, **kwargs_): + return fnc(*args_, **kwargs_) # store attributes for later use by the bot wrapper.is_command = True - wrapper.command_name = name + # store some information + wrapper.cmd = Command( + name=name, + case_sensitive=case_sensitive, + callback=wrapper, + ) return wrapper return decorator \ No newline at end of file diff --git a/src/AthenaTwitchBot/decorators/frequentoutput.py b/src/AthenaTwitchBot/decorators/scheduled_task.py similarity index 73% rename from src/AthenaTwitchBot/decorators/frequentoutput.py rename to src/AthenaTwitchBot/decorators/scheduled_task.py index 4f738f4..7b48e1d 100644 --- a/src/AthenaTwitchBot/decorators/frequentoutput.py +++ b/src/AthenaTwitchBot/decorators/scheduled_task.py @@ -7,14 +7,16 @@ # Custom Library # Custom Packages +from AthenaTwitchBot.models.wrapper_helpers.scheduled_task import ScheduledTask # ---------------------------------------------------------------------------------------------------------------------- # - Code - # ---------------------------------------------------------------------------------------------------------------------- -def frequent_output_method(delay:int=3600): # default is every hour +def scheduled_task_method(*,delay:int=3600,before:bool=True): # default is every hour """ Create a method that runs every couple of seconds. The delay parameter is defined in seconds + :param before: :param delay: :return: """ @@ -25,7 +27,11 @@ def wrapper(*args, **kwargs): # store attributes for later use by the bot # to be used by the protocol to assign it top an async call loop - wrapper.is_frequent_output = True # typo caught by NoirPi - wrapper.delay = delay + wrapper.is_task = True # typo caught by NoirPi + wrapper.tsk = ScheduledTask( + delay=delay, + before=before, + callback=wrapper + ) return wrapper return decorator \ No newline at end of file diff --git a/src/AthenaTwitchBot/functions/launch.py b/src/AthenaTwitchBot/functions/launch.py index 6d82ca4..3999c2a 100644 --- a/src/AthenaTwitchBot/functions/launch.py +++ b/src/AthenaTwitchBot/functions/launch.py @@ -11,37 +11,57 @@ # Custom Packages from AthenaTwitchBot.models.twitch_bot import TwitchBot from AthenaTwitchBot.models.twitch_bot_protocol import TwitchBotProtocol +from AthenaTwitchBot.models.outputs.output import Output +from AthenaTwitchBot.models.outputs.output_console import OutputConsole # ---------------------------------------------------------------------------------------------------------------------- # - Code - # ---------------------------------------------------------------------------------------------------------------------- def launch( + *, # after this, keywords only bot:TwitchBot=None, protocol_factory:Callable=None, - *, + output:Output=None, host:str='irc.chat.twitch.tv', - port:int=6667 #todo make this into the ssl port + port:int=6667, #todo make this into the ssl port + + auto_restart:bool=False ): - # 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() + + if output is None: + output=OutputConsole() + output.pre_launch() + + while True: + try: + # 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,output=output) + + loop.run_until_complete( + loop.create_connection( + protocol_factory=protocol_factory, + host=host, + port=port, + ) + ) + loop.run_forever() + loop.close() + except ConnectionResetError: + print("not connection") + if auto_restart: + loop = asyncio.get_running_loop() + loop.stop() + continue + else: + break + + except : # make sure everything else is caught else the loop will continue indefinitely + raise diff --git a/src/AthenaTwitchBot/functions/twitch_message_constructors.py b/src/AthenaTwitchBot/functions/twitch_message_constructors.py index 3943865..1ef958a 100644 --- a/src/AthenaTwitchBot/functions/twitch_message_constructors.py +++ b/src/AthenaTwitchBot/functions/twitch_message_constructors.py @@ -32,29 +32,34 @@ def _find_bot_only(content:list[str],message:str, bot_name:str) -> TwitchMessage 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", 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", 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), + "@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) if tag_value else HEX()), + "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(int(tag_value))), + "flags": lambda tm, tag_value: setattr(tm, "flags", tag_value), + "id": lambda tm, tag_value: setattr(tm, "message_id", tag_value), + "mod": lambda tm, tag_value: setattr(tm, "mod", bool(int(tag_value))), + "room-id": lambda tm, tag_value: setattr(tm, "room_id", tag_value), + "subscriber": lambda tm, tag_value: setattr(tm, "subscriber", bool(int(tag_value))), + "tmi-sent-ts": lambda tm, tag_value: setattr(tm, "tmi_sent_ts", int(tag_value)), + "turbo": lambda tm, tag_value: setattr(tm, "turbo", bool(int(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), + "reply-parent-display-name":lambda tm, tag_value: setattr(tm, "reply_parent_display_name", tag_value), + "reply-parent-msg-body": lambda tm, tag_value: setattr(tm, "reply_parent_msg_body", tag_value), + "reply-parent-msg-id": lambda tm, tag_value: setattr(tm, "reply_parent_msg_id", int(tag_value)), + "reply-parent-user-id": lambda tm, tag_value: setattr(tm, "reply_parent_user_id", int(tag_value)), + "reply-parent-user-login": lambda tm, tag_value: setattr(tm, "reply_parent_user_login", tag_value), + "emote-only": lambda tm, tag_value: setattr(tm, "emote_only", bool(int(tag_value))), } # ---------------------------------------------------------------------------------------------------------------------- # - Code - # ---------------------------------------------------------------------------------------------------------------------- def twitch_message_constructor_tags(message_bytes:bytearray, bot_name:str) -> TwitchMessage: - print(message_bytes) message = message_bytes.decode("UTF_8") content = message.split(" ") diff --git a/src/AthenaTwitchBot/models/outputs/__init__.py b/src/AthenaTwitchBot/models/outputs/__init__.py new file mode 100644 index 0000000..8d6d09b --- /dev/null +++ b/src/AthenaTwitchBot/models/outputs/__init__.py @@ -0,0 +1,13 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations + +# Custom Library + +# Custom Packages + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- diff --git a/src/AthenaTwitchBot/models/outputs/output.py b/src/AthenaTwitchBot/models/outputs/output.py new file mode 100644 index 0000000..8f63b56 --- /dev/null +++ b/src/AthenaTwitchBot/models/outputs/output.py @@ -0,0 +1,27 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations +from abc import ABC, abstractmethod + +# Custom Library + +# Custom Packages +from AthenaTwitchBot.models.twitch_message import TwitchMessage + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- +class Output(ABC): + + @abstractmethod + def pre_launch(self): + """Output the state of the application before anything is run""" + @abstractmethod + def message(self, message:TwitchMessage): + """Output of a received message""" + + @abstractmethod + def undefined(self,message=None): + """Output anything that is supposed to be undefined (this should eventually not be present anymore)""" \ No newline at end of file diff --git a/src/AthenaTwitchBot/models/outputs/output_console.py b/src/AthenaTwitchBot/models/outputs/output_console.py new file mode 100644 index 0000000..a42d532 --- /dev/null +++ b/src/AthenaTwitchBot/models/outputs/output_console.py @@ -0,0 +1,43 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations + +# Custom Library +from AthenaColor import StyleNest, ForeNest, BackNest + +# Custom Packages +from AthenaTwitchBot.models.outputs.output import Output +from AthenaTwitchBot.models.twitch_message import TwitchMessage, TwitchMessagePing, TwitchMessageOnlyForBot +# noinspection PyProtectedMember +from AthenaTwitchBot._info._v import _version + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- +class OutputConsole(Output): + + def pre_launch(self): + print( + ForeNest.SlateGray( + f"- AthenaTwitchBot {ForeNest.HotPink('v', _version(), sep='')} -", + f" made by Andreas Sas", + "", + sep="\n" + ), + ) + + def message(self, message:TwitchMessage): + match message: + case TwitchMessagePing(): + print(ForeNest.SlateGray(message.text), ForeNest.ForestGreen("PING RECEIVED")) + case TwitchMessageOnlyForBot(): + print(ForeNest.SlateGray(message.text)) + case TwitchMessage(first_msg=True): + print(ForeNest.BlueViolet(message.username), ForeNest.SlateGray(":"),ForeNest.White(message.text)) + case TwitchMessage(): + print(ForeNest.SlateGray(message.username, ":"),ForeNest.White(message.text)) + + def undefined(self,message=None): + print(ForeNest.SlateGray(message)) diff --git a/src/AthenaTwitchBot/models/twitch_bot.py b/src/AthenaTwitchBot/models/twitch_bot.py index 1e526e5..fe3af84 100644 --- a/src/AthenaTwitchBot/models/twitch_bot.py +++ b/src/AthenaTwitchBot/models/twitch_bot.py @@ -10,6 +10,8 @@ # Custom Library # Custom Packages +from AthenaTwitchBot.models.wrapper_helpers.command import Command +from AthenaTwitchBot.models.wrapper_helpers.scheduled_task import ScheduledTask # ---------------------------------------------------------------------------------------------------------------------- # - Code - @@ -29,8 +31,8 @@ class TwitchBot: 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) - frequent_outputs:list[tuple[Callable, int]]=field(init=False) + commands:dict[str: Command]=field(init=False) + scheduled_tasks:list[ScheduledTask]=field(init=False) # non init slots @@ -40,7 +42,7 @@ class TwitchBot: def __new__(cls, *args, **kwargs): # Loop over own functions to see if any is decorated with the command setup cls.commands = {} - cls.frequent_outputs = [] + cls.scheduled_tasks = [] # create the actual instance # Which is to be used in the commands tuple @@ -50,9 +52,9 @@ def __new__(cls, *args, **kwargs): for k,v in cls.__dict__.items(): if inspect.isfunction(v): if "is_command" in (attributes := [attribute for attribute in dir(v) if not attribute.startswith("__")]): - cls.commands[v.command_name] = v - elif "is_frequent_output" in attributes: - cls.frequent_outputs.append((v,v.delay)) + cls.commands[v.cmd.name.lower()] = v.cmd + elif "is_task" in attributes: + cls.scheduled_tasks.append(v.tsk) return obj diff --git a/src/AthenaTwitchBot/models/twitch_bot_protocol.py b/src/AthenaTwitchBot/models/twitch_bot_protocol.py index 4e91cac..06265b7 100644 --- a/src/AthenaTwitchBot/models/twitch_bot_protocol.py +++ b/src/AthenaTwitchBot/models/twitch_bot_protocol.py @@ -8,7 +8,6 @@ from typing import Callable # Custom Library -from AthenaColor import ForeNest # Custom Packages import AthenaTwitchBot.functions.twitch_irc_messages as messages @@ -17,6 +16,9 @@ from AthenaTwitchBot.models.twitch_message import TwitchMessage, TwitchMessagePing from AthenaTwitchBot.models.twitch_bot import TwitchBot from AthenaTwitchBot.models.twitch_message_context import TwitchMessageContext +from AthenaTwitchBot.models.wrapper_helpers.command import Command +from AthenaTwitchBot.models.wrapper_helpers.scheduled_task import ScheduledTask +from AthenaTwitchBot.models.outputs.output import Output # ---------------------------------------------------------------------------------------------------------------------- # - Code - @@ -24,7 +26,7 @@ @dataclass(kw_only=True, slots=True, eq=False, order=False) class TwitchBotProtocol(asyncio.Protocol): bot:TwitchBot - main_loop:asyncio.AbstractEventLoop + output:Output # non init slots transport:asyncio.transports.Transport = field(init=False) @@ -49,37 +51,53 @@ def connection_made(self, transport: asyncio.transports.Transport) -> None: # add frequent_output methods to the coroutine loop loop = asyncio.get_running_loop() - for callback, delay in self.bot.frequent_outputs: - coro = loop.create_task(self.frequent_output_call(callback,delay)) + for tsk in self.bot.scheduled_tasks: + coro = loop.create_task(self.frequent_output_call(tsk)) asyncio.ensure_future(coro, loop=loop) - async def frequent_output_call(self, callback,delay:int): + async def frequent_output_call(self, tsk:ScheduledTask): context = TwitchMessageContext( message=TwitchMessage(channel=f"#{self.bot.channel}"), transport=self.transport ) - while True: - await asyncio.sleep(delay) - callback( - self=self.bot, - context=context) + if tsk.before: # the before attribute handles if we sleep before or after the task has been called + while True: + await asyncio.sleep(tsk.delay) + tsk.callback( + self=self.bot, + context=context) + else: + while True: + tsk.callback( + self=self.bot, + context=context) + await asyncio.sleep(tsk.delay) + def data_received(self, data: bytearray) -> None: for message in data.split(b"\r\n"): + # if the bytearray is empty, just skip to the next one + if message == bytearray(b''): + continue + match (twitch_message := self.message_constructor(message, bot_name=self.bot.nickname)): # Keepalive messages : https://dev.twitch.tv/docs/irc#keepalive-messages case TwitchMessagePing(): - print(ForeNest.ForestGreen("PINGED BY TWITCH")) + self.output.message(twitch_message) self.transport.write(messages.pong(message=twitch_message.text)) + continue # catch a message which starts with a command: case TwitchMessage(text=str(user_message)) if user_message.startswith(f"{self.bot.prefix}"): - print(ForeNest.ForestGreen("COMMAND CAUGHT")) + self.output.message(twitch_message) + user_message:str try: - user_cmd = user_message.replace(f"{self.bot.prefix}", "") - # tuple unpacking because we have a callback - # and the object instance where the command is placed in - self.bot.commands[user_cmd]( + user_cmd_str = user_message.replace(self.bot.prefix, "") + twitch_command:Command = self.bot.commands[user_cmd_str.lower()] + if twitch_command.case_sensitive and user_cmd_str != twitch_command.name: + raise KeyError # the check to make the force capitalization work + + twitch_command.callback( self=self.bot, # Assign a context so the user doesn't need to write the transport messages themselves # A user only has to write the text @@ -90,7 +108,16 @@ def data_received(self, data: bytearray) -> None: ) except KeyError: pass + continue + + # if the message wasn't able to be handled by the parser above + # it will just be outputted as undefined + self.output.undefined(message) def connection_lost(self, exc: Exception | None) -> None: - self.main_loop.stop() + loop = asyncio.get_running_loop() + loop.stop() + + if exc is not None: + raise exc diff --git a/src/AthenaTwitchBot/models/twitch_message.py b/src/AthenaTwitchBot/models/twitch_message.py index 3fc9129..9f878a5 100644 --- a/src/AthenaTwitchBot/models/twitch_message.py +++ b/src/AthenaTwitchBot/models/twitch_message.py @@ -41,6 +41,12 @@ class TwitchMessage: emotes:str=EMPTY_STR flags:str=EMPTY_STR user_type:str=EMPTY_STR + emote_only:bool=False + reply_parent_display_name:str="" + reply_parent_msg_body:str="" + reply_parent_msg_id:int=0 + reply_parent_user_id:int=0 + reply_parent_user_login:str="" def __post_init__(self): self.username = self.user.split("!")[0][1:] diff --git a/src/AthenaTwitchBot/models/twitch_message_context.py b/src/AthenaTwitchBot/models/twitch_message_context.py index d5adaa2..addeafc 100644 --- a/src/AthenaTwitchBot/models/twitch_message_context.py +++ b/src/AthenaTwitchBot/models/twitch_message_context.py @@ -29,6 +29,7 @@ def reply(self, text:str): """ a "reply" method does reply to the user's message that invoked the command """ + text = text.replace('\n', '') self.transport.write( f"@reply-parent-msg-id={self.message.message_id} PRIVMSG {self.message.channel} :{text}\r\n".encode("UTF_8") ) @@ -37,6 +38,7 @@ 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 """ + text = text.replace('\n', '') self.transport.write( f"PRIVMSG {self.message.channel} :{text}\r\n".encode("UTF_8") ) \ No newline at end of file diff --git a/src/AthenaTwitchBot/models/wrapper_helpers/__init__.py b/src/AthenaTwitchBot/models/wrapper_helpers/__init__.py new file mode 100644 index 0000000..8d6d09b --- /dev/null +++ b/src/AthenaTwitchBot/models/wrapper_helpers/__init__.py @@ -0,0 +1,13 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations + +# Custom Library + +# Custom Packages + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- diff --git a/src/AthenaTwitchBot/models/wrapper_helpers/command.py b/src/AthenaTwitchBot/models/wrapper_helpers/command.py new file mode 100644 index 0000000..0eee1da --- /dev/null +++ b/src/AthenaTwitchBot/models/wrapper_helpers/command.py @@ -0,0 +1,20 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations +from dataclasses import dataclass +from typing import Callable + +# Custom Library + +# Custom Packages + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- +@dataclass(kw_only=True, match_args=True, slots=True) +class Command: + name:str + case_sensitive:bool + callback:Callable diff --git a/src/AthenaTwitchBot/models/wrapper_helpers/scheduled_task.py b/src/AthenaTwitchBot/models/wrapper_helpers/scheduled_task.py new file mode 100644 index 0000000..e150922 --- /dev/null +++ b/src/AthenaTwitchBot/models/wrapper_helpers/scheduled_task.py @@ -0,0 +1,20 @@ +# ---------------------------------------------------------------------------------------------------------------------- +# - Package Imports - +# ---------------------------------------------------------------------------------------------------------------------- +# General Packages +from __future__ import annotations +from dataclasses import dataclass +from typing import Callable + +# Custom Library + +# Custom Packages + +# ---------------------------------------------------------------------------------------------------------------------- +# - Code - +# ---------------------------------------------------------------------------------------------------------------------- +@dataclass(kw_only=True, match_args=True, slots=True) +class ScheduledTask: + delay:int + before:bool + callback:Callable \ No newline at end of file