рефактор

src/Utils.py

@@ -0,0 +1,6 @@
+import configparser
+
+
+class Utils:
+    def __init__(self):
+        pass

src/config/BotConfig.py

@@ -0,0 +1,46 @@
+import configparser
+import os
+from typing import List
+import pandas
+
+from src.model.MumbleChannel import MumbleChannel
+
+
+class BotConfig:
+    config: configparser.ConfigParser
+    adminList: List
+    userList: configparser.ConfigParser
+    channelList: List[MumbleChannel]
+
+    def __init__(self):
+        self.load_settings()
+        self.channelList = []
+        self.load_channel_list()
+
+    def load_settings(self):
+        self.config = configparser.ConfigParser()
+        self.config.read('settings.ini', encoding="utf-8")
+        self.adminList = self.config['Bot']['admins'].rsplit(sep=';')
+
+        if not os.path.exists(self.config['Bot']['userlist']):
+            f = open(self.config['Bot']['userlist'], 'tw', encoding='utf-8')
+            f.close()
+
+        self.userList = configparser.ConfigParser()
+        self.userList.read(self.config['Bot']['userlist'])
+
+    def load_channel_list(self):
+        data = pandas.read_csv('channels.csv', delimiter=";")
+        data = data.reset_index()
+        for index, row in data.iterrows():
+            self.channelList.append(MumbleChannel(row['path'], row['message']))
+
+
+
+
+# bot = BotConfig()
+# config = bot.config
+# adminList = bot.adminList
+# userList = bot.userList
+#for ch in bot.channelList:
+#    print(ch.__str__())

src/controller/MumbleController.py

@@ -0,0 +1,52 @@
+import logging
+
+from src.model import MumbleBot
+from src.service.MumbleBotService import MumbleBotService
+
+_logger = logging.getLogger(__name__)
+
+
+class MumbleController:
+    @staticmethod
+    def message_received(bot_info: MumbleBot, message_info):
+        user_id = message_info.actor
+        if user_id == 0:
+            return
+
+        username = bot_info.get_username_by_id(user_id)
+        is_admin = bot_info.is_admin(username)
+        message = message_info.message.strip()
+
+        print(f"{username} {('', '(админ бота)')[is_admin]} прислал сообщение: {message}")
+        _logger.info(f"{username} {('', '(админ бота)')[is_admin]} прислал сообщение: {message}")
+
+        match message:
+            case '!help':
+                message = ' '
+                for x in bot_info.config.config['Command user']:
+                    message = message + '<p>' + bot_info.config.config['Command user'][x] + '</p>'
+                if is_admin:
+                    for x in bot_info.config.config['Command admin']:
+                        message = message + '<p>' + bot_info.config.config['Command admin'][x] + '</p>'
+                bot_info.get_user_by_id(user_id).send_text_message(message)
+            case '!users':
+                if is_admin:
+                    message = '<p>Список пользователей:</p>'
+                    users = bot_info.get_userlist()
+                    for user in users:
+                        message = message + '<p>' + user['name'] + '</p>'
+                    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 user:  # колбэк вызывается также включением/выключением микрофона у пользователя, тут стоит фильтр "Только переходы по каналам"
+            MumbleBotService.user_change_channel(bot_info, user, action)

src/model/MumbleBot.py

@@ -0,0 +1,111 @@
+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.MumbleBotService import MumbleBotService
+
+_logger = logging.getLogger(__name__)
+
+
+class MumbleBot:
+    config: BotConfig
+    bot: Mumble
+
+    def __init__(self):
+        self.thread = None
+        self._loop_status = None
+
+        self.exit = None
+        _logger.info('!!!Starting bot!!!')
+        self.config = BotConfig()
+        config = self.config.config
+        self.bot = Mumble(host=config['Server']['address'], user=config['Bot']['Bot_name'],
+                          port=int(config['Server']['port']), password=config['Server']['password'],
+                          certfile=config['Bot']['certfile'], reconnect=True, tokens=config['Server']['tokens'])
+
+        self.bot.callbacks.set_callback(PCTMR, self.message_received)
+        self.bot.callbacks.set_callback(PCUU, self.user_change_channel)
+        # self.bot.callbacks.set_callback(PCUC, self.user_connect_server)
+
+        self.bot.start()  # start the mumble thread
+        self.bot.is_ready()  # wait for the connection
+        if self.bot.connected >= pymumble_py3.constants.PYMUMBLE_CONN_STATE_FAILED:
+            exit()
+
+    def message_received(self, message_info):
+        MumbleController.message_received(self, 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
+
+    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}'
+                time.sleep(0.01)

src/model/MumbleChannel.py

@@ -0,0 +1,13 @@
+class MumbleChannel:
+    name: str
+    path: str
+    message: str
+
+    def __init__(self, path: str, message: str):
+        self.path = path
+        self.message = message
+        self.name = path.split('/')[-1]
+
+    def __str__(self) -> str:
+        string = f"MumbleChannel:\n name: {self.name} \n path: {self.path} \n message: {self.message}"
+        return string

src/service/ChannelService.py

@@ -0,0 +1,48 @@
+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
+        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)
+
+    @classmethod
+    def create_channel(cls, mumbleBot: Mumble, name: str, parent_channel_id: int) -> int:
+        parent_channel_obj = []
+        channels = mumbleBot.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
+            if item.get('parent') is not None and item["parent"] == parent_channel_id and item["name"] == name:
+                return item["channel_id"]
+
+        _logger.info(f"Создаю подканал {name} в {parent_channel_obj['name']}")
+        mumbleBot.channels.new_channel(parent_channel_id, name, temporary=False)
+        sleep(1)
+
+        channels = mumbleBot.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
+        for channel in channel_list:
+            if channel.name == name:
+                return channel.message
+        return ''

src/service/MumbleBotService.py

@@ -0,0 +1,29 @@
+import logging
+
+from pymumble_py3 import Mumble
+
+from src.model import MumbleBot
+from src.service.ChannelService import ChannelService
+
+_logger = logging.getLogger(__name__)
+
+
+class MumbleBotService:
+
+    @staticmethod
+    def init_channels(bot_info):
+        ChannelService.init_channels(bot_info)
+
+    # 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']}")
+
+        message = ChannelService.get_message_by_channel_id(bot_info, action['channel_id'])
+        if len(message) > 0:
+            bot_info.get_user_by_id(user['user_id']).send_text_message(message)
+
+    @staticmethod
+    def test(bot: Mumble, message_info):
+        ChannelService.get_tree(bot)

src/start.py

@@ -0,0 +1,8 @@
+import logging
+
+from src.model.MumbleBot import MumbleBot
+
+logging.getLogger().setLevel(logging.INFO)
+bot = MumbleBot()
+bot.loop()
+