L_DelOff
/
MumbleConciergeBot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Немного архитектуры

main
L_DelOff 2024-10-20 17:35:20 +03:00
parent c99d491293
commit c9ad3934e5
8 changed files with 181 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
requirements.txt
View File

@ -1,2 +1,3 @@
pymumble
pandas~=2.2.1
pandas~=2.2.1
coloredlogs

6
src/Utils.py
View File

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

9
src/config/BotConfig.py
View File

@ -1,10 +1,13 @@
import configparser
import logging
import os
from typing import List
import pandas
from src.model.MumbleChannel import MumbleChannel
_logger = logging.getLogger(__name__)
class BotConfig:
config: configparser.ConfigParser
@ -15,9 +18,9 @@ class BotConfig:
def __init__(self):
self.load_settings()
self.channelList = []
self.load_channel_list()
def load_settings(self):
_logger.info("Загрузка файла с настройками бота...")
self.config = configparser.ConfigParser()
self.config.read('settings.ini', encoding="utf-8")
self.adminList = self.config['Bot']['admins'].rsplit(sep=';')
@ -28,12 +31,16 @@ class BotConfig:
self.userList = configparser.ConfigParser()
self.userList.read(self.config['Bot']['userlist'])
_logger.info("Файл с настройками бота загружен")
def load_channel_list(self):
_logger.info("Загрузка файла с настройками каналов ...")
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']))
_logger.info("Файл с настройками каналов загружен")

52
src/controller/MumbleController.py
View File

@ -1,52 +0,0 @@
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 action:
MumbleBotService.user_change_channel(bot_info, user, action)

70
src/model/MumbleBot.py
View File

@ -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}'

30
src/service/ChannelService.py
View File

@ -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

132
src/service/MumbleBotService.py
View File

@ -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 + '<p>' + self.bot_data.config.config['Command user'][x] + '</p>'
if is_admin:
for x in self.bot_data.config.config['Command admin']:
message = message + '<p>' + self.bot_data.config.config['Command admin'][x] + '</p>'
self.get_user_by_id(user_id).send_text_message(message)
case '!users':
if is_admin:
message = '<p>Список пользователей:</p>'
users = self.bot_data.get_userlist()
for user in users:
message = message + '<p>' + user['name'] + '</p>'
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()

29
src/start.py
View File

@ -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()