From fe2555ace10d9a9983f6b213e83f9787a872429d Mon Sep 17 00:00:00 2001 From: teanus Date: Fri, 2 Aug 2024 05:30:29 +0500 Subject: [PATCH] Docstring, typehints --- add_super_admin.py | 10 + bot.py | 12 +- custom_filters/text_list.py | 18 ++ keyboards/main_menu.py | 7 + logger/group_logger.py | 8 + minecraft/rcon.py | 7 + provider/databases.py | 357 ++++++++++++++++++++++++++++++++++-- resources/config.py | 47 +++++ tools/table_formatted.py | 8 + 9 files changed, 462 insertions(+), 12 deletions(-) diff --git a/add_super_admin.py b/add_super_admin.py index 686e476..67611bf 100644 --- a/add_super_admin.py +++ b/add_super_admin.py @@ -20,7 +20,14 @@ async def console_add_super_admin() -> str: + """ + Добавляет супер-администратора через консоль. + + :return: Возвращает строку с результатом выполнения. + :rtype: str + """ root = "template" + if not config.console()["give_role"]: return render_template_jinja( "add_super_admin/false_give_role.jinja2", root_directory_name=root @@ -31,18 +38,21 @@ async def console_add_super_admin() -> str: "add_super_admin/messages.jinja2", root_directory_name=root ) ) + if admin_id == "": return render_template_jinja( "add_super_admin/exit.jinja2", root_directory_name=root ) context = {"admin_id": admin_id} + if await db.check_admin(admin_id): return render_template_jinja( "add_super_admin/admin_exists.jinja2", root_directory_name=root, **context ) await db.add_admin(admin_id) + return render_template_jinja( "add_super_admin/add_admin.jinja2", root_directory_name=root, **context ) diff --git a/bot.py b/bot.py index ba99e30..080f58b 100644 --- a/bot.py +++ b/bot.py @@ -31,6 +31,12 @@ async def on_startup() -> None: + """ + Обработчик событий при запуске бота. Выполняет добавление супер-администратора + и логирует сообщение о запуске. + + :return: None + """ await console_add_super_admin() text = render_template_jinja("on_startup.jinja2", "template/bot") print(text) @@ -38,6 +44,11 @@ async def on_startup() -> None: async def on_shutdown() -> None: + """ + Обработчик событий при остановке бота. Логирует сообщение о завершении работы. + + :return: None + """ text = render_template_jinja("on_shutdown.jinja2", "template/bot") print(text) logger.info(text) @@ -50,7 +61,6 @@ async def main(): dp.include_router(client_router) dp.include_router(admin_router) dp.include_router(common_router) - # Регистрация обработчиков для каждого роутера await register_other_handlers() await register_client_handlers() await register_admin_handlers() diff --git a/custom_filters/text_list.py b/custom_filters/text_list.py index 1ad1e30..a989344 100644 --- a/custom_filters/text_list.py +++ b/custom_filters/text_list.py @@ -18,8 +18,26 @@ class TextInFilter(BaseFilter): + """ + Фильтр для проверки наличия текста сообщения в заданном списке текстов. + + :param texts: Список строк для проверки. + """ + def __init__(self, texts: List[str]): + """ + Инициализация фильтра. + + :param texts: Список строк для проверки. + """ self.texts = [text.lower() for text in texts] async def __call__(self, message: Message) -> bool: + """ + Проверка, содержится ли текст сообщения в списке текстов. + + :param message: Сообщение Telegram. + :return: True, если текст сообщения содержится в списке, иначе False. + :rtype: bool + """ return message.text.lower() in self.texts diff --git a/keyboards/main_menu.py b/keyboards/main_menu.py index d64649b..c4ad1bb 100644 --- a/keyboards/main_menu.py +++ b/keyboards/main_menu.py @@ -5,6 +5,13 @@ async def get_main_menu(user_id: int) -> types.ReplyKeyboardMarkup: + """ + Получает главное меню в зависимости от роли пользователя. + + :param user_id: ID пользователя. + :return: Главное меню в виде клавиатуры. + :rtype: types.ReplyKeyboardMarkup + """ if await db.check_admin(user_id): return kb_admin.main_menu elif await db.user_exists(user_id): diff --git a/logger/group_logger.py b/logger/group_logger.py index a082a67..f4c311f 100644 --- a/logger/group_logger.py +++ b/logger/group_logger.py @@ -23,6 +23,14 @@ async def groups_logger(prefix: str, user_id: int, message: str) -> None: + """ + Логирует сообщения в группы, если включен лагер групп. + + :param prefix: Префикс сообщения. + :param user_id: ID пользователя. + :param message: Сообщение для логирования. + :return: None + """ if config.telegram().get("on_logger_group"): is_admin = await db.check_admin(user_id) context = { diff --git a/minecraft/rcon.py b/minecraft/rcon.py index 34eaaf8..3ce2ee0 100644 --- a/minecraft/rcon.py +++ b/minecraft/rcon.py @@ -53,6 +53,13 @@ def replace_color_tag(text: str) -> str: def command_execute(command: str) -> Union[str, List[str]]: + """ + Выполняет команду через RCON и возвращает результат. + + :param command: Команда для выполнения. + :return: Результат выполнения команды в виде строки или списка строк. + :rtype: Union[str, List[str]] + """ try: with MCRcon( getenv("rcon_host"), getenv("rcon_password"), int(getenv("rcon_port")) diff --git a/provider/databases.py b/provider/databases.py index 8da94b3..342ff81 100644 --- a/provider/databases.py +++ b/provider/databases.py @@ -14,6 +14,7 @@ # ╚═╝ ╚══════╝╚═╝ ╚═╝╚═╝ ╚═══╝ ╚═════╝ ╚══════╝ from os import getenv +from typing import List, Optional import aiosqlite import asyncpg @@ -29,6 +30,12 @@ def __init__(self): self.con = None async def connect(self) -> None: + """ + Устанавливает соединение с базой данных SQLite и инициализирует таблицы. + + :return: None + :rtype: None + """ try: self.con = await aiosqlite.connect(config.sqlite()["name"]) print("SQLite: connected") @@ -37,10 +44,22 @@ async def connect(self) -> None: print(f"Error connecting to the SQLite database: {error}") async def disconnect(self) -> None: + """ + Закрывает соединение с базой данных SQLite. + + :return: None + :rtype: None + """ if self.con: await self.con.close() async def initialize_tables(self) -> None: + """ + Инициализирует таблицы в базе данных, если они еще не существуют. + + :return: None + :rtype: None + """ table_users = """ CREATE TABLE IF NOT EXISTS users( id INTEGER PRIMARY KEY, @@ -59,7 +78,17 @@ async def initialize_tables(self) -> None: await self.execute_query(table_admins) await self.con.commit() - async def execute_query(self, query: str, params=None) -> bool: + async def execute_query(self, query: str, params: Optional[tuple] = None) -> bool: + """ + Выполняет запрос к базе данных. + + :param query: SQL запрос для выполнения. + :type query: str + :param params: Параметры для запроса. + :type params: Optional[tuple] + :return: Успешность выполнения запроса. + :rtype: bool + """ try: if not self.con: await self.connect() @@ -71,7 +100,19 @@ async def execute_query(self, query: str, params=None) -> bool: print(f"Error executing SQLite query: {error}") return False - async def fetch_all(self, query: str, params=None) -> list: + async def fetch_all( + self, query: str, params: Optional[tuple] = None + ) -> List[tuple]: + """ + Выполняет запрос к базе данных и возвращает все результаты. + + :param query: SQL запрос для выполнения. + :type query: str + :param params: Параметры для запроса. + :type params: Optional[tuple] + :return: Список строк результата запроса. + :rtype: List[tuple] + """ try: if not self.con: await self.connect() @@ -83,45 +124,123 @@ async def fetch_all(self, query: str, params=None) -> list: return [] async def add_user(self, user_id: str) -> bool: + """ + Добавляет пользователя в таблицу пользователей. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Успешность добавления пользователя. + :rtype: bool + """ query = "INSERT INTO users(telegram_id) VALUES(?)" return await self.execute_query(query, (user_id,)) async def user_exists(self, user_id: str) -> bool: + """ + Проверяет, существует ли пользователь в базе данных. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Существует ли пользователь. + :rtype: bool + """ query = "SELECT 1 FROM users WHERE telegram_id = ?" result = await self.fetch_all(query, (user_id,)) return bool(result) async def user_remove(self, user_id: str) -> bool: + """ + Удаляет пользователя из таблицы пользователей. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Успешность удаления пользователя. + :rtype: bool + """ query = "DELETE FROM users WHERE telegram_id = ?" return await self.execute_query(query, (user_id,)) async def add_admin(self, user_id: str) -> bool: + """ + Добавляет администратора в таблицу администраторов. + + :param user_id: Идентификатор администратора. + :type user_id: str + :return: Успешность добавления администратора. + :rtype: bool + """ query = "INSERT INTO admins(telegram_id) VALUES(?)" return await self.execute_query(query, (user_id,)) async def check_admin(self, user_id: str) -> bool: + """ + Проверяет, является ли пользователь администратором. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Является ли пользователь администратором. + :rtype: bool + """ query = "SELECT 1 FROM admins WHERE telegram_id = ?" result = await self.fetch_all(query, (user_id,)) return bool(result) async def admin_remove(self, user_id: str) -> bool: + """ + Удаляет администратора из таблицы администраторов. + + :param user_id: Идентификатор администратора. + :type user_id: str + :return: Успешность удаления администратора. + :rtype: bool + """ query = "DELETE FROM admins WHERE telegram_id = ?" return await self.execute_query(query, (user_id,)) async def add_black_list(self, command: str) -> bool: + """ + Добавляет команду в черный список. + + :param command: Команда для добавления в черный список. + :type command: str + :return: Успешность добавления команды в черный список. + :rtype: bool + """ query = "INSERT INTO black_list(command) VALUES(?)" return await self.execute_query(query, (command,)) + async def remove_black_list(self, command: str) -> bool: + """ + Удаляет команду из черного списка. + + :param command: Команда для удаления из черного списка. + :type command: str + :return: Успешность удаления команды из черного списка. + :rtype: bool + """ + query = "DELETE FROM black_list WHERE command = ?" + return await self.execute_query(query, (command,)) + async def command_exists(self, command: str) -> bool: + """ + Проверяет, существует ли команда в черном списке. + + :param command: Команда для проверки. + :type command: str + :return: Существует ли команда в черном списке. + :rtype: bool + """ query = "SELECT 1 FROM black_list WHERE command = ?" result = await self.fetch_all(query, (command,)) return bool(result) - async def remove_black_list(self, command: str) -> bool: - query = "DELETE FROM black_list WHERE command = ?" - return await self.execute_query(query, (command,)) - async def commands_all(self) -> str: + """ + Возвращает все команды из черного списка. + + :return: Список команд из черного списка в виде строки. + :rtype: str + """ query = "SELECT command FROM black_list" result = await self.fetch_all(query) return "\n".join([row[0] for row in result]) @@ -132,6 +251,12 @@ def __init__(self): self.con = None async def connect(self) -> None: + """ + Устанавливает соединение с базой данных PostgreSQL и инициализирует таблицы. + + :return: None + :rtype: None + """ try: self.con = await asyncpg.connect( user=getenv("postgre_username"), @@ -146,10 +271,22 @@ async def connect(self) -> None: print(f"Error connecting to the postgresql database: {error}") async def disconnect(self) -> None: + """ + Закрывает соединение с базой данных PostgreSQL. + + :return: None + :rtype: None + """ if self.con: await self.con.close() async def initialize_tables(self) -> None: + """ + Инициализирует таблицы в базе данных, если они еще не существуют. + + :return: None + :rtype: None + """ table_users = """ CREATE TABLE IF NOT EXISTS users( id SERIAL PRIMARY KEY, @@ -168,7 +305,17 @@ async def initialize_tables(self) -> None: await self.execute_query(table_admins) await self.con.commit() - async def execute_query(self, query: str, params=None) -> bool: + async def execute_query(self, query: str, params: Optional[List] = None) -> bool: + """ + Выполняет запрос к базе данных. + + :param query: SQL запрос для выполнения. + :type query: str + :param params: Параметры для запроса. + :type params: Optional[List] + :return: Успешность выполнения запроса. + :rtype: bool + """ try: if not self.con: await self.connect() @@ -179,7 +326,17 @@ async def execute_query(self, query: str, params=None) -> bool: print(f"Error executing PostgreSQL query: {error}") return False - async def fetch_all(self, query: str, params=None) -> list: + async def fetch_all(self, query: str, params: Optional[List] = None) -> List[dict]: + """ + Выполняет запрос к базе данных и возвращает все результаты. + + :param query: SQL запрос для выполнения. + :type query: str + :param params: Параметры для запроса. + :type params: Optional[List] + :return: Список строк результата запроса. + :rtype: List[dict] + """ try: if not self.con: await self.connect() @@ -190,45 +347,123 @@ async def fetch_all(self, query: str, params=None) -> list: return [] async def add_user(self, user_id: str) -> bool: + """ + Добавляет пользователя в таблицу пользователей. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Успешность добавления пользователя. + :rtype: bool + """ query = "INSERT INTO users(telegram_id) VALUES($1)" return await self.execute_query(query, [user_id]) async def user_exists(self, user_id: str) -> bool: + """ + Проверяет, существует ли пользователь в базе данных. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Существует ли пользователь. + :rtype: bool + """ query = "SELECT EXISTS(SELECT 1 FROM users WHERE telegram_id = $1)" result = await self.fetch_all(query, [user_id]) return result[0]["exists"] async def user_remove(self, user_id: str) -> bool: + """ + Удаляет пользователя из таблицы пользователей. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Успешность удаления пользователя. + :rtype: bool + """ query = "DELETE FROM users WHERE telegram_id = $1" return await self.execute_query(query, [user_id]) async def add_admin(self, user_id: str) -> bool: + """ + Добавляет администратора в таблицу администраторов. + + :param user_id: Идентификатор администратора. + :type user_id: str + :return: Успешность добавления администратора. + :rtype: bool + """ query = "INSERT INTO admins(telegram_id) VALUES($1)" return await self.execute_query(query, [user_id]) async def check_admin(self, user_id: str) -> bool: + """ + Проверяет, является ли пользователь администратором. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Является ли пользователь администратором. + :rtype: bool + """ query = "SELECT EXISTS(SELECT 1 FROM admins WHERE telegram_id = $1)" result = await self.fetch_all(query, [user_id]) return result[0]["exists"] async def admin_remove(self, user_id: str) -> bool: + """ + Удаляет администратора из таблицы администраторов. + + :param user_id: Идентификатор администратора. + :type user_id: str + :return: Успешность удаления администратора. + :rtype: bool + """ query = "DELETE FROM admins WHERE telegram_id = $1" return await self.execute_query(query, [user_id]) async def add_black_list(self, command: str) -> bool: + """ + Добавляет команду в черный список. + + :param command: Команда для добавления в черный список. + :type command: str + :return: Успешность добавления команды в черный список. + :rtype: bool + """ query = "INSERT INTO black_list(command) VALUES($1)" return await self.execute_query(query, [command]) async def command_exists(self, command: str) -> bool: + """ + Проверяет, существует ли команда в черном списке. + + :param command: Команда для проверки. + :type command: str + :return: Существует ли команда в черном списке. + :rtype: bool + """ query = "SELECT EXISTS(SELECT 1 FROM black_list WHERE command = $1)" result = await self.fetch_all(query, [command]) return result[0]["exists"] async def remove_black_list(self, command: str) -> bool: + """ + Удаляет команду из черного списка. + + :param command: Команда для удаления из черного списка. + :type command: str + :return: Успешность удаления команды из черного списка. + :rtype: bool + """ query = "DELETE FROM black_list WHERE command = $1" return await self.execute_query(query, [command]) async def commands_all(self) -> str: + """ + Возвращает все команды из черного списка. + + :return: Список команд из черного списка в виде строки. + :rtype: str + """ query = "SELECT command FROM black_list" result = await self.fetch_all(query) return "\n".join([row["command"] for row in result]) @@ -236,6 +471,13 @@ async def commands_all(self) -> str: class DataBase: def __init__(self, db_type: str): + """ + Инициализирует объект базы данных в зависимости от указанного типа. + + :param db_type: Тип базы данных ('sqlite' или 'postgresql'). + :type db_type: str + :raises ValueError: Если указан неподдерживаемый тип базы данных. + """ self.db_type = db_type.lower() if self.db_type == "sqlite": self.database = SqliteDatabase() @@ -247,36 +489,129 @@ def __init__(self, db_type: str): ) async def connect(self) -> None: + """ + Устанавливает соединение с базой данных. + + :return: None + :rtype: None + """ await self.database.connect() async def disconnect(self) -> None: + """ + Закрывает соединение с базой данных. + + :return: None + :rtype: None + """ await self.database.disconnect() async def add_user(self, user_id: str) -> bool: - return await self.database.add_user(user_id) + """ + Добавляет пользователя в базу данных. - async def user_exists(self, user_id: str) -> bool: - return await self.database.user_exists(user_id) + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Успешность добавления пользователя. + :rtype: bool + """ + return await self.database.add_user(user_id) async def user_remove(self, user_id: str) -> bool: + """ + Удаляет пользователя из базы данных. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Успешность удаления пользователя. + :rtype: bool + """ return await self.database.user_remove(user_id) + async def user_exists(self, user_id: str) -> bool: + """ + Проверяет, существует ли пользователь в базе данных. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Существует ли пользователь. + :rtype: bool + """ + return await self.database.user_exists(user_id) + async def add_admin(self, user_id: str) -> bool: + """ + Добавляет администратора в базу данных. + + :param user_id: Идентификатор администратора. + :type user_id: str + :return: Успешность добавления администратора. + :rtype: bool + """ return await self.database.add_admin(user_id) + async def admin_remove(self, user_id: str) -> bool: + """ + Удаляет администратора из базы данных. + + :param user_id: Идентификатор администратора. + :type user_id: str + :return: Успешность удаления администратора. + :rtype: bool + """ + return await self.database.admin_remove(user_id) + async def check_admin(self, user_id: str) -> bool: + """ + Проверяет, является ли пользователь администратором. + + :param user_id: Идентификатор пользователя. + :type user_id: str + :return: Является ли пользователь администратором. + :rtype: bool + """ return await self.database.check_admin(user_id) async def add_black_list(self, command: str) -> bool: + """ + Добавляет команду в черный список. + + :param command: Команда для добавления в черный список. + :type command: str + :return: Успешность добавления команды в черный список. + :rtype: bool + """ return await self.database.add_black_list(command) async def remove_black_list(self, command: str) -> bool: + """ + Удаляет команду из черного списка. + + :param command: Команда для удаления из черного списка. + :type command: str + :return: Успешность удаления команды из черного списка. + :rtype: bool + """ return await self.database.remove_black_list(command) async def command_exists(self, command: str) -> bool: + """ + Проверяет, существует ли команда в черном списке. + + :param command: Команда для проверки. + :type command: str + :return: Существует ли команда в черном списке. + :rtype: bool + """ return await self.database.command_exists(command) async def commands_all(self) -> str: + """ + Возвращает все команды из черного списка. + + :return: Список команд из черного списка в виде строки. + :rtype: str + """ return await self.database.commands_all() diff --git a/resources/config.py b/resources/config.py index a61035c..f038128 100644 --- a/resources/config.py +++ b/resources/config.py @@ -22,33 +22,80 @@ def read_json() -> Dict[str, Any]: + """ + Читает и возвращает содержимое конфигурационного файла в виде словаря. + + :return: Словарь с данными из JSON-файла. + :rtype: Dict[str, Any] + """ with open(path, "r") as file: return json.load(file) def telegram() -> Dict[str, Any]: + """ + Возвращает настройки для Telegram из конфигурационного файла. + + :return: Словарь с настройками Telegram. + :rtype: Dict[str, Any] + """ return read_json()["Telegram"] def database() -> Dict[str, Any]: + """ + Возвращает настройки базы данных из конфигурационного файла. + + :return: Словарь с настройками базы данных. + """ return read_json()["database"] def sqlite() -> Dict[str, Any]: + """ + Возвращает настройки для SQLite из конфигурационного файла. + + :return: Словарь с настройками SQLite. + :rtype: Dict[str, Any] + """ return read_json()["sqlite"] def postgresql() -> Dict[str, Any]: + """ + Возвращает настройки для PostgreSQL из конфигурационного файла. + + :return: Словарь с настройками PostgreSQL. + :rtype: Dict[str, Any] + """ return read_json()["postgresql"] def console() -> Dict[str, Any]: + """ + Возвращает настройки выдачи прав из конфигурационного файла. + + :return: Словарь с настройками выдачи. + :rtype: Dict[str, Any] + """ return read_json()["console"] def logging_config() -> Dict[str, Any]: + """ + Возвращает настройки логирования из конфигурационного файла. + + :return: Словарь с настройками логирования. + :rtype: Dict[str, Any] + """ return read_json()["logging"] def name_fields_table_list_commands() -> Dict[str, Any]: + """ + Возвращает настройки полей таблицы команд из конфигурационного файла. + + :return: Словарь с настройками полей таблицы команд. + :rtype: Dict[str, Any] + """ return read_json()["name_fields_table_list_commands"] diff --git a/tools/table_formatted.py b/tools/table_formatted.py index 267b3de..2d26e81 100644 --- a/tools/table_formatted.py +++ b/tools/table_formatted.py @@ -4,8 +4,16 @@ async def get_commands_table_formatted(commands: str) -> str: + """ + Форматирует список команд в таблицу. + + :param commands: Строка с командами, разделёнными новой строкой. + :return: Строку с таблицей. + :rtype: str + """ table = PrettyTable() table.field_names = config.name_fields_table_list_commands()["name"] + for command in commands.split("\n"): table.add_row([command])