' + bot_info.config.config['Command user'][x] + '
' - if is_admin: - for x in bot_info.config.config['Command admin']: - message = message + '' + bot_info.config.config['Command admin'][x] + '
' - bot_info.get_user_by_id(user_id).send_text_message(message) - case '!users': - if is_admin: - message = 'Список пользователей:
' - users = bot_info.get_userlist() - for user in users: - message = message + '' + user['name'] + '
' - bot_info.get_user_by_id(user_id).send_text_message(message) - case '!initchannels': - MumbleBotService.init_channels(bot_info) - message = 'Каналы обновлены' - bot_info.get_user_by_id(user_id).send_text_message(message) - case 'test': - print('тестовая команда') - MumbleBotService.test(bot_info.bot, message_info) - case _: - bot_info.get_user_by_id(user_id).send_text_message(message) - - @staticmethod - def user_change_channel(bot_info: MumbleBot, user, action): - if 'channel_id' in action: - MumbleBotService.user_change_channel(bot_info, user, action) diff --git a/src/model/MumbleBot.py b/src/model/MumbleBot.py index d669174..f90da99 100644 --- a/src/model/MumbleBot.py +++ b/src/model/MumbleBot.py @@ -1,17 +1,14 @@ import logging import time -from typing import Optional import pymumble_py3.constants from pymumble_py3 import Mumble -from pymumble_py3.users import Users, User from src.config.BotConfig import BotConfig from pymumble_py3.callbacks import PYMUMBLE_CLBK_TEXTMESSAGERECEIVED as PCTMR from pymumble_py3.callbacks import PYMUMBLE_CLBK_USERUPDATED as PCUU - -from src.controller.MumbleController import MumbleController +from src.service.ChannelService import ChannelService from src.service.MumbleBotService import MumbleBotService _logger = logging.getLogger(__name__) @@ -20,6 +17,8 @@ _logger = logging.getLogger(__name__) class MumbleBot: config: BotConfig bot: Mumble + mumbleBotService: MumbleBotService + channelService: ChannelService def __init__(self): self.thread = None @@ -37,74 +36,23 @@ class MumbleBot: self.bot.callbacks.set_callback(PCUU, self.user_change_channel) # self.bot.callbacks.set_callback(PCUC, self.user_connect_server) + self.mumbleBotService = MumbleBotService(self) + self.channelService = ChannelService(self) + self.bot.start() # start the mumble thread self.bot.is_ready() # wait for the connection + self.channelService.init_channels() if self.bot.connected >= pymumble_py3.constants.PYMUMBLE_CONN_STATE_FAILED: exit() def message_received(self, message_info): - MumbleController.message_received(self, message_info) + self.mumbleBotService.message_received(message_info) def user_change_channel(self, user, action): - MumbleController.user_change_channel(self, user, action) - - def get_username_by_id(self, user_id: int) -> Optional[str]: - try: - return self.get_user_by_id(user_id)['name'] - except KeyError: - _logger.warning(f"Не удалось найти пользователя с id: {user_id}") - return None - - def get_user_by_id(self, user_id: int) -> Optional[User]: - try: - users = self.bot.users - for user in users.values(): - if user['user_id'] == user_id: - return user - _logger.warning(f"Не удалось найти пользователя с id: {user_id}") - return None - except KeyError: - _logger.warning(f"Не удалось найти пользователя с id: {user_id}") - return None - - def get_user_by_name(self, user_name: str) -> Optional[User]: - users = self.bot.users - for user in users: - if user['name'] == user_name: - return user - _logger.warning(f"Не удалось найти пользователя с username: {user_name}") - return None - - def get_userlist(self) -> Users: - return self.bot.users.values() - - # # TODO - # # обработчик колбэка подключения пользователя к серверу - # @staticmethod - # def user_connect_server(user): - # if kostyl1 == 1: - # # дополняем список всех пользователей сервера - # if mumble.users[user['session']]['name'] in userlist: - # welcome_message(user) - # check_offline_messages(user) - # else: - # first_welcome_message(user) - # userlist.add_section(mumble.users[user['session']]['name']) - # userlist.set(mumble.users[user['session']]['name'], mumble.users[user['session']]['name'], '') - # with open(config['Bot']['userlist'], "w") as config_file: - # userlist.write(config_file) - - def is_admin(self, user) -> bool: - list_admin = self.config.adminList - if user in list_admin: - return True - else: - return False + self.mumbleBotService.user_change_channel(user, action) def loop(self): - MumbleBotService.init_channels(self) while not self.exit and self.bot.is_alive(): - while self.thread and self.bot.sound_output.get_buffer_size() > 0.5 and not self.exit: # If the buffer isn't empty, I cannot send new music part, so I wait self._loop_status = f'Wait for buffer {self.bot.sound_output.get_buffer_size():.3f}' diff --git a/src/service/ChannelService.py b/src/service/ChannelService.py index a97cca6..48bec61 100644 --- a/src/service/ChannelService.py +++ b/src/service/ChannelService.py @@ -1,26 +1,29 @@ import logging from time import sleep -from pymumble_py3 import Mumble from src.model import MumbleBot _logger = logging.getLogger(__name__) class ChannelService: - @classmethod - def init_channels(cls, mumbleBot: MumbleBot): - channels = mumbleBot.config.channelList + bot_data: MumbleBot + + def __init__(self, bot_data: MumbleBot): + self.bot_data = bot_data + + def init_channels(self): + self.bot_data.config.load_channel_list() + channels = self.bot_data.config.channelList for channel in channels: parent_channel_id = 0 path_split = channel.path.split('/') for path in path_split: - parent_channel_id = cls.create_channel(mumbleBot.bot, path, parent_channel_id) + parent_channel_id = self.create_channel(path, parent_channel_id) - @classmethod - def create_channel(cls, mumbleBot: Mumble, name: str, parent_channel_id: int) -> int: + def create_channel(self, name: str, parent_channel_id: int) -> int: parent_channel_obj = [] - channels = mumbleBot.channels.values() + channels = self.bot_data.bot.channels.values() for item in channels: if item.get('channel_id') is not None and item["channel_id"] == parent_channel_id: parent_channel_obj = item @@ -28,20 +31,19 @@ class ChannelService: return item["channel_id"] _logger.info(f"Создаю подканал {name} в {parent_channel_obj['name']}") - mumbleBot.channels.new_channel(parent_channel_id, name, temporary=False) + self.bot_data.bot.channels.new_channel(parent_channel_id, name, temporary=False) sleep(1) - channels = mumbleBot.channels.values() + channels = self.bot_data.bot.channels.values() for item in channels: if item.get('parent') is not None and item["parent"] == parent_channel_id and item["name"] == name: return item["channel_id"] return 0 - @staticmethod - def get_message_by_channel_id(mumbleBot: MumbleBot, channel_id: int) -> str: - name = mumbleBot.bot.channels[channel_id]['name'] - channel_list = mumbleBot.config.channelList + def get_message_by_channel_id(self, channel_id: int) -> str: + name = self.bot_data.bot.channels[channel_id]['name'] + channel_list = self.bot_data.config.channelList for channel in channel_list: if channel.name == name: return channel.message diff --git a/src/service/MumbleBotService.py b/src/service/MumbleBotService.py index 4b96f93..c3bc31f 100644 --- a/src/service/MumbleBotService.py +++ b/src/service/MumbleBotService.py @@ -1,6 +1,7 @@ import logging +from typing import Optional -from pymumble_py3 import Mumble +from pymumble_py3.users import User, Users from src.model import MumbleBot from src.service.ChannelService import ChannelService @@ -9,21 +10,126 @@ _logger = logging.getLogger(__name__) class MumbleBotService: + bot_data: MumbleBot + channel_service: ChannelService - @staticmethod - def init_channels(bot_info): - ChannelService.init_channels(bot_info) + def __init__(self, bot_data: MumbleBot): + self.bot_data = bot_data + self.channel_service = ChannelService(bot_data) + + def message_received(self, message_info): + actor_id = message_info.actor + if actor_id == 0: + return + user_id = self.get_user_id_by_actor_id(actor_id) + username = self.get_username_by_id(user_id) + is_admin = self.is_admin(username) + message = message_info.message.strip() + + _logger.info(f"{username} {('', '(админ бота)')[is_admin]} прислал сообщение: {message}") + + match message: + case '!help': + message = ' ' + for x in self.bot_data.config.config['Command user']: + message = message + '' + self.bot_data.config.config['Command user'][x] + '
' + if is_admin: + for x in self.bot_data.config.config['Command admin']: + message = message + '' + self.bot_data.config.config['Command admin'][x] + '
' + self.get_user_by_id(user_id).send_text_message(message) + case '!users': + if is_admin: + message = 'Список пользователей:
' + users = self.bot_data.get_userlist() + for user in users: + message = message + '' + user['name'] + '
' + self.get_user_by_id(user_id).send_text_message(message) + case '!initchannels': + self.channel_service.init_channels() + message = 'Каналы обновлены' + self.get_user_by_id(user_id).send_text_message(message) + case 'test': + self.test(self.bot_data.bot, message_info) + case _: + self.get_user_by_id(user_id).send_text_message(message) # TODO: одинаковые каналы (коллизия сообщений) - @staticmethod - def user_change_channel(bot_info: MumbleBot, user, action): - _logger.info(f"{user['name']} перешёл в канал: { bot_info.bot.channels[action['channel_id']]['name']}") - print(f"{user['name']} перешёл в канал: { bot_info.bot.channels[action['channel_id']]['name']}") + def user_change_channel(self, user: User, action): + _logger.info(f"{user['name']} перешёл в канал: {self.bot_data.bot.channels[action['channel_id']]['name']}") - message = ChannelService.get_message_by_channel_id(bot_info, action['channel_id']) + message = self.channel_service.get_message_by_channel_id(action['channel_id']) if len(message) > 0: - bot_info.get_user_by_id(user['user_id']).send_text_message(message) + self.get_user_by_id(user['user_id']).send_text_message(message) - @staticmethod - def test(bot: Mumble, message_info): - ChannelService.get_tree(bot) + def get_user_id_by_actor_id(self, actor_id: int) -> Optional[int]: + user = self.get_user_by_actor_id(actor_id) + if user is not None: + return user['user_id'] + else: + return None + + def get_user_by_actor_id(self, actor_id: int) -> Optional[User]: + try: + users = self.bot_data.bot.users + for user in users.values(): + if user['session'] == actor_id: + return user + except KeyError: + _logger.warning(f"Не удалось найти пользователя с actorId: {actor_id}") + return None + + def get_username_by_id(self, user_id: int) -> Optional[str]: + try: + return self.get_user_by_id(user_id)['name'] + except KeyError: + _logger.warning(f"Не удалось найти пользователя с id: {user_id}") + return None + + def get_user_by_id(self, user_id: int) -> Optional[User]: + try: + users = self.bot_data.bot.users + for user in users.values(): + if user['user_id'] == user_id: + return user + _logger.warning(f"Не удалось найти пользователя с id: {user_id}") + return None + except KeyError: + _logger.warning(f"Не удалось найти пользователя с id: {user_id}") + return None + + def get_user_by_name(self, user_name: str) -> Optional[User]: + users = self.bot_data.bot.users + for user in users: + if user['name'] == user_name: + return user + _logger.warning(f"Не удалось найти пользователя с username: {user_name}") + return None + + def get_userlist(self) -> Users: + return self.bot_data.bot.users.values() + + # # TODO + # # обработчик колбэка подключения пользователя к серверу + # @staticmethod + # def user_connect_server(user): + # if kostyl1 == 1: + # # дополняем список всех пользователей сервера + # if mumble.users[user['session']]['name'] in userlist: + # welcome_message(user) + # check_offline_messages(user) + # else: + # first_welcome_message(user) + # userlist.add_section(mumble.users[user['session']]['name']) + # userlist.set(mumble.users[user['session']]['name'], mumble.users[user['session']]['name'], '') + # with open(config['Bot']['userlist'], "w") as config_file: + # userlist.write(config_file) + + def is_admin(self, user) -> bool: + list_admin = self.bot_data.config.adminList + if user in list_admin: + return True + else: + return False + + def test(self): + self.channel_service.get_tree() diff --git a/src/start.py b/src/start.py index 4c6f93b..0216d3a 100644 --- a/src/start.py +++ b/src/start.py @@ -1,8 +1,33 @@ import logging +import coloredlogs from src.model.MumbleBot import MumbleBot -logging.getLogger().setLevel(logging.INFO) +# Настройка цветов для уровней логирования +level_styles = { + 'debug': {'color': 'white'}, + 'info': {'color': 'white'}, + 'warning': {'color': 'yellow'}, + 'error': {'color': 'red'}, # Жирный красный для ошибок + 'critical': {'color': 'red', 'bold': True}, # Жирный магента с белым фоном +} + +field_styles = { + 'asctime': {'color': 'cyan'}, # Время логирования в цвете циан + 'name': {'color': 'yellow'}, # Имя логгера в желтом + 'levelname': {'color': 'blue'}, # Уровень логирования в жирном красном + 'message': {'color': 'white'}, # Сообщение в белом +} + +coloredlogs.install(level='DEBUG', level_styles=level_styles, field_styles=field_styles) + +_logger = logging.getLogger(__name__) + +_logger.debug("Это сообщение DEBUG уровня") +_logger.info("Это сообщение INFO уровня") +_logger.warning("Это сообщение WARNING уровня") +_logger.error("Это сообщение ERROR уровня") +_logger.critical("Это сообщение CRITICAL уровня") + bot = MumbleBot() bot.loop() -