diff --git a/src/complements_bot/bot.py b/src/complements_bot/bot.py index cae774a..3c11ea6 100644 --- a/src/complements_bot/bot.py +++ b/src/complements_bot/bot.py @@ -1,17 +1,17 @@ """ -Holds all commands (and their logic) for how ComplementsBot should complement Twitch chatters +Holds all commands for how ComplementsBot should complement Twitch chatters """ import asyncio import itertools import os import random import textwrap -from typing import Awaitable, Callable, Optional, Tuple, Union +from typing import Awaitable, Callable, Optional, Tuple, Union, Iterable from twitchio import Message from twitchio.ext import commands # , routines , eventsub -from . import database +from . import database, bot_controller from .utilities import Awaitables, remove_chars, run_with_appropriate_awaiting from ..app.app import run_app_and_bot from ..env_reader import CLIENT_SECRET, TMI_TOKEN @@ -49,14 +49,6 @@ # allow users to make complement redeems in their channel -def custom_log(msg: str) -> None: - """ - Any messages which we want to log should be passed through this method - """ - - print(msg) - - class ComplementsBot(commands.Bot): """ Inherits from TwitchIO's commands.Bot class, and adds twitch chat commands for using the bot @@ -136,18 +128,7 @@ async def event_ready(self) -> None: database.join_channel(username=self.nick, name_to_id=self.name_to_id)) if ComplementsBot.SHOULD_LOG: - custom_log(f"{self.nick} is online!") - - @staticmethod - def is_bot(username: str) -> bool: - """ - checks if a username matches that of a known or assumed bot; currently the following count as bots: - - any username ending in 'bot' - - streamlabs - """ - - return (len(username) >= 3 and username[-3:].lower() == 'bot' or - username in ("streamlabs", "streamelements")) + bot_controller.custom_log(f"{self.nick} is online!") async def event_message(self, message: Message) -> None: """ @@ -161,7 +142,7 @@ async def event_message(self, message: Message) -> None: # make sure the bot ignores itself return if ComplementsBot.SHOULD_LOG: - custom_log( + bot_controller.custom_log( f"In channel {message.channel.name}, at {message.timestamp}, " f"{message.author.name} said: {message.content}") @@ -193,7 +174,7 @@ async def event_message(self, message: Message) -> None: else: is_author_ignored, chance, is_ignoring_bots, random_complements_enabled = await awaitables.gather() should_rng_choose: bool = (100 - chance) <= 100 * random.random() < 100 - is_author_bot: bool = is_ignoring_bots and ComplementsBot.is_bot(sender) + is_author_bot: bool = is_ignoring_bots and bot_controller.is_bot(sender) if (should_rng_choose and (not is_author_ignored) @@ -204,7 +185,7 @@ async def event_message(self, message: Message) -> None: if exists: await message.channel.send(comp_msg) if ComplementsBot.SHOULD_LOG: - custom_log(f"In channel {message.channel.name}, at {message.timestamp}, {message.author.name} " + bot_controller.custom_log(f"In channel {message.channel.name}, at {message.timestamp}, {message.author.name} " f"was complemented (randomly) with: {comp_msg}") async def choose_complement(self, channel: str) -> Tuple[str, bool]: @@ -276,7 +257,7 @@ async def complement(self, ctx: commands.Context) -> None: behaviour of the command. """ - who: str = self.isolate_args(ctx.message.content) + who: str = bot_controller.isolate_args(ctx.message.content) if len(who) > 0: if who[0] == "@": who = who[1:] @@ -307,8 +288,10 @@ async def complement(self, ctx: commands.Context) -> None: if exists: await ctx.channel.send(comp_msg) if ComplementsBot.SHOULD_LOG: - custom_log(f"In channel {ctx.channel.name}, at {ctx.message.timestamp}, {ctx.message.author.name} " - f"was complemented (by command) with: {comp_msg}") + bot_controller.custom_log( + f"In channel {ctx.channel.name}, at {ctx.message.timestamp}, {ctx.message.author.name} " + f"was complemented (by command) with: {comp_msg}" + ) @commands.command() async def compunignoreme(self, ctx: commands.Context) -> None: @@ -369,7 +352,7 @@ async def send_and_log(ctx: commands.Context, msg: Optional[str]) -> None: await ctx.channel.send(msg) if ComplementsBot.SHOULD_LOG: - custom_log(msg) + bot_controller.custom_log(msg) class DoIfElse: """ @@ -416,7 +399,7 @@ async def cmd_body(ctx: commands.Context, do_always: Optional[ Union[Callable[[commands.Context], Awaitable[None]], Callable[[commands.Context], None]]] = None, do_if_else: Optional[DoIfElse] = None, - always_msg: Optional[str] = None) -> bool: + always_msg: Optional[str] = None) -> Optional[Iterable[Optional[str]]]: """ The main structure in which commands sent to the bot's channel need to be processed :param ctx: context from the original call @@ -436,33 +419,31 @@ async def cmd_body(ctx: commands.Context, permission_check_task: asyncio.Task = asyncio.create_task(run_with_appropriate_awaiting(permission_check, ctx)) + to_send: list[str] = [] if do_if_else is not None: permission_check_res, if_check_res = await asyncio.gather(permission_check_task, run_with_appropriate_awaiting(do_if_else.if_check, ctx)) if not permission_check_res: - return False + return None - to_send: Optional[str] = None if if_check_res: awaitables.add_task(run_with_appropriate_awaiting(do_if_else.do_true, ctx)) if do_if_else.true_msg: - to_send = do_if_else.true_msg.replace(ComplementsBot.F_USER, user) + to_send.append(do_if_else.true_msg.replace(ComplementsBot.F_USER, user)) else: awaitables.add_task(run_with_appropriate_awaiting(do_if_else.do_false, ctx)) if do_if_else.false_msg: - to_send = do_if_else.false_msg.replace(ComplementsBot.F_USER, user) - awaitables.add_task(ComplementsBot.send_and_log(ctx, to_send)) + to_send.append(do_if_else.false_msg.replace(ComplementsBot.F_USER, user)) elif not await permission_check_task: - return False + return None if always_msg is not None: - to_send = always_msg.replace(ComplementsBot.F_USER, user) - awaitables.add_task(ComplementsBot.send_and_log(ctx, to_send)) + to_send.append(always_msg.replace(ComplementsBot.F_USER, user)) awaitables.add_task(run_with_appropriate_awaiting(do_always, ctx)) await awaitables.gather() - return True + return to_send @commands.command() async def joinme(self, ctx: commands.Context) -> None: @@ -587,7 +568,7 @@ async def userid(self, ctx: commands.Context) -> None: Get the twitch user's ID from their username """ - userid: Optional[str] = await self.name_to_id(self.isolate_args(ctx.message.content)) + userid: Optional[str] = await self.name_to_id(bot_controller.isolate_args(ctx.message.content)) userid = userid or (await self.name_to_id(ctx.author.name)) await ComplementsBot.cmd_body( @@ -604,7 +585,7 @@ async def username(self, ctx: commands.Context) -> None: Get the twitch user's ID from their username """ - username: Optional[str] = await self.id_to_name(self.isolate_args(ctx.message.content)) + username: Optional[str] = await self.id_to_name(bot_controller.isolate_args(ctx.message.content)) await ComplementsBot.cmd_body( ctx, @@ -688,7 +669,7 @@ async def setchance(self, ctx: commands.Context) -> None: channel: str = ctx.channel.name to_send: str exception: bool = False - chance_str: str = self.isolate_args(ctx.message.content) + chance_str: str = bot_controller.isolate_args(ctx.message.content) chance: float try: chance = float(chance_str) @@ -1147,23 +1128,3 @@ async def do_true(ctx: commands.Context) -> None: None ) ) - - @staticmethod - def isolate_args(full_cmd_msg: str) -> str: - """ - :param full_cmd_msg: the command message which includes the command name itself - :return: removes the '!command' part of the msg along with exactly one single space after it - """ - - full_cmd_msg = full_cmd_msg.strip() - # ^ Twitch should already do this before getting - # the message, but just done in case they don't - - first_space_at: int = full_cmd_msg.find(" ") - space_found: bool = first_space_at >= 0 - - if not space_found: - return "" - - # won't give 'index out of range' as message can't end on a space due to the strip() - return full_cmd_msg[first_space_at + 1:] diff --git a/src/complements_bot/bot_controller.py b/src/complements_bot/bot_controller.py new file mode 100644 index 0000000..3a4461e --- /dev/null +++ b/src/complements_bot/bot_controller.py @@ -0,0 +1,44 @@ +""" +Holds all logic for how ComplementsBot should complement Twitch chatters +""" +# import asyncio +# from typing import Awaitable, Callable, Optional, Tuple, Union + + +def custom_log(msg: str) -> None: + """ + Any messages which we want to log should be passed through this method + """ + + print(msg) + + +def is_bot(username: str) -> bool: + """ + checks if a username matches that of a known or assumed bot; currently the following count as bots: + - any username ending in 'bot' + - streamlabs + """ + + return (len(username) >= 3 and username[-3:].lower() == 'bot' or + username in ("streamlabs", "streamelements")) + + +def isolate_args(full_cmd_msg: str) -> str: + """ + :param full_cmd_msg: the command message which includes the command name itself + :return: removes the '!command' part of the msg along with exactly one single space after it + """ + + full_cmd_msg = full_cmd_msg.strip() + # ^ Twitch should already do this before getting + # the message, but just done in case they don't + + first_space_at: int = full_cmd_msg.find(" ") + space_found: bool = first_space_at >= 0 + + if not space_found: + return "" + + # won't give 'index out of range' as message can't end on a space due to the strip() + return full_cmd_msg[first_space_at + 1:]