diff --git a/demo.py b/demo.py index bf528da..fb7adae 100644 --- a/demo.py +++ b/demo.py @@ -4,8 +4,13 @@ from dotenv import load_dotenv -from wiring import (Bot, MultiPlatformMessage, MultiPlatformBot, MultiPlatformUser, - Command) +from wiring import ( + Bot, + MultiPlatformMessage, + MultiPlatformBot, + MultiPlatformUser, + Command, +) from wiring.errors.action_not_supported_error import ActionNotSupportedError from wiring.platforms.discord import DiscordBot from wiring.platforms.telegram import TelegramBot @@ -14,22 +19,23 @@ load_dotenv() -logging.basicConfig(level=logging.INFO, - format='%(asctime)s: [%(levelname)s] %(name)s - %(message)s', - datefmt='%m/%d/%Y %I:%M:%S %p') +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s: [%(levelname)s] %(name)s - %(message)s", + datefmt="%m/%d/%Y %I:%M:%S %p", +) logger = logging.getLogger() -async def send_commands_list(bot: Bot, message: MultiPlatformMessage, - args: list[str]): +async def send_commands_list(bot: Bot, message: MultiPlatformMessage, args: list[str]): if message.chat is None: return await bot.send_message( message.chat.id, - 'available commands:\n' + '\n'.join(['/' + str(command.name) for command - in bot.commands]), - reply_message_id=message.id + "available commands:\n" + + "\n".join(["/" + str(command.name) for command in bot.commands]), + reply_message_id=message.id, ) @@ -37,7 +43,7 @@ async def send_greetings(bot: Bot, user: MultiPlatformUser): if user.from_chat_group is not None: for chat in await bot.get_chats_from_group(user.from_chat_group.id): try: - await bot.send_message(chat.id, 'hi ' + user.username) + await bot.send_message(chat.id, "hi " + user.username) except Exception as error: logger.error(error) @@ -47,31 +53,38 @@ async def ban(bot: Bot, message: MultiPlatformMessage, args: list[str]): return if len(args) < 1: - await bot.send_message(message.chat.id, 'specify a name/id of a user', - reply_message_id=message.id) + await bot.send_message( + message.chat.id, "specify a name/id of a user", reply_message_id=message.id + ) return try: - user = await bot.get_user_by_name(args[0].removeprefix('@'), - message.chat_group.id) + user = await bot.get_user_by_name( + args[0].removeprefix("@"), message.chat_group.id + ) if user is not None: await bot.ban(message.chat_group.id, user.id) - await bot.send_message(message.chat.id, 'banned', - reply_message_id=message.id) + await bot.send_message( + message.chat.id, "banned", reply_message_id=message.id + ) else: - await bot.send_message(message.chat.id, 'cant find the user', - reply_message_id=message.id) + await bot.send_message( + message.chat.id, "cant find the user", reply_message_id=message.id + ) except ActionNotSupportedError: - await bot.send_message(message.chat.id, 'banning is not supported here', - reply_message_id=message.id) + await bot.send_message( + message.chat.id, + "banning is not supported here", + reply_message_id=message.id, + ) async def send_goodbye(bot: Bot, user: MultiPlatformUser): if user.from_chat_group is not None: for chat in await bot.get_chats_from_group(user.from_chat_group.id): try: - await bot.send_message(chat.id, 'bye ' + user.username) + await bot.send_message(chat.id, "bye " + user.username) except Exception as error: logger.error(error) @@ -80,20 +93,25 @@ async def start_bots(): bot = MultiPlatformBot() bot.platform_bots = [ - DiscordBot(os.environ['DISCORD_BOT_TOKEN']), - TelegramBot(os.environ['TELEGRAM_BOT_TOKEN']), - TwitchBot(os.environ['TWITCH_BOT_TOKEN'], - streamer_usernames_to_connect=[os.environ['TWITCH_TESTING_CHANNEL']]) + DiscordBot(os.environ["DISCORD_BOT_TOKEN"]), + TelegramBot(os.environ["TELEGRAM_BOT_TOKEN"]), + TwitchBot( + os.environ["TWITCH_BOT_TOKEN"], + streamer_usernames_to_connect=[os.environ["TWITCH_TESTING_CHANNEL"]], + ), ] async with bot: - await bot.setup_commands([ - Command(['start', 'help', 'help1'], send_commands_list), - Command('ban-user', ban) - ], '!') - - bot.add_event_handler('join', send_greetings) - bot.add_event_handler('leave', send_goodbye) + await bot.setup_commands( + [ + Command(["start", "help", "help1"], send_commands_list), + Command("ban-user", ban), + ], + "!", + ) + + bot.add_event_handler("join", send_greetings) + bot.add_event_handler("leave", send_goodbye) await bot.listen_to_events() diff --git a/tests/fixtures/multi_platform_bot.py b/tests/fixtures/multi_platform_bot.py index 3c86160..40f24f8 100644 --- a/tests/fixtures/multi_platform_bot.py +++ b/tests/fixtures/multi_platform_bot.py @@ -4,7 +4,9 @@ from dotenv import load_dotenv import pytest -from tests.errors.missing_environment_variables_error import MissingEnvironmentVariablesError +from tests.errors.missing_environment_variables_error import ( + MissingEnvironmentVariablesError, +) from wiring.multi_platform_bot import MultiPlatformBot from wiring.platforms.discord import DiscordBot from wiring.platforms.telegram import TelegramBot @@ -16,13 +18,13 @@ class NoBotsAddedError(Exception): """at least one platform bot must be added to run tests""" + def __init__(self): - super().__init__('at least one platform bot must be added to run tests') + super().__init__("at least one platform bot must be added to run tests") async def cancel_pending_asyncio_tasks(event_loop): - pending_tasks = [task for task in asyncio.all_tasks(event_loop) - if not task.done()] + pending_tasks = [task for task in asyncio.all_tasks(event_loop) if not task.done()] for task in pending_tasks: task.cancel() @@ -33,29 +35,31 @@ async def cancel_pending_asyncio_tasks(event_loop): pass -@pytest.fixture(scope='session') +@pytest.fixture(scope="session") async def multi_platform_bot(): bot = MultiPlatformBot() - if 'DISCORD_BOT_TOKEN' in os.environ: - bot.platform_bots.append(DiscordBot(os.environ['DISCORD_BOT_TOKEN'])) + if "DISCORD_BOT_TOKEN" in os.environ: + bot.platform_bots.append(DiscordBot(os.environ["DISCORD_BOT_TOKEN"])) - if 'TELEGRAM_BOT_TOKEN' in os.environ: - bot.platform_bots.append(TelegramBot(os.environ['TELEGRAM_BOT_TOKEN'])) + if "TELEGRAM_BOT_TOKEN" in os.environ: + bot.platform_bots.append(TelegramBot(os.environ["TELEGRAM_BOT_TOKEN"])) - if 'TWITCH_BOT_TOKEN' in os.environ: - testing_channel = os.environ.get('TWITCH_TESTING_CHANNEL') + if "TWITCH_BOT_TOKEN" in os.environ: + testing_channel = os.environ.get("TWITCH_TESTING_CHANNEL") if testing_channel is None: raise MissingEnvironmentVariablesError( - 'to test twitch bot, you must specify a twitch channel id ' - + 'for testing like that:\nTWITCH_TESTING_CHANNEL=' + "to test twitch bot, you must specify a twitch channel id " + + "for testing like that:\nTWITCH_TESTING_CHANNEL=" ) - bot.platform_bots.append(TwitchBot( - os.environ['TWITCH_BOT_TOKEN'], - streamer_usernames_to_connect=[testing_channel] - )) + bot.platform_bots.append( + TwitchBot( + os.environ["TWITCH_BOT_TOKEN"], + streamer_usernames_to_connect=[testing_channel], + ) + ) if len(bot.platform_bots) == 0: raise NoBotsAddedError() diff --git a/tests/test_discord_bot_actions.py b/tests/test_discord_bot_actions.py index dc5296f..ba22bbf 100644 --- a/tests/test_discord_bot_actions.py +++ b/tests/test_discord_bot_actions.py @@ -10,38 +10,44 @@ logger = logging.getLogger() -@pytest.mark.asyncio(scope='session') +@pytest.mark.asyncio(scope="session") async def test_message_sending(multi_platform_bot: MultiPlatformBot): - discord_bots = [bot for bot in multi_platform_bot.platform_bots - if bot.platform == 'discord'] + discord_bots = [ + bot for bot in multi_platform_bot.platform_bots if bot.platform == "discord" + ] if len(discord_bots) == 0: - pytest.skip('skipping discord bot actions testing because it\'s not added ' - + 'via DISCORD_BOT_TOKEN environment variable') + pytest.skip( + "skipping discord bot actions testing because it's not added " + + "via DISCORD_BOT_TOKEN environment variable" + ) - guilds = await multi_platform_bot.get_chat_groups('discord') + guilds = await multi_platform_bot.get_chat_groups("discord") if len(guilds) == 0: - raise UnusableBotError('current discord bot must be at least in one guild to ' - + 'run this test') + raise UnusableBotError( + "current discord bot must be at least in one guild to " + "run this test" + ) - channels = await multi_platform_bot.get_chats_from_group({'platform': 'discord', - 'value': guilds[0].id}) + channels = await multi_platform_bot.get_chats_from_group( + {"platform": "discord", "value": guilds[0].id} + ) assert len(channels) > 0 guild_has_messageable_channels = False for channel in channels: try: - await multi_platform_bot.send_message({ - 'discord': channel.id - }, 'test message') + await multi_platform_bot.send_message( + {"discord": channel.id}, "test message" + ) guild_has_messageable_channels = True break except NotMessageableChatError: - logger.debug(f'channel \'{channel.name}\' cant be messaged') + logger.debug(f"channel '{channel.name}' cant be messaged") if not guild_has_messageable_channels: - raise UnusableBotError('current discord bot first guild must have a ' - + 'messageable channel') + raise UnusableBotError( + "current discord bot first guild must have a " + "messageable channel" + ) diff --git a/tests/test_telegram_bot_actions.py b/tests/test_telegram_bot_actions.py index dd427e0..e17dbf6 100644 --- a/tests/test_telegram_bot_actions.py +++ b/tests/test_telegram_bot_actions.py @@ -1,28 +1,35 @@ import os import pytest -from tests.errors.missing_environment_variables_error import MissingEnvironmentVariablesError +from tests.errors.missing_environment_variables_error import ( + MissingEnvironmentVariablesError, +) from wiring.multi_platform_bot import MultiPlatformBot -@pytest.mark.asyncio(scope='session') +@pytest.mark.asyncio(scope="session") async def test_message_sending(multi_platform_bot: MultiPlatformBot): - telegram_bots = [bot for bot in multi_platform_bot.platform_bots - if bot.platform == 'telegram'] + telegram_bots = [ + bot for bot in multi_platform_bot.platform_bots if bot.platform == "telegram" + ] if len(telegram_bots) == 0: - pytest.skip('skipping telegram bot actions testing because it\'s not added ' - + 'via TELEGRAM_BOT_TOKEN environment variable') + pytest.skip( + "skipping telegram bot actions testing because it's not added " + + "via TELEGRAM_BOT_TOKEN environment variable" + ) - testing_chat_id = os.environ.get('TELEGRAM_TESTING_CHAT_ID') + testing_chat_id = os.environ.get("TELEGRAM_TESTING_CHAT_ID") if testing_chat_id is None: - raise MissingEnvironmentVariablesError('it\'s not possible to get current ' - + 'bot chats with telegram api. so ' - + 'you must specify some testing chat id ' - + 'in `.env` file like that: ' - + 'TELEGRAM_TESTING_CHAT_ID=') + raise MissingEnvironmentVariablesError( + "it's not possible to get current " + + "bot chats with telegram api. so " + + "you must specify some testing chat id " + + "in `.env` file like that: " + + "TELEGRAM_TESTING_CHAT_ID=" + ) - await multi_platform_bot.send_message({ - 'telegram': int(testing_chat_id) - }, 'test message') + await multi_platform_bot.send_message( + {"telegram": int(testing_chat_id)}, "test message" + ) diff --git a/tests/test_twitch_bot_actions.py b/tests/test_twitch_bot_actions.py index e137c33..93dc254 100644 --- a/tests/test_twitch_bot_actions.py +++ b/tests/test_twitch_bot_actions.py @@ -3,19 +3,20 @@ from wiring.multi_platform_bot import MultiPlatformBot -@pytest.mark.asyncio(scope='session') +@pytest.mark.asyncio(scope="session") async def test_message_sending(multi_platform_bot: MultiPlatformBot): - twitch_bots = [bot for bot in multi_platform_bot.platform_bots - if bot.platform == 'twitch'] + twitch_bots = [ + bot for bot in multi_platform_bot.platform_bots if bot.platform == "twitch" + ] if len(twitch_bots) == 0: - pytest.skip('skipping twitch bot actions testing because it\'s not added ' - + 'via TWITCH_BOT_TOKEN environment variable') + pytest.skip( + "skipping twitch bot actions testing because it's not added " + + "via TWITCH_BOT_TOKEN environment variable" + ) - channels = await multi_platform_bot.get_chat_groups('twitch') + channels = await multi_platform_bot.get_chat_groups("twitch") assert len(channels) > 0 - await multi_platform_bot.send_message({ - 'twitch': channels[0].id - }, 'test message') + await multi_platform_bot.send_message({"twitch": channels[0].id}, "test message") diff --git a/wiring/__init__.py b/wiring/__init__.py index 501e892..c348bce 100644 --- a/wiring/__init__.py +++ b/wiring/__init__.py @@ -1,6 +1,12 @@ from wiring.multi_platform_bot import MultiPlatformBot, PlatformBotNotFoundError -from wiring.multi_platform_resources import (PlatformSpecificValue, MultiPlatformChat, - MultiPlatformChatGroup, MultiPlatformId, - MultiPlatformMessage, MultiPlatformUser, - MultiPlatformValue, Platform) +from wiring.multi_platform_resources import ( + PlatformSpecificValue, + MultiPlatformChat, + MultiPlatformChatGroup, + MultiPlatformId, + MultiPlatformMessage, + MultiPlatformUser, + MultiPlatformValue, + Platform, +) from wiring.bot_base import Bot, Command, CommandHandler diff --git a/wiring/_to_multi_platform_converter.py b/wiring/_to_multi_platform_converter.py index dc6a470..4323145 100644 --- a/wiring/_to_multi_platform_converter.py +++ b/wiring/_to_multi_platform_converter.py @@ -1,9 +1,11 @@ from abc import ABC, abstractmethod -from wiring.multi_platform_resources import (MultiPlatformChatGroup, - MultiPlatformMessage, - MultiPlatformChat, - MultiPlatformUser) +from wiring.multi_platform_resources import ( + MultiPlatformChatGroup, + MultiPlatformMessage, + MultiPlatformChat, + MultiPlatformUser, +) class ToMultiPlatformConverter(ABC): @@ -16,7 +18,9 @@ def convert_to_multi_platform_user(self, user) -> MultiPlatformUser: pass @abstractmethod - def convert_to_multi_platform_chat_group(self, chat_group) -> MultiPlatformChatGroup: + def convert_to_multi_platform_chat_group( + self, chat_group + ) -> MultiPlatformChatGroup: pass @abstractmethod diff --git a/wiring/_utils/find.py b/wiring/_utils/find.py index d69d333..383fe66 100644 --- a/wiring/_utils/find.py +++ b/wiring/_utils/find.py @@ -1,7 +1,7 @@ from typing import Callable, TypeVar -ItemT = TypeVar('ItemT') +ItemT = TypeVar("ItemT") def find_item(items: list[ItemT], does_item_match: Callable[[ItemT], bool]): diff --git a/wiring/bot_base.py b/wiring/bot_base.py index 77dd5cb..cee59d3 100644 --- a/wiring/bot_base.py +++ b/wiring/bot_base.py @@ -4,14 +4,18 @@ from io import BufferedReader from typing import Any, Callable, Coroutine, Literal, Optional, Awaitable -from wiring.multi_platform_resources import (MultiPlatformChatGroup, MultiPlatformMessage, - MultiPlatformChat, MultiPlatformUser, - Platform) +from wiring.multi_platform_resources import ( + MultiPlatformChatGroup, + MultiPlatformMessage, + MultiPlatformChat, + MultiPlatformUser, + Platform, +) CommandHandler = Callable[[Any, MultiPlatformMessage, list[str]], Coroutine] -Event = Literal['message', 'join', 'leave'] +Event = Literal["message", "join", "leave"] @dataclass @@ -27,10 +31,9 @@ class EventHandler: class Bot(ABC): - def __init__(self, - platform: Optional[Platform] = None): + def __init__(self, platform: Optional[Platform] = None): self.platform = platform - self.commands_prefix = '/' + self.commands_prefix = "/" self.commands = [] self.event_listening_coroutine: Optional[Awaitable] = None @@ -46,10 +49,13 @@ async def stop(self): pass @abstractmethod - async def send_message(self, chat_id, - text: str, - reply_message_id: Any = None, - files: Optional[list[BufferedReader]] = None): + async def send_message( + self, + chat_id, + text: str, + reply_message_id: Any = None, + files: Optional[list[BufferedReader]] = None, + ): """sends message in the chat Args: @@ -104,8 +110,13 @@ async def get_chats_from_group(self, chat_group_id) -> list[MultiPlatformChat]: """ @abstractmethod - async def ban(self, chat_group_id, user_id, reason: Optional[str] = None, - seconds_duration: Optional[int] = None): + async def ban( + self, + chat_group_id, + user_id, + reason: Optional[str] = None, + seconds_duration: Optional[int] = None, + ): """bans the user from the specified chat group Args: @@ -123,9 +134,9 @@ async def ban(self, chat_group_id, user_id, reason: Optional[str] = None, """ @abstractmethod - async def get_user_by_name(self, - username, - chat_group_id) -> Optional[MultiPlatformUser]: + async def get_user_by_name( + self, username, chat_group_id + ) -> Optional[MultiPlatformUser]: """get user that takes part in specified chat group by username Args: @@ -151,7 +162,7 @@ async def delete_messages(self, chat_id, *messages_ids): example if you dont have the permission to delete specific message """ - async def setup_commands(self, commands: list[Command], prefix: str = '/'): + async def setup_commands(self, commands: list[Command], prefix: str = "/"): self.commands_prefix = prefix self.commands = commands @@ -184,21 +195,30 @@ def has_command(text: str, command: Command): cleaned_text = text.removeprefix(self.commands_prefix).strip().casefold() if isinstance(command.name, list): - return len([name for name in command.name - if cleaned_text == name.casefold()]) > 0 + return ( + len( + [ + name + for name in command.name + if cleaned_text == name.casefold() + ] + ) + > 0 + ) return command.name.casefold() == cleaned_text if not message.text or not message.text.startswith(self.commands_prefix): return - message_parts = message.text.split(' ') + message_parts = message.text.split(" ") if len(message_parts) == 0: return matched_commands = [ - some_command for some_command in self.commands + some_command + for some_command in self.commands if has_command(message_parts[0], some_command) ] diff --git a/wiring/errors/bot_api_error.py b/wiring/errors/bot_api_error.py index f907259..e22441e 100644 --- a/wiring/errors/bot_api_error.py +++ b/wiring/errors/bot_api_error.py @@ -11,8 +11,13 @@ class BotApiError(Exception): Attributes: status_code: http error status code """ - def __init__(self, platform: Platform, explanation: Optional[str] = None, - status_code: Optional[int] = None): + + def __init__( + self, + platform: Platform, + explanation: Optional[str] = None, + status_code: Optional[int] = None, + ): """creates common http error that occurred on some platform api interaction Args: @@ -22,12 +27,12 @@ def __init__(self, platform: Platform, explanation: Optional[str] = None, self.explanation = explanation self.status_code = status_code - error_text = f'failed to perform some action through {platform} api' + error_text = f"failed to perform some action through {platform} api" if explanation is not None: - error_text = explanation + f'. platform: {platform}' + error_text = explanation + f". platform: {platform}" if status_code is not None: - error_text += f', status code: {status_code}' + error_text += f", status code: {status_code}" super().__init__(error_text) diff --git a/wiring/errors/not_found_error.py b/wiring/errors/not_found_error.py index 254d83c..3736c22 100644 --- a/wiring/errors/not_found_error.py +++ b/wiring/errors/not_found_error.py @@ -7,5 +7,6 @@ class NotFoundError(BotApiError): subclass of `BotApiError` """ + def __init__(self, platform: Platform, explanation: str): super().__init__(platform, explanation, 404) diff --git a/wiring/errors/not_messageable_chat_error.py b/wiring/errors/not_messageable_chat_error.py index aa50f88..c191aa5 100644 --- a/wiring/errors/not_messageable_chat_error.py +++ b/wiring/errors/not_messageable_chat_error.py @@ -4,6 +4,6 @@ class NotMessageableChatError(Exception): def __init__(self, platform: Platform, chat_id=None): if chat_id is not None: - super().__init__(f'cant send message in chat \'{chat_id}\' on {platform}') + super().__init__(f"cant send message in chat '{chat_id}' on {platform}") else: - super().__init__(f'cant send message in specified chat on {platform}') + super().__init__(f"cant send message in specified chat on {platform}") diff --git a/wiring/multi_platform_bot.py b/wiring/multi_platform_bot.py index 045c27c..370038c 100644 --- a/wiring/multi_platform_bot.py +++ b/wiring/multi_platform_bot.py @@ -2,13 +2,16 @@ from typing import Optional from wiring.bot_base import Bot, Command, Event -from wiring.multi_platform_resources import (MultiPlatformValue, Platform, - PlatformSpecificValue) +from wiring.multi_platform_resources import ( + MultiPlatformValue, + Platform, + PlatformSpecificValue, +) class PlatformBotNotFoundError(Exception): def __init__(self, requested_platform: str): - super().__init__(f'bot with platform \'{requested_platform}\' was not added') + super().__init__(f"bot with platform '{requested_platform}' was not added") class MultiPlatformBot(Bot): @@ -40,14 +43,15 @@ class MultiPlatformBot(Bot): text='test message') ``` """ + def __init__(self): super().__init__() self.platform_bots: list[Bot] = [] - self.logger = logging.getLogger('wiring.multi_platform') + self.logger = logging.getLogger("wiring.multi_platform") async def start(self): - self.logger.info('started') + self.logger.info("started") for bot in self.platform_bots: await bot.start() @@ -60,9 +64,13 @@ async def listen_to_events(self): if bot.event_listening_coroutine is not None: await bot.event_listening_coroutine - async def send_message(self, chat_id: MultiPlatformValue, text: str, - reply_message_id: Optional[MultiPlatformValue] = None, - files: Optional[list] = None): + async def send_message( + self, + chat_id: MultiPlatformValue, + text: str, + reply_message_id: Optional[MultiPlatformValue] = None, + files: Optional[list] = None, + ): for bot in self.platform_bots: if bot.platform not in chat_id: continue @@ -74,32 +82,37 @@ async def send_message(self, chat_id: MultiPlatformValue, text: str, platform_reply_message_id = reply_message_id.get(bot.platform) if platform_chat_id is not None: - self.logger.info(f'sending message to chat \'{platform_chat_id}\' ' - + f'on \'{bot.platform}\'') + self.logger.info( + f"sending message to chat '{platform_chat_id}' " + + f"on '{bot.platform}'" + ) - await bot.send_message(platform_chat_id, - text, - platform_reply_message_id, - files) + await bot.send_message( + platform_chat_id, text, platform_reply_message_id, files + ) async def get_chat_groups(self, on_platform=None): if on_platform is None: - raise ValueError('param `on_platform` must be specified when using ' - + '`MultiPlatformBot` class') + raise ValueError( + "param `on_platform` must be specified when using " + + "`MultiPlatformBot` class" + ) needed_bots = self.__get_bots_on_platform(on_platform) return await needed_bots[0].get_chat_groups() async def get_chats_from_group(self, chat_group_id: PlatformSpecificValue): - needed_bots = self.__get_bots_on_platform(chat_group_id['platform']) + needed_bots = self.__get_bots_on_platform(chat_group_id["platform"]) - return await needed_bots[0].get_chats_from_group(chat_group_id['value']) + return await needed_bots[0].get_chats_from_group(chat_group_id["value"]) - async def ban(self, - chat_group_id: MultiPlatformValue, - user_id: MultiPlatformValue, - reason=None, - seconds_duration=None): + async def ban( + self, + chat_group_id: MultiPlatformValue, + user_id: MultiPlatformValue, + reason=None, + seconds_duration=None, + ): for bot in self.platform_bots: if bot.platform not in chat_group_id or bot.platform not in user_id: continue @@ -107,42 +120,46 @@ async def ban(self, platform_chat_group_id = chat_group_id[bot.platform] platform_user_id = user_id[bot.platform] - await bot.ban(platform_chat_group_id, platform_user_id, reason, - seconds_duration) + await bot.ban( + platform_chat_group_id, platform_user_id, reason, seconds_duration + ) - async def get_user_by_name(self, - username: PlatformSpecificValue, - chat_group_id: PlatformSpecificValue): - platform_bot = self.__get_bots_on_platform(username['platform'])[0] + async def get_user_by_name( + self, username: PlatformSpecificValue, chat_group_id: PlatformSpecificValue + ): + platform_bot = self.__get_bots_on_platform(username["platform"])[0] - return await platform_bot.get_user_by_name(username['value'], - chat_group_id['value']) + return await platform_bot.get_user_by_name( + username["value"], chat_group_id["value"] + ) - async def delete_messages(self, - chat_id: MultiPlatformValue, - *messages_ids: MultiPlatformValue): + async def delete_messages( + self, chat_id: MultiPlatformValue, *messages_ids: MultiPlatformValue + ): for bot in self.platform_bots: if bot.platform not in chat_id or bot.platform not in messages_ids: continue - self.logger.info(f'deleting message in chat \'{chat_id}\' ' - + 'on \'{bot.platform}\'') - await bot.delete_messages(chat_id[bot.platform], *messages_ids[bot.platform]) + self.logger.info( + f"deleting message in chat '{chat_id}' " + "on '{bot.platform}'" + ) + await bot.delete_messages( + chat_id[bot.platform], *messages_ids[bot.platform] + ) def add_event_handler(self, event: Event, handler): super().add_event_handler(event, handler) for bot in self.platform_bots: bot.add_event_handler(event, handler) - async def setup_commands(self, commands: list[Command], prefix: str = '/'): + async def setup_commands(self, commands: list[Command], prefix: str = "/"): for bot in self.platform_bots: await bot.setup_commands(commands, prefix) await super().setup_commands(commands, prefix) def __get_bots_on_platform(self, platform: Platform): - needed_bots = [bot for bot in self.platform_bots - if bot.platform == platform] + needed_bots = [bot for bot in self.platform_bots if bot.platform == platform] if len(needed_bots) == 0: raise PlatformBotNotFoundError(platform) diff --git a/wiring/multi_platform_resources.py b/wiring/multi_platform_resources.py index 1cc8724..48cf469 100644 --- a/wiring/multi_platform_resources.py +++ b/wiring/multi_platform_resources.py @@ -2,7 +2,7 @@ from typing import Any, Literal, Optional, TypedDict -Platform = Literal['discord', 'telegram', 'twitch'] +Platform = Literal["discord", "telegram", "twitch"] MultiPlatformValue = dict[Platform, Any] @@ -23,6 +23,7 @@ class MultiPlatformChatGroup: if platform doesnt support chat groups, they are considered identical to chats """ + platform: Platform id: MultiPlatformId name: Optional[str] @@ -33,6 +34,7 @@ class MultiPlatformChat: """ telegram chat, discord channel or other platform chat where message was sent """ + platform: Platform id: MultiPlatformId name: Optional[str] @@ -47,7 +49,7 @@ class MultiPlatformUser: @dataclass -class MultiPlatformMessage(): +class MultiPlatformMessage: platform: Platform id: MultiPlatformId chat_group: Optional[MultiPlatformChatGroup] diff --git a/wiring/platforms/discord/__init__.py b/wiring/platforms/discord/__init__.py index dece88a..59b6e56 100644 --- a/wiring/platforms/discord/__init__.py +++ b/wiring/platforms/discord/__init__.py @@ -13,9 +13,7 @@ class CustomClient(discord.Client): - def __init__(self, - intents: discord.Intents, - event_handlers: dict[str, Callable]): + def __init__(self, intents: discord.Intents, event_handlers: dict[str, Callable]): super().__init__(intents=intents) self._event_handlers = event_handlers @@ -24,33 +22,35 @@ async def on_message(self, message: discord.Message): return self.__run_event_handler_if_exists( - 'all', 'message', discord_entities_converter.convert_to_multi_platform_message(message) + "all", + "message", + discord_entities_converter.convert_to_multi_platform_message(message), ) self.__run_event_handler_if_exists( - 'message', discord_entities_converter.convert_to_multi_platform_message(message) + "message", + discord_entities_converter.convert_to_multi_platform_message(message), ) async def on_member_join(self, member: discord.Member): self.__run_event_handler_if_exists( - 'all', 'join', discord_entities_converter.convert_to_multi_platform_user(member) + "all", + "join", + discord_entities_converter.convert_to_multi_platform_user(member), ) self.__run_event_handler_if_exists( - 'join', discord_entities_converter.convert_to_multi_platform_user(member) + "join", discord_entities_converter.convert_to_multi_platform_user(member) ) async def on_member_remove(self, member: discord.Member): - multi_platform_user = (discord_entities_converter - .convert_to_multi_platform_user(member)) - - self.__run_event_handler_if_exists( - 'all', 'leave', multi_platform_user + multi_platform_user = discord_entities_converter.convert_to_multi_platform_user( + member ) - self.__run_event_handler_if_exists( - 'leave', multi_platform_user - ) + self.__run_event_handler_if_exists("all", "leave", multi_platform_user) + + self.__run_event_handler_if_exists("leave", multi_platform_user) def __run_event_handler_if_exists(self, event: str, *args): do_on_event = self._event_handlers.get(event) @@ -61,19 +61,22 @@ def __run_event_handler_if_exists(self, event: str, *args): class DiscordBot(Bot): def __init__(self, token: str): - super().__init__('discord') + super().__init__("discord") intents = discord.Intents.default() intents.message_content = True intents.members = True - self.client = CustomClient(intents, { - 'all': lambda event, data: self._run_event_handlers(event, data), - 'message': self._check_message_for_command, - }) + self.client = CustomClient( + intents, + { + "all": lambda event, data: self._run_event_handlers(event, data), + "message": self._check_message_for_command, + }, + ) self._token = token - self.logger = logging.getLogger('wiring.discord') + self.logger = logging.getLogger("wiring.discord") async def start(self): await self.client.login(self._token) @@ -82,13 +85,17 @@ async def start(self): async def stop(self): await self.client.close() - async def send_message(self, chat_id: int, text: str, - reply_message_id: Optional[int] = None, - files: Optional[list] = None): + async def send_message( + self, + chat_id: int, + text: str, + reply_message_id: Optional[int] = None, + files: Optional[list] = None, + ): channel: Any = await self.client.fetch_channel(chat_id) - if not hasattr(channel, 'send'): - raise NotMessageableChatError('discord', channel.id) + if not hasattr(channel, "send"): + raise NotMessageableChatError("discord", channel.id) files = [discord.File(file) for file in files or []] @@ -99,9 +106,9 @@ async def send_message(self, chat_id: int, text: str, else: await channel.send(text, files=files) except discord.NotFound as discord_not_found_error: - raise NotFoundError('discord', discord_not_found_error.text) + raise NotFoundError("discord", discord_not_found_error.text) except discord.HTTPException as discord_error: - raise BotApiError('discord', discord_error.text, discord_error.status) + raise BotApiError("discord", discord_error.text, discord_error.status) async def get_chat_groups(self, on_platform=None): guilds = [guild async for guild in self.client.fetch_guilds(limit=None)] @@ -113,65 +120,73 @@ async def get_chat_groups(self, on_platform=None): async def get_chats_from_group(self, chat_group_id: int): try: - channels = await (await self.client.fetch_guild(chat_group_id)).fetch_channels() - return [discord_entities_converter.convert_to_multi_platform_chat(channel) - for channel in channels] + channels = await ( + await self.client.fetch_guild(chat_group_id) + ).fetch_channels() + return [ + discord_entities_converter.convert_to_multi_platform_chat(channel) + for channel in channels + ] except discord.NotFound as discord_not_found_error: - raise NotFoundError('discord', discord_not_found_error.text) + raise NotFoundError("discord", discord_not_found_error.text) except discord.HTTPException as discord_http_error: - raise BotApiError('discord', discord_http_error.text, - discord_http_error.status) + raise BotApiError( + "discord", discord_http_error.text, discord_http_error.status + ) except discord.errors.InvalidData: - raise BotApiError('discord', 'received invalid data from api') + raise BotApiError("discord", "received invalid data from api") - async def ban(self, - chat_group_id: int, - user_id: int, - reason=None, - seconds_duration=None): + async def ban( + self, chat_group_id: int, user_id: int, reason=None, seconds_duration=None + ): try: guild = await self.client.fetch_guild(chat_group_id) member_to_ban = await guild.fetch_member(user_id) if seconds_duration is not None: - await member_to_ban.timeout(datetime.timedelta(seconds=seconds_duration), - reason=reason) + await member_to_ban.timeout( + datetime.timedelta(seconds=seconds_duration), reason=reason + ) else: await member_to_ban.ban(reason=reason) except discord.NotFound as discord_not_found_error: - raise NotFoundError('discord', discord_not_found_error.text) + raise NotFoundError("discord", discord_not_found_error.text) except discord.HTTPException as discord_http_error: - raise BotApiError('discord', discord_http_error.text, - discord_http_error.status) + raise BotApiError( + "discord", discord_http_error.text, discord_http_error.status + ) async def get_user_by_name(self, username: str, chat_group_id: int): try: guild = await self.client.fetch_guild(chat_group_id) - member = await discord.utils.get(guild.fetch_members(limit=None), - name=username) + member = await discord.utils.get( + guild.fetch_members(limit=None), name=username + ) if member is None: return None return discord_entities_converter.convert_to_multi_platform_user(member) except discord.NotFound as discord_not_found_error: - raise NotFoundError('discord', discord_not_found_error.text) + raise NotFoundError("discord", discord_not_found_error.text) except discord.HTTPException as discord_http_error: - raise BotApiError('discord', discord_http_error.text, - discord_http_error.status) + raise BotApiError( + "discord", discord_http_error.text, discord_http_error.status + ) async def delete_messages(self, chat_id: int, *messages_ids: int): try: channel: Any = await self.client.fetch_channel(chat_id) - if not hasattr(channel, 'fetch_message'): - raise NotMessageableChatError('discord', channel.id) + if not hasattr(channel, "fetch_message"): + raise NotMessageableChatError("discord", channel.id) for message_id in messages_ids: message = await channel.fetch_message(message_id) await message.delete() except discord.NotFound as discord_not_found_error: - raise NotFoundError('discord', discord_not_found_error.text) + raise NotFoundError("discord", discord_not_found_error.text) except discord.HTTPException as discord_http_error: - raise BotApiError('discord', discord_http_error.text, - discord_http_error.status) + raise BotApiError( + "discord", discord_http_error.text, discord_http_error.status + ) diff --git a/wiring/platforms/discord/_entities_converter.py b/wiring/platforms/discord/_entities_converter.py index 57dacb8..dc77d0d 100644 --- a/wiring/platforms/discord/_entities_converter.py +++ b/wiring/platforms/discord/_entities_converter.py @@ -1,50 +1,53 @@ import discord -from wiring.multi_platform_resources import (MultiPlatformChatGroup, MultiPlatformMessage, - MultiPlatformChat, MultiPlatformUser) +from wiring.multi_platform_resources import ( + MultiPlatformChatGroup, + MultiPlatformMessage, + MultiPlatformChat, + MultiPlatformUser, +) from wiring.platforms.discord.channels import MessageableChannel from wiring._to_multi_platform_converter import ToMultiPlatformConverter class DiscordEntitiesConverter(ToMultiPlatformConverter): def convert_to_multi_platform_chat_group(self, chat_group: discord.Guild): - return MultiPlatformChatGroup('discord', chat_group.id, chat_group.name) + return MultiPlatformChatGroup("discord", chat_group.id, chat_group.name) - def convert_to_multi_platform_chat( - self, chat: MessageableChannel - ): + def convert_to_multi_platform_chat(self, chat: MessageableChannel): name = None - if (not isinstance(chat, discord.PartialMessageable) - and not isinstance(chat, discord.DMChannel)): + if not isinstance(chat, discord.PartialMessageable) and not isinstance( + chat, discord.DMChannel + ): name = chat.name - return MultiPlatformChat('discord', - chat.id, - name) + return MultiPlatformChat("discord", chat.id, name) def convert_to_multi_platform_user(self, user: discord.User | discord.Member): if isinstance(user, discord.Member): - return MultiPlatformUser('discord', user.id, user.name, - self.convert_to_multi_platform_chat_group(user.guild)) + return MultiPlatformUser( + "discord", + user.id, + user.name, + self.convert_to_multi_platform_chat_group(user.guild), + ) - return MultiPlatformUser('discord', user.id, user.name, None) + return MultiPlatformUser("discord", user.id, user.name, None) - def convert_to_multi_platform_message( - self, message: discord.Message - ): + def convert_to_multi_platform_message(self, message: discord.Message): chat = None if message.guild is not None: chat = self.convert_to_multi_platform_chat_group(message.guild) return MultiPlatformMessage( - 'discord', + "discord", message.id, chat, self.convert_to_multi_platform_chat(message.channel), message.content, - self.convert_to_multi_platform_user(message.author) + self.convert_to_multi_platform_user(message.author), ) diff --git a/wiring/platforms/discord/channels.py b/wiring/platforms/discord/channels.py index 89c55a8..0bc120f 100644 --- a/wiring/platforms/discord/channels.py +++ b/wiring/platforms/discord/channels.py @@ -3,8 +3,14 @@ from typing import Union -PartialMessageableChannel = Union[discord.TextChannel, discord.VoiceChannel, - discord.StageChannel, discord.Thread, - discord.DMChannel, discord.PartialMessageable, - discord.CategoryChannel, discord.ForumChannel] +PartialMessageableChannel = Union[ + discord.TextChannel, + discord.VoiceChannel, + discord.StageChannel, + discord.Thread, + discord.DMChannel, + discord.PartialMessageable, + discord.CategoryChannel, + discord.ForumChannel, +] MessageableChannel = Union[PartialMessageableChannel, discord.GroupChannel] diff --git a/wiring/platforms/telegram/__init__.py b/wiring/platforms/telegram/__init__.py index 0008a7d..3cc4af4 100644 --- a/wiring/platforms/telegram/__init__.py +++ b/wiring/platforms/telegram/__init__.py @@ -6,8 +6,14 @@ from telegram.ext import ApplicationBuilder, MessageHandler, ChatMemberHandler from telegram.error import TelegramError -from telegram import (InputFile, InputMediaAudio, InputMediaDocument, - InputMediaPhoto, InputMediaVideo, Update) +from telegram import ( + InputFile, + InputMediaAudio, + InputMediaDocument, + InputMediaPhoto, + InputMediaVideo, + Update, +) from wiring.bot_base import Bot from wiring.errors.action_not_supported_error import ActionNotSupportedError @@ -17,47 +23,47 @@ class TelegramBot(Bot): def __init__(self, token: str): - super().__init__('telegram') + super().__init__("telegram") self.client = ApplicationBuilder().token(token).build() self.__setup_event_handlers() - self.logger = logging.getLogger('wiring.telegram') + self.logger = logging.getLogger("wiring.telegram") def __setup_event_handlers(self): # mess, need to move it to a different module/class async def handle_update(update: Update, context): if update.message is not None: multi_platform_chat = ( - telegram_entities_converter - .convert_to_multi_platform_chat_group(update.message.chat) + telegram_entities_converter.convert_to_multi_platform_chat_group( + update.message.chat + ) ) multi_platform_message = ( - telegram_entities_converter - .convert_to_multi_platform_message(update.message) + telegram_entities_converter.convert_to_multi_platform_message( + update.message + ) ) - self._run_event_handlers('message', multi_platform_message) + self._run_event_handlers("message", multi_platform_message) self._check_message_for_command(multi_platform_message) for new_member in update.message.new_chat_members or []: self._run_event_handlers( - 'join', + "join", telegram_entities_converter.convert_to_multi_platform_user( - new_member, - multi_platform_chat - ) + new_member, multi_platform_chat + ), ) if update.message.left_chat_member is not None: self._run_event_handlers( - 'leave', + "leave", telegram_entities_converter.convert_to_multi_platform_user( - update.message.left_chat_member, - multi_platform_chat - ) + update.message.left_chat_member, multi_platform_chat + ), ) # registering the same handler for each needed update @@ -69,8 +75,10 @@ async def start(self): await self.client.initialize() if self.client.updater is None: - raise Exception('cant start polling in telegram bot.' - + '\'client.updater\' attribute is \'None\'') + raise Exception( + "cant start polling in telegram bot." + + "'client.updater' attribute is 'None'" + ) try: self.event_listening_coroutine = asyncio.create_task( @@ -86,30 +94,37 @@ async def stop(self): await self.client.stop() await self.client.shutdown() - async def send_message(self, chat_id, text: str, - reply_message_id=None, - files: Optional[list[BufferedReader]] = None): + async def send_message( + self, + chat_id, + text: str, + reply_message_id=None, + files: Optional[list[BufferedReader]] = None, + ): try: if files is not None: await self.client.bot.send_media_group( chat_id, [self.__convert_stream_to_telegram_media(file) for file in files], caption=text, - reply_to_message_id=reply_message_id + reply_to_message_id=reply_message_id, ) return - await self.client.bot.send_message(chat_id, text, - reply_to_message_id=reply_message_id) + await self.client.bot.send_message( + chat_id, text, reply_to_message_id=reply_message_id + ) except TelegramError as telegram_error: - raise BotApiError('telegram', telegram_error.message) + raise BotApiError("telegram", telegram_error.message) async def get_chat_groups(self, on_platform=None): - raise ActionNotSupportedError('it seems telegram api does not permit to get ' - + 'all chats your bot are member of \n' - + 'what you can do is to keep track of chats ' - + 'bot is invited to or is removed from in ' - + 'some database with events') + raise ActionNotSupportedError( + "it seems telegram api does not permit to get " + + "all chats your bot are member of \n" + + "what you can do is to keep track of chats " + + "bot is invited to or is removed from in " + + "some database with events" + ) async def get_chats_from_group(self, chat_group_id: int): try: @@ -119,56 +134,63 @@ async def get_chats_from_group(self, chat_group_id: int): ) ] except TelegramError as telegram_error: - raise BotApiError('telegram', telegram_error.message) + raise BotApiError("telegram", telegram_error.message) - async def ban(self, - chat_group_id: int, - user_id: int, - reason=None, - seconds_duration=None): + async def ban( + self, chat_group_id: int, user_id: int, reason=None, seconds_duration=None + ): try: if reason is not None: - self.logger.warning('ignoring `reason` param for `Bot.ban` method, ' - + 'as it\'s not supported in telegram') + self.logger.warning( + "ignoring `reason` param for `Bot.ban` method, " + + "as it's not supported in telegram" + ) until_date = None if seconds_duration is not None: - until_date = (datetime.datetime.now() - + datetime.timedelta(seconds=seconds_duration)) + until_date = datetime.datetime.now() + datetime.timedelta( + seconds=seconds_duration + ) - await self.client.bot.ban_chat_member(chat_group_id, user_id, - until_date=until_date) + await self.client.bot.ban_chat_member( + chat_group_id, user_id, until_date=until_date + ) except TelegramError as telegram_error: - raise BotApiError('telegram', telegram_error.message) + raise BotApiError("telegram", telegram_error.message) async def get_user_by_name(self, username: str, chat_group_id: int): - raise ActionNotSupportedError('getting users by their usernames is not ' - + 'possible on telegram. to be more precise, ' - + 'it is impossible to get all users from ' - + 'chat group \n' - + 'what you can do is to keep track of new/left' - + 'members with events in some database') + raise ActionNotSupportedError( + "getting users by their usernames is not " + + "possible on telegram. to be more precise, " + + "it is impossible to get all users from " + + "chat group \n" + + "what you can do is to keep track of new/left" + + "members with events in some database" + ) async def delete_messages(self, chat_id: int, *messages_ids: int): try: successful = await self.client.bot.delete_messages(chat_id, messages_ids) if not successful: - raise BotApiError('telegram', 'failed to delete messages, perhaps ' - + 'you dont have the permission to do this') + raise BotApiError( + "telegram", + "failed to delete messages, perhaps " + + "you dont have the permission to do this", + ) except TelegramError as telegram_error: - raise BotApiError('telegram', telegram_error.message) + raise BotApiError("telegram", telegram_error.message) def __convert_stream_to_telegram_media(self, stream: BufferedReader): file = InputFile(stream) mimetype = file.mimetype - if mimetype.startswith('video'): + if mimetype.startswith("video"): return InputMediaVideo(media=file.input_file_content) - elif mimetype.startswith('image'): + elif mimetype.startswith("image"): return InputMediaPhoto(media=file.input_file_content) - elif mimetype.startswith('audio'): + elif mimetype.startswith("audio"): return InputMediaAudio(media=file.input_file_content) else: return InputMediaDocument(media=file.input_file_content) diff --git a/wiring/platforms/telegram/_entities_converter.py b/wiring/platforms/telegram/_entities_converter.py index 9e38101..cdd7a64 100644 --- a/wiring/platforms/telegram/_entities_converter.py +++ b/wiring/platforms/telegram/_entities_converter.py @@ -3,44 +3,49 @@ from telegram import Chat, Message, User, ChatFullInfo from wiring._to_multi_platform_converter import ToMultiPlatformConverter -from wiring.multi_platform_resources import (MultiPlatformChatGroup, MultiPlatformMessage, - MultiPlatformChat, MultiPlatformUser) +from wiring.multi_platform_resources import ( + MultiPlatformChatGroup, + MultiPlatformMessage, + MultiPlatformChat, + MultiPlatformUser, +) class TelegramEntitiesConverter(ToMultiPlatformConverter): def convert_to_multi_platform_chat_group(self, chat_group: Chat): - logger = logging.getLogger('wiring.telegram') - logger.warning('chat group is the same entity as the chat for telegram') + logger = logging.getLogger("wiring.telegram") + logger.warning("chat group is the same entity as the chat for telegram") - return MultiPlatformChatGroup('telegram', - chat_group.id, - chat_group.title or chat_group.full_name) + return MultiPlatformChatGroup( + "telegram", chat_group.id, chat_group.title or chat_group.full_name + ) def convert_to_multi_platform_chat(self, chat: Chat | ChatFullInfo): - return MultiPlatformChat('telegram', - chat.id, - chat.title or chat.full_name) - - def convert_to_multi_platform_user(self, user: User, - from_chat: Optional[MultiPlatformChatGroup] = None): - return MultiPlatformUser('telegram', user.id, user.username or user.full_name, - from_chat) + return MultiPlatformChat("telegram", chat.id, chat.title or chat.full_name) - def convert_to_multi_platform_message( - self, message: Message + def convert_to_multi_platform_user( + self, user: User, from_chat: Optional[MultiPlatformChatGroup] = None ): + return MultiPlatformUser( + "telegram", user.id, user.username or user.full_name, from_chat + ) + + def convert_to_multi_platform_message(self, message: Message): chat = self.convert_to_multi_platform_chat_group(message.chat) - author_user = (self.convert_to_multi_platform_user(message.from_user, chat) - if message.from_user is not None else None) + author_user = ( + self.convert_to_multi_platform_user(message.from_user, chat) + if message.from_user is not None + else None + ) return MultiPlatformMessage( - 'telegram', + "telegram", message.id, chat, self.convert_to_multi_platform_chat(message.chat), message.text, - author_user + author_user, ) diff --git a/wiring/platforms/twitch/__init__.py b/wiring/platforms/twitch/__init__.py index acfb460..827fae2 100644 --- a/wiring/platforms/twitch/__init__.py +++ b/wiring/platforms/twitch/__init__.py @@ -9,40 +9,44 @@ from wiring.bot_base import Bot, Event from wiring.errors.bot_api_error import BotApiError from wiring.errors.not_found_error import NotFoundError -from wiring.platforms.twitch._entities_converter import (twitch_entities_converter, - TwitchMessageWithUser) +from wiring.platforms.twitch._entities_converter import ( + twitch_entities_converter, + TwitchMessageWithUser, +) class CustomTwitchClient(twitchio.ext.commands.Bot): - def __init__(self, - access_token: str, - streamer_usernames_to_connect: list[str], - get_commands_prefix: Callable[[], str], - do_on_event: Callable): + def __init__( + self, + access_token: str, + streamer_usernames_to_connect: list[str], + get_commands_prefix: Callable[[], str], + do_on_event: Callable, + ): self._do_on_event = do_on_event self.access_token = access_token - super().__init__(access_token, - prefix=get_commands_prefix, - initial_channels=streamer_usernames_to_connect) + super().__init__( + access_token, + prefix=get_commands_prefix, + initial_channels=streamer_usernames_to_connect, + ) async def event_message(self, message: twitchio.Message): if message.echo: return self._do_on_event( - 'message', + "message", twitch_entities_converter.convert_to_multi_platform_message( TwitchMessageWithUser(message, await message.author.user()) - ) + ), ) class TwitchBot(Bot): - def __init__(self, - access_token: str, - streamer_usernames_to_connect: list[str]): + def __init__(self, access_token: str, streamer_usernames_to_connect: list[str]): """initializes a twitch bot for usage with `MultiPlatformBot` class Args: @@ -51,19 +55,21 @@ def __init__(self, the specific stream without explicitly connecting to it by the streamer username """ - super().__init__('twitch') + super().__init__("twitch") - self.client = CustomTwitchClient(access_token, - streamer_usernames_to_connect, - lambda: self.commands_prefix, - self._run_event_from_twitch_client) + self.client = CustomTwitchClient( + access_token, + streamer_usernames_to_connect, + lambda: self.commands_prefix, + self._run_event_from_twitch_client, + ) - self.logger = logging.getLogger('wiring.twitch') + self.logger = logging.getLogger("wiring.twitch") def _run_event_from_twitch_client(self, event: Event, event_data=None): self._run_event_handlers(event, event_data) - if event == 'message' and event_data is not None: + if event == "message" and event_data is not None: self._check_message_for_command(event_data) async def start(self): @@ -72,29 +78,29 @@ async def start(self): async def stop(self): await self.client.close() - async def send_message(self, - chat_id: str, - text, - reply_message_id: Optional[int] = None, - files=None): + async def send_message( + self, chat_id: str, text, reply_message_id: Optional[int] = None, files=None + ): target_channel = self._get_channel_or_raise(chat_id) if reply_message_id is not None or files is not None: - self.logger.warning('replying to messages and attaching files are not ' - + 'supported for twitch') + self.logger.warning( + "replying to messages and attaching files are not " + + "supported for twitch" + ) try: await target_channel.send(text) except twitchio.InvalidContent: - raise BotApiError('twitch', 'message contains inappropriate/invalid content') + raise BotApiError( + "twitch", "message contains inappropriate/invalid content" + ) except twitchio.TwitchIOException as twitch_error: - raise BotApiError('twitch', str(twitch_error)) + raise BotApiError("twitch", str(twitch_error)) async def get_chat_groups(self, on_platform=None): return [ - twitch_entities_converter.convert_to_multi_platform_chat_group( - channel - ) + twitch_entities_converter.convert_to_multi_platform_chat_group(channel) for channel in self.client.connected_channels ] @@ -102,11 +108,9 @@ async def get_chats_from_group(self, chat_group_id: str): channel = self._get_channel_or_raise(chat_group_id) return [twitch_entities_converter.convert_to_multi_platform_chat(channel)] - async def ban(self, - chat_group_id: str, - user_id: int, - reason=None, - seconds_duration=None): + async def ban( + self, chat_group_id: str, user_id: int, reason=None, seconds_duration=None + ): streamer = await self._get_channel_or_raise(chat_group_id).user() if self.client.user_id is None: @@ -114,18 +118,22 @@ async def ban(self, try: if seconds_duration is None: - await streamer.ban_user(self.client.access_token, - self.client.user_id, - user_id, - reason or '-') + await streamer.ban_user( + self.client.access_token, + self.client.user_id, + user_id, + reason or "-", + ) else: - await streamer.timeout_user(self.client.access_token, - self.client.user_id, - user_id, - seconds_duration, - reason or '-') + await streamer.timeout_user( + self.client.access_token, + self.client.user_id, + user_id, + seconds_duration, + reason or "-", + ) except twitchio.TwitchIOException as twitch_error: - raise BotApiError('twitch', str(twitch_error)) + raise BotApiError("twitch", str(twitch_error)) async def get_user_by_name(self, username: str, chat_group_id: int): users = await self.client.fetch_users(names=[username]) @@ -140,15 +148,17 @@ async def delete_messages(self, chat_id: str, *messages_ids: str): try: for message_id in messages_ids: - await channel.send(f'/delete {message_id}') + await channel.send(f"/delete {message_id}") except twitchio.TwitchIOException as twitch_error: - raise BotApiError('twitch', str(twitch_error)) + raise BotApiError("twitch", str(twitch_error)) def _get_channel_or_raise(self, username: str): - channel = find_item(self.client.connected_channels, - lambda channel: channel.name == username) + channel = find_item( + self.client.connected_channels, lambda channel: channel.name == username + ) if channel is None: - raise NotFoundError('twitch', 'twitch channel with username ' - + f'\'{username}\' not found') + raise NotFoundError( + "twitch", "twitch channel with username " + f"'{username}' not found" + ) return channel diff --git a/wiring/platforms/twitch/_entities_converter.py b/wiring/platforms/twitch/_entities_converter.py index e9e2c76..81d4e1d 100644 --- a/wiring/platforms/twitch/_entities_converter.py +++ b/wiring/platforms/twitch/_entities_converter.py @@ -4,10 +4,12 @@ from wiring._to_multi_platform_converter import ToMultiPlatformConverter -from wiring.multi_platform_resources import (MultiPlatformChat, - MultiPlatformChatGroup, - MultiPlatformMessage, - MultiPlatformUser) +from wiring.multi_platform_resources import ( + MultiPlatformChat, + MultiPlatformChatGroup, + MultiPlatformMessage, + MultiPlatformUser, +) class TwitchMessageWithUser: @@ -20,32 +22,31 @@ def __init__(self, message: twitchio.Message, user: twitchio.User): class TwitchEntitiesConverter(ToMultiPlatformConverter): def convert_to_multi_platform_chat_group(self, chat_group: twitchio.Channel): - logger = logging.getLogger('wiring.twitch') - logger.warning('chat group is the same entity as the chat for twitch') + logger = logging.getLogger("wiring.twitch") + logger.warning("chat group is the same entity as the chat for twitch") - return MultiPlatformChatGroup('twitch', chat_group.name, chat_group.name) + return MultiPlatformChatGroup("twitch", chat_group.name, chat_group.name) def convert_to_multi_platform_chat(self, chat: twitchio.Channel): - return MultiPlatformChat('twitch', chat.name, chat.name) + return MultiPlatformChat("twitch", chat.name, chat.name) def convert_to_multi_platform_message(self, message: TwitchMessageWithUser): return MultiPlatformMessage( - 'twitch', + "twitch", message.id, self.convert_to_multi_platform_chat_group(message.channel), self.convert_to_multi_platform_chat(message.channel), message.text, - self.convert_to_multi_platform_user(message.user) + self.convert_to_multi_platform_user(message.user), ) - def convert_to_multi_platform_user(self, - user: twitchio.User): + def convert_to_multi_platform_user(self, user: twitchio.User): from_chat_group = None if user.channel is not None: from_chat_group = self.convert_to_multi_platform_chat_group(user.channel) - return MultiPlatformUser('twitch', user.id, user.name, from_chat_group) + return MultiPlatformUser("twitch", user.id, user.name, from_chat_group) twitch_entities_converter = TwitchEntitiesConverter()