Merge pull request #2 from KerradKerridi/dev-1

Dev-1
This commit was merged in pull request #2.
This commit is contained in:
ANDREY KATYKHIN
2024-07-16 00:36:40 +05:00
committed by GitHub
54 changed files with 2966 additions and 1063 deletions

1
.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
/database/tg-bot-database

BIN
Stick/Hello_11.webp Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 26 KiB

BIN
Stick/Hello_12.webp Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 16 KiB

BIN
Stick/Hello_13.webp Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 23 KiB

BIN
Stick/Universal_11.webp Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 32 KiB

940
database/db.py Normal file
View File

@@ -0,0 +1,940 @@
import sqlite3
import os
from datetime import datetime
from logs.custom_logger import Logger
# Инициализируем логгер
db_logger = Logger(name='db')
# Получение абсолютного пути к текущей директории
current_dir = os.getcwd()
class BotDB:
def __init__(self, name):
self.db_file = os.path.join(current_dir, name)
self.conn = None
self.cursor = None
self.logger = db_logger.get_logger()
self.logger.info(f'Подключен к базе данных: {self.db_file}')
def connect(self):
"""Создание соединения и курсора."""
self.conn = sqlite3.connect(self.db_file)
self.cursor = self.conn.cursor()
def create_table(self, sql_script):
"""
Создает таблицу в базе. Используется в миграциях
Args:
sql_script: DDL скрипт таблицы
Returns:
None
"""
try:
self.connect()
self.cursor.execute(sql_script)
self.conn.commit()
self.logger.info(f'Таблица создана: {sql_script}')
except Exception as e:
self.logger.error(f'Ошибка при создании таблицы. Данные: {sql_script} Ошибка: {e}')
raise
finally:
self.close()
def get_current_version(self):
"""
Возвращает текущую последнюю версию миграции
Args:
None
Returns:
int: Версия последней миграции.
"""
self.logger.info(f'Попытка получения версии миграции')
try:
self.connect()
self.cursor.execute("SELECT version FROM migrations ORDER BY version DESC LIMIT 1")
version = self.cursor.fetchone()[0]
self.logger.info(f'Получена текущая версия миграции: {version}')
return version
except Exception as e:
self.logger.error(f'Ошибка при получении текущей версии миграции: {e}')
raise
finally:
self.close()
def update_version(self, new_version: int, script_name: str):
"""
Обновляет версию миграций в таблице migrations.
Добавляет новую запись в таблицу migrations с указанной версией,
именем скрипта и текущей датой и временем.
Args:
new_version (int): Новая версия миграции
script_name (str): Имя скрипта миграции
Returns:
None
Raises:
sqlite3. IntegrityError: Если возникает ошибка целостности при вставке
данных в таблицу migrations.
Exception: Если возникает любая другая ошибка при обновлении версии.
"""
self.logger.info(f'Попытка обновления версии: {new_version}, название скрипта: {script_name}')
try:
self.connect()
today = datetime.now().strftime("%d-%m-%Y %H:%M:%S")
self.cursor.execute(
"INSERT INTO migrations (version, script_name, created_at) VALUES(?, ?, ?)",
(new_version, script_name, today),
)
self.conn.commit()
self.logger.info(f"Версия обновлена: {new_version}, название скрипта: {script_name}")
except sqlite3.IntegrityError as e:
self.logger.error(f"Ошибка при обновлении версии: {e}")
raise
except Exception as e:
self.logger.error(f"Ошибка при обновлении версии: {e}")
raise
finally:
self.close()
# TODO: Deprecated. Остался только в voice боте, удалить и оттуда
def get_error_message_from_db(self, id: int):
"""
@deprecated
Функция для запроса к базе данных и получения сообщений ошибки. В аргументы передаются:
id - идентификатор ошибки
"""
# Подключаемся к базе
try:
self.connect()
self.cursor.execute(f"SELECT * FROM error_messages WHERE id=?", (id,))
response_from_database = str(self.cursor.fetchone()[1])
return response_from_database
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении сообщения об ошибка voice_bot: {error}")
finally:
self.close()
def add_new_user_in_db(self, user_id: int, first_name: str, full_name: str, username: str, is_bot: bool,
language_code: str, date_added: str,
date_changed: str):
"""
Добавляет нового пользователя в базу данных.
Args:
user_id (int): Идентификатор пользователя в Telegram.
first_name (str): Имя пользователя.
full_name (str): Полное имя пользователя.
username (str): Username пользователя в Telegram.
is_bot (bool): Флаг, указывающий, является ли пользователь ботом.
language_code (str): Код языка пользователя.
date_added (str): Дата добавления пользователя в базу.
date_changed (str): Дата последнего изменения данных пользователя.
Returns:
None: Если запись успешно добавлена в базу.
Exception: Если произошла ошибка при добавлении записи.
"""
self.logger.info(f"Попытка добавить пользователя в базу данных: user_id={user_id}, first_name={first_name}")
try:
self.connect()
self.cursor.execute("INSERT INTO 'our_users' ('user_id', 'first_name', 'full_name', 'username', 'is_bot', "
"'language_code', 'date_added', 'date_changed') VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(user_id, first_name, full_name,
username, is_bot, language_code, date_added, date_changed))
self.conn.commit()
self.logger.info(f"Новый пользователь добавлен в базу: user_id={user_id}, first_name={first_name}")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при добавлении пользователя в базу: {error}. "
f"Данные пользователя: user_id={user_id}, first_name={first_name}")
raise
finally:
self.close()
def user_exists(self, user_id: int):
"""
Проверяет, существует ли пользователь в базе данных.
Args:
user_id (int): Идентификатор пользователя.
Returns:
bool: True, если пользователь найден, False - иначе.
"""
self.logger.info(f"Попытка проверки существования пользователя: user_id={user_id}")
try:
self.connect()
self.cursor.execute("SELECT id FROM our_users WHERE user_id = ?", (user_id,))
result = self.cursor.fetchall()
self.logger.info(f"Проверка существования пользователя: user_id={user_id}, результат={result}")
return bool(len(result))
except sqlite3.Error as error:
self.logger.error(f"Ошибка при проверке существования пользователя: {error}")
raise
finally:
self.close()
def get_user_id(self, user_id: int):
"""
@deprecated
Возвращает ID пользователя в базе данных по его user_id.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
int: ID пользователя в базе данных.
None: Если пользователь не найден.
"""
self.logger.info(f"Попытка получения ID пользователя в базе данных для user_id={user_id}")
try:
self.connect()
self.cursor.execute("SELECT id FROM our_users WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
if result:
user_id_db = result[0]
self.logger.info(f"ID пользователя в базе найден: user_id={user_id}, id_db={user_id_db}")
return user_id_db
else:
self.logger.info(f"Пользователь с user_id={user_id} не найден в базе данных.")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении ID пользователя из базы данных: {error}")
raise
finally:
self.close()
def get_username(self, user_id: int):
"""
Возвращает username пользователя из базы данных по его user_id в Telegram.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
str: Username пользователя.
None: Если пользователь не найден.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
try:
self.connect()
self.cursor.execute("SELECT username FROM our_users WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
if result:
username = result[0]
self.logger.info(f"Username пользователя найден: user_id={user_id}, username={username}")
return username
else:
self.logger.info(f"Пользователь с user_id={user_id} не найден в базе данных.")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении username из базы данных: {error}")
raise
finally:
self.close()
def get_all_user_id(self):
"""
Возвращает список всех user_id из базы данных.
Returns:
list: Список user_id.
[]: Если в базе данных нет пользователей.
Raises:
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Попытка получения всех user_id")
try:
self.connect()
self.cursor.execute("SELECT user_id FROM our_users")
fetch_all = self.cursor.fetchall()
list_of_users = [user_id[0] for user_id in fetch_all]
self.logger.info(f"Получен список всех user_id: {list_of_users}")
return list_of_users
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении списка user_id из базы данных: {error}")
raise
finally:
self.close()
def get_user_first_name(self, user_id: int):
"""
Возвращает имя пользователя из базы данных по его user_id в Telegram.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
str: Имя пользователя.
None: Если пользователь не найден.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Попытка получения имени пользователя по user_id={user_id}")
try:
self.connect()
self.cursor.execute("SELECT first_name FROM our_users WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
if result:
first_name = result[0]
self.logger.info(f"Имя пользователя найдено: user_id={user_id}, first_name={first_name}")
return first_name
else:
self.logger.info(f"Пользователь с user_id={user_id} не найден в базе данных.")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении имени пользователя из базы данных: {error}")
raise
finally:
self.close()
def change_name(self, user_id):
#TODO: реализовать функцию изменения имени пользователя по которому к нему обращается бот. Обновляем поля first_name, date_changed
#result = self.cursor.execute("UPDATE 'our_users' SET (?) WHERE user_id = (?)", (new_user_name), )
pass
def get_info_about_stickers(self, user_id: int):
"""
Проверяет, получил ли пользователь стикеры.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
bool: True, если пользователь получил стикеры, False - иначе.
None: Если пользователь не найден.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Попытка проверки получил ли пользователь с user_id={user_id} стикеры.")
try:
self.connect()
self.cursor.execute("SELECT has_stickers FROM our_users WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
if result:
has_stickers = result[0] == 1
self.logger.info(
f"Проверено получение стикеров пользователем: user_id={user_id}, has_stickers={has_stickers}")
return has_stickers
else:
self.logger.info(f"Пользователь с user_id={user_id} не найден в базе данных.")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении информации о получении стикеров: {error}")
raise
finally:
self.close()
def update_info_about_stickers(self, user_id):
"""
Обновляет информацию о получении стикеров пользователем.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
None: Если обновление успешно выполнено.
Raises:
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции update_info_about_stickers. Параметры: user_id={user_id}")
try:
self.connect()
self.cursor.execute("UPDATE our_users SET has_stickers = 1 WHERE user_id = ?", (user_id,))
self.conn.commit()
self.logger.info(f"Информация о получении стикеров обновлена: user_id={user_id}")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при обновлении информации о получении стикеров: {error}")
raise
finally:
self.close()
def get_users_blacklist(self):
"""
Возвращает список пользователей в черном списке.
Returns:
dict: Словарь, где ключ - user_id, значение - username.
{}: Если в черном списке нет пользователей.
Raises:
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции get_users_blacklist")
try:
self.connect()
self.cursor.execute("SELECT user_id, user_name FROM blacklist")
fetch_all = self.cursor.fetchall()
list_of_users = {user_id: username for user_id, username in fetch_all}
self.logger.info(f"Получен список пользователей в черном списке")
return list_of_users
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении списка пользователей в черном списке: {error}")
raise
finally:
self.close()
def get_users_for_unblock_today(self, date_to_unban: str):
"""
Возвращает список пользователей, у которых истекает срок блокировки сегодня.
Args:
date_to_unban (str): Дата разблокировки.
Returns:
dict: Словарь, где ключ - user_id, значение - username.
{}: Если сегодня нет пользователей, у которых истекает срок блокировки.
Raises:
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции get_users_for_unblock_today: date_to_unban={date_to_unban}")
try:
self.connect()
result = self.cursor.execute("SELECT user_id, user_name "
"FROM blacklist WHERE date_to_unban = ?", (date_to_unban,))
fetch_all = result.fetchall()
list_of_users = {user_id: username for user_id, username in fetch_all}
self.logger.info(f"Получен список пользователей для разблокировки сегодня: {list_of_users}")
return list_of_users
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении списка пользователей для разблокировки: {error}")
raise
finally:
self.close()
def get_blacklist_users_by_id(self, user_id: int):
"""
Возвращает информацию о пользователе в черном списке по user_id.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
tuple: Кортеж (user_id, user_name, message_for_user, date_to_unban).
None: Если пользователь не найден в черном списке.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции get_blacklist_users_by_id: user_id={user_id}")
try:
self.connect()
result = self.cursor.execute("SELECT user_id, user_name, message_for_user, date_to_unban "
"FROM blacklist WHERE user_id = ?", (user_id,))
return self.cursor.fetchone()
except sqlite3.Error as error:
self.logger.error(f"Ошибка при получении информации о пользователе в черном списке: {error}")
raise
finally:
self.close()
def check_user_in_blacklist(self, user_id: int):
"""
Проверяет, существует ли запись с данным user_id в blacklist.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
bool: True, если пользователь найден в черном списке, False - иначе.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции check_user_in_blacklist: user_id={user_id}")
try:
self.connect()
self.cursor.execute("SELECT 1 FROM blacklist WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
self.logger.info(f"Существует ли пользователь: user_id={user_id} Итог: {result}")
return bool(result)
except sqlite3.Error as error:
self.logger.error(f"Ошибка при проверке пользователя в черном списке. user_id: {user_id} : {error}")
raise
finally:
self.close()
def set_user_blacklist(self, user_id: int, user_name=None, message_for_user=None, date_to_unban=None):
"""
Добавляет пользователя в черный список.
Args:
user_id (int): Идентификатор пользователя в Telegram.
user_name (str, optional): Username пользователя. Defaults to None.
message_for_user (str, optional): Сообщение для пользователя. Defaults to None.
date_to_unban (datetime, optional): Дата разблокировки. Defaults to None.
Returns:
None: Если добавление в черный список успешно выполнено.
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции set_user_blacklist: user_id={user_id}, user_name={user_name},"
f" message_for_user={message_for_user}, date_to_unban={date_to_unban}")
try:
self.connect()
result = self.cursor.execute("INSERT INTO 'blacklist' ('user_id', 'user_name',"
" 'message_for_user', 'date_to_unban') VALUES (?, ?, ?, ?)",
(user_id, user_name, message_for_user, date_to_unban,))
self.conn.commit()
self.logger.info(f"Пользователь добавлен в черный список: user_id={user_id}")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка при добавлении пользователя в черный список: {error}")
return error
finally:
self.close()
def delete_user_blacklist(self, user_id: int):
"""
Удаляет пользователя из черного списка.
Args:
user_id (int): Идентификатор пользователя в Telegram.
Returns:
bool: True, если удаление прошло успешно, False - в случае ошибки.
Raises:
None: Ошибки обрабатываются в блоке except, возвращая False.
"""
self.logger.info(f"Запуск функции delete_user_blacklist: user_id={user_id}")
try:
self.connect()
self.cursor.execute("DELETE FROM blacklist WHERE user_id = ?", (user_id,))
self.conn.commit()
self.logger.info(f"Пользователь с идентификатором {user_id} успешно удален из черного списка.")
return True
except sqlite3.Error as error:
self.logger.error(f"Ошибка удаления пользователя с идентификатором {user_id} "
f"из таблицы blacklist. Ошибка: {str(error)}")
return False
finally:
self.close()
def add_new_message_in_db(self, message_text: str, user_id: int, message_id: int, date: str):
"""
Добавляет новое сообщение пользователя в базу данных.
Args:
message_text (str): Текст сообщения.
user_id (int): Идентификатор пользователя в Telegram.
message_id (int): Идентификатор сообщения в Telegram.
date (str): Дата отправки сообщения.
Returns:
None: Если добавление прошло успешно.
sqlite3.Error: Если произошла ошибка при выполнении запроса.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(
f"Запуск функции add_new_message_in_db: user_id={user_id}, message_id={message_id}, date={date}")
try:
self.connect()
self.cursor.execute(
"INSERT INTO user_messages (message_text, user_id, message_id, date) "
"VALUES (?, ?, ?, ?)",
(message_text, user_id, message_id, date))
self.conn.commit()
self.logger.info(f"Новое сообщение добавлено в базу данных: message_id={message_id}")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка добавления сообщения в базу данных: {error}")
raise
finally:
self.close()
def get_username_and_full_name(self, user_id: int):
"""
Получает full_name и username пользователя по ID из базы
Args:
date (str): Новая дата изменения.
user_id (int): Идентификатор пользователя в Telegram.
Returns:
username (str): username пользователя
full_name (str): full_name пользователя
"""
self.logger.info(
f"Запуск функции check_username_and_first_name: user_id={user_id}")
try:
self.connect()
self.cursor.execute("SELECT username FROM our_users WHERE user_id = ?", (user_id,))
username = self.cursor.fetchone()[0]
self.cursor.execute("SELECT full_name FROM our_users WHERE user_id = ?", (user_id,))
full_name = self.cursor.fetchone()[0]
self.logger.info(
f"Функция check_username_and_first_name успешно отработала: user_id={user_id}, username={username}, full_name={full_name}")
return username, full_name
except sqlite3.Error as error:
self.logger.error(f"Ошибка в функции get_username_and_first_name: {error}")
return None
finally:
self.close()
def update_username_and_full_name(self, user_id: int, username: str, full_name: str):
"""
Обновляет full_name и username пользователя
Args:
username (str): username пользователя
full_name (str): full_name пользователя
user_id (int): Идентификатор пользователя в Telegram
Returns:
True (bool): Если обновления прошли успешно
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(
f"Запуск функции update_username_and_full_name: user_id={user_id}, username={username}, full_name={full_name}")
try:
self.connect()
self.cursor.execute("UPDATE our_users SET username = ?, full_name = ? WHERE user_id = ?", (username, full_name, user_id,))
self.conn.commit()
self.logger.info(f"Функция update_username_and_full_name. Данные пользователя: user_id={user_id} успешно обновлены")
return True
except sqlite3.Error as error:
self.logger.error(f"Ошибка в функции update_username_and_full_name: {error}")
raise
finally:
self.close()
def update_date_for_user(self, date: str, user_id: int):
"""
#TODO: Не возвращается ошибка sqlite3. Error. Тест не перехватывает. Возвращается no such table: our_users
Обновляет дату последнего изменения данных пользователя в базе.
Args:
date (str): Новая дата изменения.
user_id (int): Идентификатор пользователя в Telegram.
Returns:
None: Если обновление прошло успешно.
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции update_date_for_user: user_id={user_id}, date={date}")
try:
self.connect()
self.cursor.execute("UPDATE our_users SET date_changed = ? WHERE user_id = ?",
(date, user_id,))
self.conn.commit()
self.logger.info(f"Дата изменения обновлена для пользователя: user_id={user_id}")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка обновления даты изменения для пользователя: {error}")
return error
finally:
self.close()
def is_admin(self, user_id: int):
"""
Проверяет, является ли пользователь администратором.
Args:
user_id: ID пользователя Telegram.
Returns:
True, если пользователь администратор, иначе False.
Raises:
None: В случае ошибки возвращается None
"""
self.logger.info(f"Запуск функции is_admin: user_id={user_id}")
try:
self.connect()
self.cursor.execute("SELECT 1 FROM admins WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
return bool(result)
except sqlite3.Error as error:
self.logger.error(f"Ошибка добавления сообщения в базу данных: {error}")
return None
finally:
self.close()
def add_admin(self, user_id: int, role: str):
"""
Добавляет пользователя в список администраторов.
Args:
user_id (int): ID пользователя Telegram.
role (str): Роль пользователя. Доступные варианты:
1. creator - создатель
2. admin - обычная роль
Returns:
None: Если добавление прошло успешно.
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции add_admin: user_id={user_id}, role={role}")
try:
self.connect()
self.cursor.execute("INSERT INTO admins (user_id, role) VALUES (?, ?)", (user_id, role))
self.conn.commit()
self.logger.info(f"Пользователь с user_id={user_id} добавлен в список администраторов с ролью {role}.")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка добавления пользователя в список администраторов: {error}")
return error
finally:
self.close()
def remove_admin(self, user_id: int):
"""
Удаляет пользователя из списка администраторов.
Args:
user_id (int): ID пользователя Telegram.
Returns:
None: Если удаление прошло успешно.
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции remove_admin: user_id={user_id}")
try:
self.connect()
self.cursor.execute("DELETE FROM admins WHERE user_id = ?", (user_id,))
self.conn.commit()
self.logger.info(f"Пользователь с user_id={user_id} удален из списка администраторов.")
return None
except sqlite3.Error as error:
self.logger.error(f"Ошибка удаления пользователя из списка администраторов: {error}")
return error
finally:
self.connect()
def get_user_by_message_id(self, message_id: int):
"""
#TODO: Возвращается TypeError вместо None
Возвращает идентификатор пользователя по идентификатору сообщения.
Args:
message_id (int): Идентификатор сообщения в Telegram.
Returns:
int: Идентификатор пользователя.
None: Если пользователь не найден.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции get_user_by_message_id: message_id={message_id}")
try:
self.connect()
result = self.cursor.execute("SELECT user_id FROM user_messages WHERE message_id = ?", (message_id,))
user = result.fetchone()[0]
self.logger.info(f"Пользователь успешно получен user_id={user} по message_id={message_id}")
return user
except sqlite3.Error as error:
self.logger.error(f"Ошибка получения user_id по message_id: {error}")
raise
finally:
self.close()
def get_last_users_from_db(self):
"""
Возвращает список идентификаторов последних 30 пользователей, обращавшихся в бот.
Returns:
list: Список кортежей (full_name, user_id) последних 30 пользователей.
[]: Если в базе данных нет пользователей.
Raises:
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info("Запуск функции get_last_users_from_db")
try:
self.connect()
result = self.cursor.execute("SELECT full_name, user_id FROM our_users ORDER BY date_changed DESC LIMIT 30")
users = result.fetchall()
self.logger.info(f"Получен список последних 30 пользователей: {users}")
return users
except sqlite3.Error as error:
self.logger.error(f"Ошибка получения списка последних пользователей: {error}")
raise
finally:
self.close()
def get_banned_users_from_db(self):
"""
Возвращает список идентификаторов пользователей в черном списке бота.
Returns:
list: Список кортежей (user_name, user_id, message_for_user, date_to_unban) пользователей в черном списке.
[]: Если в черном списке нет пользователей.
Raises:
sqlite3.Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info("Запуск функции get_banned_users_from_db")
try:
self.connect()
result = self.cursor.execute("SELECT user_name, user_id, message_for_user, date_to_unban FROM blacklist")
users = result.fetchall()
self.logger.info(f"Получен список пользователей в черном списке: {users}")
return users
except sqlite3.Error as error:
self.logger.error(f"Ошибка получения списка пользователей в черном списке: {error}")
raise
finally:
self.close()
def get_banned_users_from_db_with_limits(self, offset: int, limit: int):
"""
Возвращает список идентификаторов пользователей в черном списке бота с учетом смещения и ограничения.
Args:
offset (int): Смещение для выборки
limit (int): Ограничение количества возвращаемых записей.
Returns:
list: Список кортежей (user_name, user_id, message_for_user, date_to_unban) пользователей в черном списке.
[]: Если в черном списке нет пользователей или количество записей с учетом смещения и ограничения равно 0.
Raises:
sqlite3. Error: Если произошла ошибка при выполнении запроса.
"""
self.logger.info(f"Запуск функции get_banned_users_from_db_with_limits: offset={offset}, limit={limit}")
try:
self.connect()
result = self.cursor.execute("SELECT user_name, user_id, message_for_user, date_to_unban "
"FROM blacklist LIMIT ?, ?", (offset, limit,))
users = result.fetchall()
self.logger.info(f"Получен список пользователей в черном списке (offset={offset}, limit={limit}): {users}")
return users
except sqlite3.Error as error:
self.logger.error(f"Ошибка получения списка пользователей в черном списке: {error}")
raise
finally:
self.close()
def add_audio_record(self, file_name, author_id, date_added, listen_count, file_id):
"""Добавляет информацию о войсе юзера в БД"""
self.logger.info(
f"Запуск функции add_audio_record (file_name = {file_name}, author_id = {author_id}, date_added = {date_added}")
try:
self.connect()
result = self.cursor.execute(
"INSERT INTO `audio_message_reference` (file_name, author_id, date_added, listen_count, file_id) VALUES (?, ?, ?, ?, ?)",
(file_name, author_id, date_added, listen_count, file_id))
self.conn.commit()
self.logger.info(
f"Аудио успешно добавлено в БД (file_name = {file_name}, author_id = {author_id}, date_added = {date_added}")
return None
except sqlite3.Error as error:
print(error)
raise
finally:
self.close()
def last_date_audio(self):
"""Получаем дату последнего войса"""
try:
self.connect()
result = self.cursor.execute(
"SELECT `date_added` FROM `audio_message_reference` ORDER BY date_added DESC LIMIT 1")
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
finally:
self.close()
def get_last_user_audio_record(self, user_id):
"""Получает данные о количестве записей пользователя"""
try:
self.connect()
result = self.cursor.execute("SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ?",
(user_id,))
return bool(len(result.fetchall()))
except sqlite3.Error as error:
print(error)
finally:
self.close()
def get_id_for_audio_record(self, user_id):
"""Получает ID аудио сообщения пользователя"""
try:
self.connect()
result = self.cursor.execute(
"SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ? ORDER BY date_added DESC LIMIT 1",
(user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
finally:
self.close()
def get_path_for_audio_record(self, user_id):
"""Получает данные о названии файла"""
try:
self.connect()
result = self.cursor.execute(
"SELECT `file_name` FROM `audio_message_reference` WHERE `author_id` = ? ORDER BY date_added DESC LIMIT 1",
(user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
finally:
self.close()
def check_listen_audio(self, user_id):
"""Проверяет прослушано ли аудио пользователем"""
try:
self.connect()
query_listen_audio = self.cursor.execute(
"""SELECT l.file_name
FROM audio_message_reference a
LEFT JOIN listen_audio_users l ON l.file_name = a.file_name
WHERE l.user_id = ?
AND l.file_name IS NOT NULL""", (user_id,))
check_sign = query_listen_audio.fetchall()
query_all_audio = self.cursor.execute('SELECT file_name FROM audio_message_reference WHERE author_id <> ?',
(user_id,))
sign_all_audio = query_all_audio.fetchall()
new_sign1 = list(set(sign_all_audio) - set(check_sign))
new_sign = []
for i in new_sign1:
new_sign.append(i[0])
return new_sign
except sqlite3.Error as error:
print(error)
finally:
self.close()
def mark_listened_audio(self, file_name, user_id):
"""Отмечает аудио прослушанным для конкретного пользователя."""
try:
self.connect()
result = self.cursor.execute(
"INSERT INTO `listen_audio_users` (file_name, user_id, is_listen) VALUES (?, ?, ?)",
(file_name, user_id, 1))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
finally:
self.close()
def close(self):
"""Закрытие соединения и курсора."""
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()

406
db.py
View File

@@ -1,406 +0,0 @@
import sqlite3
import configparser
import os
from datetime import datetime
from loguru import logger
from custom_logger import Logger
db_logger = Logger(name='db')
script_dir = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(script_dir, 'settings.ini')
config = configparser.ConfigParser()
config.read(config_path)
LOGS = config.getboolean('Settings', 'logs')
IMPORTANT_LOGS = config.get('Telegram', 'important_logs')
class BotDB:
def __init__(self):
db_file_path = os.path.dirname(os.path.abspath(__file__))
db_file = os.path.join(db_file_path, 'tg-bot-database')
self.conn = sqlite3.connect(db_file, check_same_thread=False)
self.cursor = self.conn.cursor()
logger.info(f'Подключен к базе данных: {db_file_path}')
def create_table(self, sql_script):
"""Создает таблицу в базе."""
try:
cursor = self.conn.cursor()
cursor.execute(sql_script)
except Exception as e:
print(f"Ошибка при создании таблицы: {e}")
def get_current_version(self):
"""Получает текущую версию миграций из таблицы migrations."""
try:
cursor = self.conn.cursor()
cursor.execute("SELECT version FROM migrations ORDER BY version DESC LIMIT 1")
version = cursor.fetchone()[0]
return version
except Exception as e:
print(f"Ошибка при получении версии: {e}")
return 0
def update_version(self, new_version, script_name):
"""Обновляет версию миграций в таблице migrations."""
logger.info(f'Попытка обновления версии: {new_version}, скрипт: {script_name}')
try:
current_date = datetime.now()
today = current_date.strftime("%d-%m-%Y %H:%M:%S")
cursor = self.conn.cursor()
cursor.execute(
"INSERT INTO migrations (version, script_name, created_at) VALUES(?, ?, ?)",
(new_version, script_name, today),
)
self.conn.commit()
logger.info(f"Версия обновлена: {new_version}, скрипт: {script_name}")
except sqlite3.IntegrityError as e:
logger.error(f"Ошибка при обновлении версии: {e}")
except Exception as e:
logger.error(f"Ошибка при обновлении версии: {e}")
# TODO: Deprecated, удалить
def get_message_from_db(self, type: str, username):
"""Функция для запроса к базе данных и получения сообщений для бота. В аргументы передаются:
type - тип получаемой обратной связи, строковое значение, сохраненное в БД
username - имя пользователя
"""
# Подключаемся к базе
try:
cursor = self.conn.cursor()
cursor.execute(f"SELECT * FROM messages WHERE type=?", (type,))
# Забираем данные из таблицы, преобразуем в строку, заменяем поле username на имя пользователя,
# и вместо амберсанда подставляем перенос строки
if type == 'connect_with_admin' or type == 'del_message' or type == 'suggest_news' or type == 'start_message':
response_from_database = str(cursor.fetchone()[1]).replace('username', username).replace('&', '\n')
else:
response_from_database = str(cursor.fetchone()[1]).replace('&', '\n')
return response_from_database
except sqlite3.Error as error:
print(error)
# TODO: Deprecated. Остался только в voice боте, удалить и оттуда
def get_error_message_from_db(self, id: int):
"""Функция для запроса к базе данных и получения сообщений ошибки. В аргументы передаются:
id - идентификатор ошибки
"""
# Подключаемся к базе
try:
cursor = self.conn.cursor()
cursor.execute(f"SELECT * FROM error_messages WHERE id=?", (id,))
response_from_database = str(cursor.fetchone()[1])
return response_from_database
except sqlite3.Error as error:
print(error)
def add_new_user_in_db(self, user_id, first_name, full_name, username, is_bot, language_code, date_added,
date_changed):
"""Добавляем юзера в базу"""
try:
self.cursor.execute("INSERT INTO 'our_users' ('user_id', 'first_name', 'full_name', 'username', 'is_bot', "
"'language_code', 'date_added', 'date_changed') VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
(user_id, first_name, full_name,
username, is_bot, language_code, date_added, date_changed))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
def user_exists(self, user_id):
"""Проверяем, есть ли юзер в базе"""
try:
result = self.cursor.execute("SELECT `id` FROM `our_users` WHERE `user_id` = ?", (user_id,))
return bool(len(result.fetchall()))
except sqlite3.Error as error:
print(error)
def get_user_id(self, user_id):
"""Достаем id юзера в базе по его user_id"""
try:
result = self.cursor.execute("SELECT `id` FROM `our_users` WHERE `user_id` = ?", (user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def get_username(self, user_id):
"""Достаем id юзера в базе по его user_id"""
try:
result = self.cursor.execute("SELECT `username` FROM `our_users` WHERE `user_id` = ?", (user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def get_all_user_id(self):
"""Достаем все айдишники юзеров из БД и преобразуем их в список"""
try:
result = self.cursor.execute("SELECT `user_id` FROM `our_users`", )
fetch_all = result.fetchall()
list_of_users = []
for i in fetch_all:
list_of_users.append(i[0])
return list_of_users
except sqlite3.Error as error:
print(error)
def get_user_first_name(self, user_id):
try:
result = self.cursor.execute("SELECT `first_name` FROM `our_users` WHERE `user_id` = ?", (user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def change_name(self, user_id):
#TODO: реализовать функцию изменения имени пользователя по которому к нему обращается бот. ОБновляем поля first_name, date_changed
#result = self.cursor.execute("UPDATE 'our_users' SET (?) WHERE user_id = (?)", (new_user_name), )
pass
def add_audio_record(self, file_name, author_id, date_added, listen_count, file_id):
"""Добавляет информацию о войсе юзера в БД"""
try:
result = self.cursor.execute(
"INSERT INTO `audio_message_reference` (file_name, author_id, date_added, listen_count, file_id) VALUES (?, ?, ?, ?, ?)",
(file_name, author_id, date_added, listen_count, file_id))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
def last_date_audio(self):
"""Получаем дату последнего войса"""
try:
result = self.cursor.execute(
"SELECT `date_added` FROM `audio_message_reference` ORDER BY date_added DESC LIMIT 1")
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def get_last_user_audio_record(self, user_id):
"""Получает данные о количестве записей пользователя"""
try:
result = self.cursor.execute("SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ?",
(user_id,))
return bool(len(result.fetchall()))
except sqlite3.Error as error:
print(error)
def get_id_for_audio_record(self, user_id):
"""Получает ID аудио сообщения пользователя"""
try:
result = self.cursor.execute(
"SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ? ORDER BY date_added DESC LIMIT 1",
(user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def get_path_for_audio_record(self, user_id):
"""Получает данные о названии файла"""
try:
result = self.cursor.execute(
"SELECT `file_name` FROM `audio_message_reference` WHERE `author_id` = ? ORDER BY date_added DESC LIMIT 1",
(user_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def check_listen_audio(self, user_id):
"""Проверяет прослушано ли аудио пользователем"""
try:
query_listen_audio = self.cursor.execute(
"""SELECT l.file_name
FROM audio_message_reference a
LEFT JOIN listen_audio_users l ON l.file_name = a.file_name
WHERE l.user_id = ?
AND l.file_name IS NOT NULL""", (user_id,))
check_sign = query_listen_audio.fetchall()
query_all_audio = self.cursor.execute('SELECT file_name FROM audio_message_reference WHERE author_id <> ?',
(user_id,))
sign_all_audio = query_all_audio.fetchall()
new_sign1 = list(set(sign_all_audio) - set(check_sign))
new_sign = []
for i in new_sign1:
new_sign.append(i[0])
return new_sign
except sqlite3.Error as error:
print(error)
def mark_listened_audio(self, file_name, user_id):
"""Отмечает аудио прослушанным для конкретного пользователя."""
try:
result = self.cursor.execute(
"INSERT INTO `listen_audio_users` (file_name, user_id, is_listen) VALUES (?, ?, ?)",
(file_name, user_id, 1))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
def get_info_about_stickers(self, user_id):
"""Получает данные о получении стикеров пользователем"""
try:
result = self.cursor.execute("SELECT `has_stickers` FROM `our_users` WHERE `user_id` = ?", (user_id,))
return result.fetchone()[0] == 1
#return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def update_info_about_stickers(self, user_id):
"""Обновляет данные о получении стикеров пользователем"""
try:
result = self.cursor.execute("UPDATE `our_users` SET `has_stickers` = 1 WHERE `user_id` = ?", (user_id,))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
def get_users_blacklist(self):
"""Возвращает список пользователей в черном списке"""
try:
result = self.cursor.execute("SELECT user_id, user_name FROM `blacklist`")
fetch_all = result.fetchall()
list_of_users = {}
for i in fetch_all:
list_of_users[i[0]] = i[1]
return list_of_users
except sqlite3.Error as error:
print(error)
def get_users_for_unblock_today(self, date_to_unban):
"""Возвращает пользователей у которых истекает срок блокировки сегодня"""
try:
result = self.cursor.execute("SELECT user_id, user_name FROM `blacklist` WHERE date_to_unban = ?", (date_to_unban,))
fetch_all = result.fetchall()
list_of_users = {}
for i in fetch_all:
list_of_users[i[0]] = i[1]
return list_of_users
except sqlite3.Error as error:
print(error)
def get_blacklist_users_by_id(self, user_id):
"""Возвращает список пользователей в черном списке по user_id"""
try:
result = self.cursor.execute("SELECT user_id, user_name, message_for_user, date_to_unban FROM `blacklist` WHERE user_id = ?", (user_id, ))
return self.cursor.fetchone()
except sqlite3.Error as error:
print(error)
def check_user_in_blacklist(self, user_id):
"""Проверяет, существует ли запись с данным user_id в blacklist."""
self.cursor.execute("SELECT 1 FROM blacklist WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
return bool(result)
def set_user_blacklist(self, user_id, user_name=None, message_for_user=None, date_to_unban=None):
"""Добавляет пользователя в черный список"""
try:
result = self.cursor.execute("INSERT INTO 'blacklist' ('user_id', 'user_name',"
" 'message_for_user', 'date_to_unban') VALUES (?, ?, ?, ?)",
(user_id, user_name, message_for_user, date_to_unban,))
return self.conn.commit()
except sqlite3.Error as error:
return error
def delete_user_blacklist(self, user_id):
"""Удаляет пользователя из черного списка"""
try:
#TODO: Функция всегда возвращает true, даже если такого id нет в таблице
self.cursor.execute("DELETE FROM blacklist WHERE user_id = ?", (user_id,))
self.conn.commit()
logger.info(f"Пользователь с идентификатором {user_id} успешно удален.")
return True
except sqlite3.Error as error:
logger.error(f"Ошибка удаления пользователя с идентификатором {user_id} из таблицы blacklist. Ошибка: {str(error)}")
return False
def add_new_message_in_db(self, message_text, user_id, message_id, date):
"""Добавляем сообщение юзера в базу"""
try:
self.cursor.execute(
"INSERT INTO `user_messages` (message_text, user_id, message_id, date) "
"VALUES (?, ?, ?, ?)",
(message_text, message_id, user_id, date))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
def update_date_for_user(self, date, user_id: int):
try:
result = self.cursor.execute("UPDATE `our_users` SET `date_changed` = ? WHERE `user_id` = ?",
(date, user_id,))
return self.conn.commit()
except sqlite3.Error as error:
print(error)
def is_admin(self, user_id):
"""
Проверяет, является ли пользователь администратором.
Args:
user_id: ID пользователя Telegram.
Returns:
True, если пользователь администратор, иначе False.
"""
self.cursor.execute("SELECT 1 FROM admins WHERE user_id = ?", (user_id,))
result = self.cursor.fetchone()
return bool(result)
def add_admin(self, user_id, role):
"""
Добавляет пользователя в список администраторов.
Args:
user_id: ID пользователя Telegram.
role: Роль пользователя.
Доступные варианты:
1. creator - создатель
2. admin - обычная роль
"""
self.cursor.execute("INSERT INTO admins (user_id, role) VALUES (?, ?)", (user_id, role))
return self.conn.commit()
def remove_admin(self, user_id):
"""
Удаляет пользователя из списка администраторов.
Args:
user_id: ID пользователя Telegram.
"""
self.cursor.execute("DELETE FROM admins WHERE user_id = ?", (user_id,))
return self.conn.commit()
def get_user_by_message_id(self, message_id):
"""Возвращает идентификатор пользователя по идентификатору сообщения"""
try:
result = self.cursor.execute("SELECT user_id FROM `user_messages` WHERE message_id = ?", (message_id,))
return result.fetchone()[0]
except sqlite3.Error as error:
print(error)
def get_last_users_from_db(self):
"""Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот"""
try:
result = self.cursor.execute("SELECT full_name, user_id FROM our_users ORDER BY date_changed DESC")
users = result.fetchall()
return users
except sqlite3.Error as error:
print(error)
def get_banned_users_from_db(self):
"""Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот"""
try:
result = self.cursor.execute("SELECT user_name, user_id, message_for_user, date_to_unban FROM blacklist", )
users = result.fetchall()
return users
except sqlite3.Error as error:
print(error)
def get_banned_users_from_db_with_limits(self, offset: int, limit: int):
"""Возвращает список идентификаторов последних 10 пользователей обращавшихся в бот"""
try:
result = self.cursor.execute("SELECT user_name, user_id, message_for_user, date_to_unban FROM blacklist LIMIT ?, ?", (offset, limit,) )
users = result.fetchall()
return users
except sqlite3.Error as error:
print(error)
def close(self):
"""Закрываем соединение с БД"""
self.conn.close()

0
helper_bot/__init__.py Normal file
View File

View File

@@ -0,0 +1,15 @@
from typing import Union
from aiogram.filters import BaseFilter
from aiogram.types import Message
class ChatTypeFilter(BaseFilter): # [1]
def __init__(self, chat_type: Union[str, list]): # [2]
self.chat_type = chat_type
async def __call__(self, message: Message) -> bool: # [3]
if isinstance(self.chat_type, str):
return message.chat.type == self.chat_type
else:
return message.chat.type in self.chat_type

View File

@@ -0,0 +1 @@
from .main import admin_router

View File

@@ -0,0 +1,143 @@
import traceback
from aiogram import Router, types, F
from aiogram.filters import Command, StateFilter
from aiogram.fsm.context import FSMContext
from helper_bot.filters.main import ChatTypeFilter
from helper_bot.keyboards.main import get_reply_keyboard_admin, create_keyboard_with_pagination, \
create_keyboard_for_ban_days, create_keyboard_for_approve_ban
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
from helper_bot.utils.helper_func import check_access, add_days_to_date, get_banned_users_buttons, get_banned_users_list
from logs.custom_logger import Logger
from database.db import BotDB
admin_router = Router()
#Инициализируем логгер
admin_logger = Logger(name='admin_handler')
logger = admin_logger.get_logger()
bdf = BaseDependencyFactory()
GROUP_FOR_POST = bdf.settings['Telegram']['group_for_posts']
GROUP_FOR_MESSAGE = bdf.settings['Telegram']['group_for_message']
MAIN_PUBLIC = bdf.settings['Telegram']['main_public']
GROUP_FOR_LOGS = bdf.settings['Telegram']['group_for_logs']
IMPORTANT_LOGS = bdf.settings['Telegram']['important_logs']
PREVIEW_LINK = bdf.settings['Telegram']['preview_link']
LOGS = bdf.settings['Settings']['logs']
TEST = bdf.settings['Settings']['test']
BotDB = BotDB('database/tg-bot-database')
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
Command('admin')
)
async def admin_panel(message: types.Message, state: FSMContext):
try:
if check_access(message.from_user.id):
await state.set_state("ADMIN")
logger.info(f"Запуск админ панели для пользователя: {message.from_user.id}")
markup = get_reply_keyboard_admin()
await message.answer("Добро пожаловать в админку. Выбери что хочешь:",
reply_markup=markup)
else:
await message.answer('Доступ запрещен, досвидания!')
except Exception as e:
logger.error(f"Ошибка при запуске админ панели: {e}")
await message.bot.send_message(IMPORTANT_LOGS,
f'Ошибка в функции admin_panel {e}. Traceback: {traceback.format_exc()}')
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
StateFilter("ADMIN"),
F.text == 'Бан (Список)'
)
async def get_last_users(message: types.Message):
logger.info(
f"Попытка получения списка последних пользователей. Текст сообщения: {message.text} Имя автора сообщения: {message.from_user.full_name})")
list_users = BotDB.get_last_users_from_db()
keyboard = create_keyboard_with_pagination(1, len(list_users), list_users, 'ban')
await message.answer(text="Список пользователей которые последними обращались к боту",
reply_markup=keyboard)
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
StateFilter("ADMIN"),
F.text == 'Разбан (список)'
)
async def get_banned_users(message):
logger.info(
f"Попытка получения списка заблокированных пользователей. Текст сообщения: {message.text} Имя автора сообщения: {message.from_user.full_name})")
message_text = get_banned_users_list(0)
buttons_list = get_banned_users_buttons()
if buttons_list:
k = create_keyboard_with_pagination(1, len(buttons_list), buttons_list, 'unlock')
await message.answer(text=message_text, reply_markup=k)
else:
await message.answer(text="В списке забанненых пользователей никого нет")
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
StateFilter("BAN_2")
)
async def ban_user_step_2(message: types.Message, state: FSMContext):
user_data = await state.get_data()
logger.info(f"Переход на шаг 2 бана пользователя. Словарь с данными для бана: {user_data})")
await state.update_data(message_for_user=message.text)
markup = create_keyboard_for_ban_days()
await message.answer(f"Выбрана причина: {message.text}. Выбери срок бана в днях или напиши "
f"его в чат", reply_markup=markup)
await state.set_state("BAN_3")
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
StateFilter("BAN_3")
)
async def ban_user_step_3(message: types.Message, state: FSMContext):
logger.info(f"ban_user_step_3. Расчет даты разбана. Входные данные {message.text}")
if message.text != 'Навсегда':
count_days = int(message.text)
date_to_unban = add_days_to_date(count_days)
else:
date_to_unban = None
logger.info(f"ban_user_step_3. Расчет даты разбана. date_to_unban: {date_to_unban}")
await state.update_data(date_to_unban=date_to_unban)
user_data = await state.get_data()
markup = create_keyboard_for_approve_ban()
await message.answer(
f"Необходимо подтверждение:\nПользователь:{user_data['user_id']}\nПричина бана:{user_data['message_for_user']}\nСрок бана:{user_data['date_to_unban']}",
reply_markup=markup)
await state.set_state("BAN_FINAL")
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
StateFilter("BAN_FINAL")
)
async def approve_ban(message: types.Message, state: FSMContext):
user_data = await state.get_data()
logger.info(f"Переход на финальный шаг бана пользователя. Словарь с данными для бана: {user_data})")
if message.text == 'Подтвердить':
exists = BotDB.check_user_in_blacklist(user_data['user_id'])
if exists:
await message.reply(f"Пользователь уже был заблокирован ранее.")
logger.info(f"Пользователь: {user_data['user_id']} был заблокирован ранее)")
await state.set_state('ADMIN')
else:
BotDB.set_user_blacklist(user_data['user_id'],
user_data['user_name'],
user_data['message_for_user'],
user_data['date_to_unban'])
await message.reply(f"Пользователь {user_data['user_name']} успешно заблокирован.")
logger.info(f"Пользователь: {user_data['user_id']} успешно заблокирован)")
await state.set_state('ADMIN')
markup = get_reply_keyboard_admin()
await message.answer('Вернулись в меню', reply_markup=markup)

View File

@@ -0,0 +1 @@
from .main import callback_router

View File

@@ -0,0 +1,165 @@
import traceback
from aiogram import Router, F, types
from aiogram.fsm.context import FSMContext
from aiogram.types import CallbackQuery
from database.db import BotDB
from helper_bot.keyboards.main import create_keyboard_with_pagination, get_reply_keyboard_admin, \
create_keyboard_for_ban_reason
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
from helper_bot.utils.helper_func import send_text_message, send_photo_message, get_banned_users_list, \
get_banned_users_buttons, delete_user_blacklist, get_help_message_id, send_media_group_message
from logs.custom_logger import Logger
callback_router = Router()
#Инициализируем логгер
callback_logger = Logger(name='callback_logger')
logger = callback_logger.get_logger()
bdf = BaseDependencyFactory()
GROUP_FOR_POST = bdf.settings['Telegram']['group_for_posts']
GROUP_FOR_MESSAGE = bdf.settings['Telegram']['group_for_message']
MAIN_PUBLIC = bdf.settings['Telegram']['main_public']
GROUP_FOR_LOGS = bdf.settings['Telegram']['group_for_logs']
IMPORTANT_LOGS = bdf.settings['Telegram']['important_logs']
PREVIEW_LINK = bdf.settings['Telegram']['preview_link']
LOGS = bdf.settings['Settings']['logs']
TEST = bdf.settings['Settings']['test']
BotDB = BotDB('database/tg-bot-database')
@callback_router.callback_query(
F.data == "publish"
)
async def post_for_group(call: CallbackQuery, state: FSMContext):
logger.info(
f'Получен callback-запрос с данными: {call.data} от пользователя {call.from_user.full_name} (ID: {call.from_user.id})')
if call.data == 'publish' and call.message.content_type == 'text' and call.message.text != "^":
try:
await send_text_message(MAIN_PUBLIC, call.message, call.message.text)
await call.bot.delete_message(chat_id=GROUP_FOR_POST, message_id=call.message.message_id)
logger.info(f'Текст сообщения опубликован в канале {MAIN_PUBLIC}.')
await call.answer(text='Выложено!', show_alert=True, cache_time=3)
except Exception as e:
await call.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
logger.error(f'Ошибка при публикации текста в канал {MAIN_PUBLIC}: {str(e)}')
await call.answer(text='Что-то пошло не так!', show_alert=True, cache_time=3)
elif call.data == 'publish' and call.message.content_type == 'photo':
try:
print(f'CALLMESSAGE - {call.message.text}')
await send_photo_message(MAIN_PUBLIC, call.message, call.message.photo[-1].file_id, call.message.caption)
await call.bot.delete_message(chat_id=GROUP_FOR_POST, message_id=call.message.message_id)
logger.info(f'Пост с фото опубликован в канале {MAIN_PUBLIC}.')
await call.answer(text='Выложено!', show_alert=True, cache_time=3)
except Exception as e:
await call.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
logger.error(f'Ошибка при публикации фотографии в канал {MAIN_PUBLIC}: {str(e)}')
await call.answer(text='Что-то пошло не так!', show_alert=True, cache_time=3)
elif call.data == 'publish' and call.message.text == "^":
user_data = await state.get_data()
media_group_message_id = get_help_message_id(call.message.message_id, user_data)
await call.bot.copy_message(chat_id=MAIN_PUBLIC, from_chat_id=GROUP_FOR_POST,message_id=media_group_message_id, reply_markup=None)
#await call.bot.copy_messages(chat_id=MAIN_PUBLIC, from_chat_id=GROUP_FOR_POST, message_ids=[media_group_message_id, media_group_message_id-1])
await call.bot.delete_message(chat_id=MAIN_PUBLIC, message_id=media_group_message_id)
print(user_data['media_group_message_id'])
print(user_data['help_message_id'])
await call.answer(text='Выложено!', show_alert=True, cache_time=3)
@callback_router.callback_query(
F.data == "decline"
)
async def decline_post_for_group(call: CallbackQuery, state: FSMContext):
logger.info(
f'Получен callback-запрос с данными: {call.data} от пользователя {call.from_user.full_name} (ID: {call.from_user.id})')
try:
if call.message.content_type == 'text' and call.message.text != "^":
await call.bot.delete_message(chat_id=GROUP_FOR_POST, message_id=call.message.message_id)
logger.info(
f'Сообщение отклонено админом {call.from_user.full_name} (ID: {call.from_user.id}).')
await call.answer(text='Отклонено!', show_alert=True, cache_time=3)
if call.message.text == '^':
user_data = await state.get_data()
media_group_message_id = get_help_message_id(call.message.message_id, user_data)
await call.bot.delete_message(chat_id=MAIN_PUBLIC, message_id=media_group_message_id)
except Exception as e:
await call.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
logger.error(f'Ошибка при удалении сообщения в группе {GROUP_FOR_POST}: {str(e)}')
await call.answer(text='Что-то пошло не так!', show_alert=True, cache_time=3)
@callback_router.callback_query(
F.data.contains('ban')
)
async def process_ban_user(call: CallbackQuery, state: FSMContext):
user_id = call.data[4:]
logger.info(
f"Вызов функции process_ban_user. Данные callback: {call.data} пользователь: {user_id}")
user_name = BotDB.get_username(user_id=user_id)
if user_name:
await state.update_data(user_id=user_id, user_name=user_name, message_for_user=None,
date_to_unban=None)
markup = create_keyboard_for_ban_reason()
await call.message.answer(
text=f"Выбран пользователь:\nid: {user_id}\nusername:{user_name}. Выбери причину бана из списка или напиши ее в чат",
reply_markup=markup)
await state.set_state('BAN_2')
else:
markup = get_reply_keyboard_admin()
await call.message.answer(text='Пользователь с таким ID не найден в базе', markup=markup)
await state.set_state('ADMIN')
@callback_router.callback_query(
F.data.contains('unlock')
)
async def process_unlock_user(call: CallbackQuery):
user_id = call.data[7:]
user_name = BotDB.get_username(user_id=user_id)
delete_user_blacklist(user_id)
logger.info(f"Разблокирован пользователь с ID: {user_id}\nusername:{user_name}")
username = BotDB.get_username(user_id)
await call.answer(f'Пользователь разблокирован {username}', show_alert=True)
@callback_router.callback_query(
F.data == 'return'
)
async def return_to_main_menu(call: CallbackQuery):
await call.message.delete()
logger.info(f"Запуск админ панели для пользователя: {call.message.from_user.id}")
markup = get_reply_keyboard_admin()
await call.message.answer("Добро пожаловать в админку. Выбери что хочешь:",
reply_markup=markup)
@callback_router.callback_query(
F.data.contains('page')
)
async def change_page(call: CallbackQuery):
page_number = int(call.data[5:])
logger.info(f"Переход на страницу {page_number}")
if call.message.text == 'Список пользователей которые последними обращались к боту':
list_users = BotDB.get_last_users_from_db()
#TODO: Здесь где-то надо добавить обработку ошибки IndexError: list index out of range
keyboard = create_keyboard_with_pagination(int(page_number), len(list_users), list_users,
'ban')
await call.bot.edit_message_reply_markup(chat_id=call.message.chat.id, message_id=call.message.message_id,
reply_markup=keyboard)
else:
#Готовим сообщения
message_user = get_banned_users_list(int(page_number) * 7 - 7)
await call.bot.edit_message_text(chat_id=call.message.chat.id, message_id=call.message.message_id,
text=message_user)
#Готовим клавиатуру
buttons = get_banned_users_buttons()
keyboard = create_keyboard_with_pagination(int(call.data[5:]), len(buttons), buttons, 'unlock')
await call.bot.edit_message_reply_markup(chat_id=call.message.chat.id, message_id=call.message.message_id,
reply_markup=keyboard)

View File

@@ -0,0 +1 @@
from .main import group_router

View File

@@ -0,0 +1,54 @@
from aiogram import Router, types
from aiogram.fsm.context import FSMContext
from database.db import BotDB
from helper_bot.filters.main import ChatTypeFilter
from helper_bot.keyboards.main import get_reply_keyboard_leave_chat
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
from helper_bot.utils.helper_func import send_text_message
from logs.custom_logger import Logger
group_router = Router()
#Инициализируем логгер
group_logger = Logger(name='group_logger')
logger = group_logger.get_logger()
bdf = BaseDependencyFactory()
GROUP_FOR_POST = bdf.settings['Telegram']['group_for_posts']
GROUP_FOR_MESSAGE = bdf.settings['Telegram']['group_for_message']
MAIN_PUBLIC = bdf.settings['Telegram']['main_public']
GROUP_FOR_LOGS = bdf.settings['Telegram']['group_for_logs']
IMPORTANT_LOGS = bdf.settings['Telegram']['important_logs']
PREVIEW_LINK = bdf.settings['Telegram']['preview_link']
LOGS = bdf.settings['Settings']['logs']
TEST = bdf.settings['Settings']['test']
BotDB = BotDB('database/tg-bot-database')
@group_router.message(
ChatTypeFilter(chat_type=["group", "supergroup"]),
)
async def handle_message(message: types.Message, state: FSMContext):
"""Функция ответа админа пользователю через закрытый чат"""
logger.info(
f'Получено сообщение в группе {message.chat.title} (ID: {message.chat.id}) от пользователя {message.from_user.full_name} (ID: {message.from_user.id}): "{message.text}"')
markup = get_reply_keyboard_leave_chat()
message_id = 0
try:
message_id = message.reply_to_message.message_id
except AttributeError as e:
await message.answer('Блять, выдели сообщение!')
logger.warning(
f'В группе {message.chat.title} (ID: {message.chat.id}) админ не выделил сообщение для ответа. Ошибка {str(e)}')
message_from_admin = message.text
try:
chat_id = BotDB.get_user_by_message_id(message_id)
await send_text_message(chat_id, message, message_from_admin, markup)
await state.set_state("CHAT")
logger.info(f'Ответ админа "{message.text}" отправлен пользователю с ID: {chat_id} на сообщение {message_id}')
except TypeError as e:
await message.answer('Не могу найти кому ответить в базе, проебали сообщение.')
logger.error(
f'Ошибка при поиске пользователя в базе для ответа на сообщение: {message.text} в группе {message.chat.title} (ID сообщения: {message.message_id}) Ошибка: {str(e)}')

View File

@@ -0,0 +1 @@
from .main import private_router

View File

@@ -0,0 +1,287 @@
import random
import traceback
from datetime import datetime
from pathlib import Path
from time import sleep
from aiogram import types, Router, F
from aiogram.filters import Command, StateFilter
from aiogram.fsm.context import FSMContext
from aiogram.types import FSInputFile
from helper_bot.filters.main import ChatTypeFilter
from helper_bot.keyboards import get_reply_keyboard, get_reply_keyboard_for_post
from helper_bot.keyboards.main import get_reply_keyboard_leave_chat
from helper_bot.middlewares.text_middleware import AlbumMiddleware
from helper_bot.utils import messages
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
from helper_bot.utils.helper_func import get_first_name, get_text_message, send_text_message, send_photo_message, \
process_photo_album, send_media_group_message, check_username_and_full_name
from logs.custom_logger import Logger
from database.db import BotDB
private_router = Router()
private_router.message.middleware(AlbumMiddleware())
#Инициализируем логгер
private_logger = Logger(name='private_handler')
logger = private_logger.get_logger()
bdf = BaseDependencyFactory()
GROUP_FOR_POST = bdf.settings['Telegram']['group_for_posts']
GROUP_FOR_MESSAGE = bdf.settings['Telegram']['group_for_message']
MAIN_PUBLIC = bdf.settings['Telegram']['main_public']
GROUP_FOR_LOGS = bdf.settings['Telegram']['group_for_logs']
IMPORTANT_LOGS = bdf.settings['Telegram']['important_logs']
PREVIEW_LINK = bdf.settings['Telegram']['preview_link']
LOGS = bdf.settings['Settings']['logs']
TEST = bdf.settings['Settings']['test']
BotDB = BotDB('database/tg-bot-database')
@private_router.message(
ChatTypeFilter(chat_type=["private"]),
Command("start")
)
@private_router.message(
ChatTypeFilter(chat_type=["private"]),
F.text == 'Вернуться в бота'
)
async def handle_start_message(message: types.Message, state: FSMContext):
try:
user_id = message.from_user.id
full_name = message.from_user.full_name
username = message.from_user.username
await message.forward(chat_id=GROUP_FOR_LOGS)
is_need_update = check_username_and_full_name(user_id, username, full_name)
if is_need_update:
BotDB.update_username_and_full_name(user_id, username, full_name)
await message.answer(f"Давно не виделись! Вижу что ты изменился;) Теперь буду звать тебя: {full_name} и ник @{username}")
await message.bot.send_message(chat_id=GROUP_FOR_LOGS, text=f'Для пользователя: {user_id} обновлены данные в БД.\nНовое имя: {full_name}\nНовый ник:{username}')
sleep(2)
await state.set_state("START")
logger.info(
f"Формирование приветственного сообщения для пользователя. Сообщение: {message.text} "
f"Имя автора сообщения: {message.from_user.full_name})")
name_stick_hello = list(Path('Stick').rglob('Hello_*'))
random_stick_hello = random.choice(name_stick_hello)
random_stick_hello = FSInputFile(path=random_stick_hello)
logger.info(f"Стикер успешно получен из БД")
await message.answer_sticker(random_stick_hello)
sleep(0.3)
except Exception as e:
logger.error(f"Произошла ошибка handle_start_message при получении стикеров. Ошибка:{str(e)}")
await message.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка при получении стикеров: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
try:
user_id = message.from_user.id
full_name = message.from_user.full_name
username = message.from_user.username
first_name = message.from_user.first_name
is_bot = message.from_user.is_bot
language_code = message.from_user.language_code
current_date = datetime.now()
date = current_date.strftime("%Y-%m-%d %H:%M:%S")
if not BotDB.user_exists(user_id):
BotDB.add_new_user_in_db(user_id, first_name, full_name, username, is_bot, language_code, date,
date)
BotDB.update_date_for_user(date, user_id)
markup = get_reply_keyboard(BotDB, message.from_user.id)
hello_message = messages.get_message(get_first_name(message), 'HELLO_MESSAGE')
await message.answer(hello_message, reply_markup=markup)
except Exception as e:
logger.error(
f"Произошла ошибка при отправке приветственного сообщения для пользователя {message.from_user.id} Имя: {message.from_user.full_name}. Ошибка: {str(e)}")
await message.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
@private_router.message(
StateFilter("START"),
ChatTypeFilter(chat_type=["private"]),
F.text == '📢Предложить свой пост'
)
async def suggest_post(message: types.Message, state: FSMContext):
try:
await message.forward(chat_id=GROUP_FOR_LOGS)
await state.set_state("SUGGEST")
current_state = await state.get_state()
logger.info(
f"Вызов функции suggest_post. Сообщение: {message.text} Имя автора сообщения: {message.from_user.full_name} Идентификатор сообщения: {message.message_id}. State - {current_state}")
markup = types.ReplyKeyboardRemove()
suggest_news = messages.get_message(get_first_name(message), 'SUGGEST_NEWS')
await message.answer(suggest_news)
sleep(0.3)
suggest_news_2 = messages.get_message(get_first_name(message), 'SUGGEST_NEWS_2')
await message.answer(suggest_news_2, reply_markup=markup)
except Exception as e:
await message.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
@private_router.message(
ChatTypeFilter(chat_type=["private"]),
F.text == '👋🏼Сказать пока!'
)
@private_router.message(
ChatTypeFilter(chat_type=["private"]),
F.text == 'Выйти из чата'
)
async def end_message(message: types.Message, state: FSMContext):
try:
logger.info(
f"Вызов функции end_message. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
name_stick_bye = list(Path('Stick').rglob('Universal_*'))
random_stick_bye = random.choice(name_stick_bye)
random_stick_bye = FSInputFile(path=random_stick_bye)
await message.answer_sticker(random_stick_bye)
except Exception as e:
logger.error(
f"Ошибка в функции end_message при получении стикера: {str(e)} Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
await message.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
try:
markup = types.ReplyKeyboardRemove()
bye_message = messages.get_message(get_first_name(message), 'BYE_MESSAGE')
await message.answer(bye_message, reply_markup=markup)
await state.set_state("START")
except Exception as e:
logger.error(
f"Ошибка в функции stickers при получении сообщения: {str(e)} Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
await message.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
@private_router.message(
StateFilter("SUGGEST"),
ChatTypeFilter(chat_type=["private"]),
)
async def suggest_router(message: types.Message, state: FSMContext, album: list = None):
logger.info(
f"Вызов функции suggest_router. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
try:
if message.content_type == 'text':
lower_text = message.text.lower()
post_text, is_anonymous = get_text_message(lower_text, message.from_user.full_name,
message.from_user.username)
markup = get_reply_keyboard_for_post()
if is_anonymous:
await send_text_message(GROUP_FOR_POST, message, post_text, markup)
else:
await send_text_message(GROUP_FOR_POST, message, post_text, markup)
markup_for_user = get_reply_keyboard(BotDB, message.from_user.id)
success_send_message = messages.get_message(get_first_name(message), 'SUCCESS_SEND_MESSAGE')
await message.answer(success_send_message, reply_markup=markup_for_user)
await state.set_state("START")
elif message.content_type == 'photo' and message.media_group_id is None:
lower_caption = message.caption.lower()
markup = get_reply_keyboard_for_post()
post_caption, is_anonymous = get_text_message(lower_caption, message.from_user.full_name,
message.from_user.username)
#TODO: тут какая-то шляпа
if is_anonymous:
await send_photo_message(GROUP_FOR_POST, message,
message.photo[-1].file_id, post_caption, markup)
else:
await send_photo_message(GROUP_FOR_POST, message,
message.photo[-1].file_id, post_caption, markup)
markup_for_user = get_reply_keyboard(BotDB, message.from_user.id)
success_send_message = messages.get_message(get_first_name(message), 'SUCCESS_SEND_MESSAGE')
await message.answer(success_send_message, reply_markup=markup_for_user)
await state.set_state("START")
elif message.media_group_id is not None:
post_caption = " "
if album[0].caption:
lower_caption = album[0].caption.lower()
post_caption, is_anonymous = get_text_message(lower_caption, message.from_user.full_name,
message.from_user.username)
media_group = process_photo_album(album, post_caption)
media_group_message_id = await send_media_group_message(GROUP_FOR_POST, message,
media_group)
sleep(0.2)
markup = get_reply_keyboard_for_post()
help_message_id = await send_text_message(GROUP_FOR_POST, message, "^", markup)
await state.update_data(media_group_message_id=media_group_message_id, help_message_id=help_message_id)
markup_for_user = get_reply_keyboard(BotDB, message.from_user.id)
success_send_message = messages.get_message(get_first_name(message), 'SUCCESS_SEND_MESSAGE')
await message.answer(success_send_message, reply_markup=markup_for_user)
await state.set_state("START")
else:
await message.bot.send_message(message.chat.id,
'Я пока не умею работать с таким сообщением. Пришли текст и фото/фоты(ы)')
except Exception as e:
await message.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
@private_router.message(
ChatTypeFilter(chat_type=["private"]),
F.text == '🤪Хочу стикеры'
)
async def stickers(message: types.Message, state: FSMContext):
logger.info(
f"Вызов функции stickers. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
markup = get_reply_keyboard(BotDB, message.from_user.id)
try:
BotDB.update_info_about_stickers(user_id=message.from_user.id)
await message.forward(chat_id=GROUP_FOR_LOGS)
await message.answer(text='Хорошо, лови, добавить можно отсюда: https://t.me/addstickers/love_biysk',
reply_markup=markup)
await state.set_state("START")
except Exception as e:
await message.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
logger.error(
f"Ошибка функции stickers. Ошибка: {str(e)} Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
@private_router.message(
StateFilter("START"),
ChatTypeFilter(chat_type=["private"]),
F.text == '📩Связаться с админами'
)
async def connect_with_admin(message: types.Message, state: FSMContext):
logger.info(
f"Вызов функции connect_with_admin. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}")
admin_message = messages.get_message(get_first_name(message), 'CONNECT_WITH_ADMIN')
await message.answer(admin_message, parse_mode="html")
await message.forward(chat_id=GROUP_FOR_LOGS)
await state.set_state("PRE_CHAT")
@private_router.message(
StateFilter("PRE_CHAT"),
ChatTypeFilter(chat_type=["private"]),
)
@private_router.message(
StateFilter("CHAT"),
ChatTypeFilter(chat_type=["private"]),
)
async def resend_message_in_group_for_message(message: types.Message, state: FSMContext):
logger.info(
f"Попытка пересылки сообщения в связь с админами. Сообщение: {message.text} Имя автора сообщения: {message.from_user.full_name} Идентификатор сообщения: {message.message_id})")
await message.forward(chat_id=GROUP_FOR_MESSAGE)
current_date = datetime.now()
date = current_date.strftime("%Y-%m-%d %H:%M:%S")
BotDB.add_new_message_in_db(message.text, message.from_user.id, message.message_id + 1, date)
question = messages.get_message(get_first_name(message), 'QUESTION')
user_state = await state.get_state()
if user_state == "PRE_CHAT":
markup = get_reply_keyboard(BotDB, message.from_user.id)
await message.answer(question, reply_markup=markup)
await state.set_state("START")
elif user_state == "CHAT":
markup = get_reply_keyboard_leave_chat()
await message.answer(question, reply_markup=markup)
# @private_router.message(
# ChatTypeFilter(chat_type=["private"])
# )
# async def default(message: types.Message, state: FSMContext):
# markup = get_reply_keyboard(BotDB, message.from_user.id)
# await message.answer('Кажется ты заблудился. Держи клавиатуру, твое состояние сброшено на начало', reply_markup=markup)
# await state.set_state("START")

View File

@@ -0,0 +1 @@
from .main import get_reply_keyboard_for_post, get_reply_keyboard

Binary file not shown.

View File

@@ -0,0 +1,110 @@
from aiogram import types
from aiogram.utils.keyboard import ReplyKeyboardBuilder, InlineKeyboardBuilder
def get_reply_keyboard_for_post():
builder = InlineKeyboardBuilder()
builder.row(types.InlineKeyboardButton(
text="Опубликовать", callback_data="publish"),
types.InlineKeyboardButton(
text="Отклонить", callback_data="decline")
)
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup
def get_reply_keyboard(BotDB, user_id):
builder = ReplyKeyboardBuilder()
builder.add(types.KeyboardButton(text="📢Предложить свой пост"))
builder.add(types.KeyboardButton(text="📩Связаться с админами"))
builder.add(types.KeyboardButton(text="👋🏼Сказать пока!"))
if not BotDB.get_info_about_stickers(user_id=user_id):
builder.add(types.KeyboardButton(text="🤪Хочу стикеры"))
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup
def get_reply_keyboard_leave_chat():
builder = ReplyKeyboardBuilder()
builder.add(types.KeyboardButton(text="Выйти из чата"))
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup
def get_reply_keyboard_admin():
builder = ReplyKeyboardBuilder()
builder.add(types.KeyboardButton(text="Бан (Список)"))
builder.add(types.KeyboardButton(text="Разбан (список)"))
builder.add(types.KeyboardButton(text="Вернуться в бота"))
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup
def create_keyboard_with_pagination(page: int, total_items: int, array_items: list[tuple[any, any]], callback: str):
"""
Создает клавиатуру с пагинацией для заданного набора элементов и устанавливает необходимый callback
Args:
page: Номер текущей страницы.
total_items: Общее количество элементов.
array_items: Лист кортежей. Содержит в себе user_name: user_id
callback: Действие в коллбеке. Вернет callback вида ({callback}_{user_id})
Returns:
InlineKeyboardMarkup: Клавиатура с кнопками пагинации.
"""
# Определяем общее количество страниц
total_pages = (total_items + 9 - 1) // 9
# Создаем билдер для клавиатуры
keyboard = InlineKeyboardBuilder()
# TODO: Тут поправить на 9 объектов, а не 7. Клавиатуру переделал
# Вычисляем стартовый номер для текущей страницы
start_index = (page - 1) * 9
# Кнопки с номерами страниц
for i in range(start_index, min(start_index + 9, len(array_items))):
keyboard.add(types.InlineKeyboardButton(
text=f"{array_items[i][0]}", callback_data=f"{callback}_{array_items[i][1]}"
))
keyboard.adjust(3)
next_button = types.InlineKeyboardButton(
text="➡️ Следующая", callback_data=f"page_{page + 1}"
)
prev_button = types.InlineKeyboardButton(
text="⬅️ Предыдущая", callback_data=f"page_{page - 1}"
)
keyboard.row(prev_button, next_button)
home_button = types.InlineKeyboardButton(
text="🏠 Назад", callback_data="return")
keyboard.row(home_button)
k = keyboard.as_markup()
return k
def create_keyboard_for_ban_reason():
builder = ReplyKeyboardBuilder()
builder.add(types.KeyboardButton(text="Спам"))
builder.add(types.KeyboardButton(text="Заебал стикерами"))
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup
def create_keyboard_for_ban_days():
builder = ReplyKeyboardBuilder()
builder.add(types.KeyboardButton(text="1"))
builder.add(types.KeyboardButton(text="7"))
builder.add(types.KeyboardButton(text="30"))
builder.add(types.KeyboardButton(text="Навсегда"))
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup
def create_keyboard_for_approve_ban():
builder = ReplyKeyboardBuilder()
builder.add(types.KeyboardButton(text="Подтвердить"))
builder.add(types.KeyboardButton(text="Отменить"))
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
return markup

21
helper_bot/main.py Normal file
View File

@@ -0,0 +1,21 @@
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.fsm.strategy import FSMStrategy
from helper_bot.handlers.admin import admin_router
from helper_bot.handlers.callback import callback_router
from helper_bot.handlers.group import group_router
from helper_bot.handlers.private import private_router
async def start_bot(bdf):
token = bdf.settings['Telegram']['bot_token']
bot = Bot(token=token, default=DefaultBotProperties(
parse_mode='HTML',
link_preview_is_disabled=bdf.settings['Telegram']['preview_link']
))
dp = Dispatcher(storage=MemoryStorage(), fsm_strategy=FSMStrategy.GLOBAL_USER)
dp.include_routers(private_router, callback_router, group_router, admin_router)
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot, skip_updates=True)

View File

View File

@@ -0,0 +1,46 @@
import asyncio
from collections import defaultdict
from typing import Any, Dict, Union
from aiogram import BaseMiddleware
from aiogram.types import Message
class BulkTextMiddleware(BaseMiddleware):
def __init__(self, latency: Union[int, float] = 0.1):
# Initialize latency and album_data dictionary
self.latency = latency
self.texts = defaultdict(list)
#
async def __call__(self, handler, event: Message, data: Dict[str, Any]) -> Any:
"""
Main middleware logic.
"""
# # If the event has no media_group_id, pass it to the handler immediately
key = (event.chat.id, event.from_user.id)
if not event.text:
return await handler(event, data)
self.texts[key].append(event)
total_before = len(self.texts[key])
# # Wait for a specified latency period
await asyncio.sleep(self.latency)
#
# # Check the total number of messages after the latency
total_after = len(self.texts[key])
#
# # If new messages were added during the latency, exit
if total_before != total_after:
return
#
# # Sort the album messages by message_id and add to data
msg_texts = self.texts[key]
msg_texts.sort(key=lambda x: x.message_id)
data["texts"] = ''.join([msg.text for msg in msg_texts])
#
# Remove the media group from tracking to free up memory
del self.texts[key]
# # Call the original event handler
return await handler(event, data)
#

View File

@@ -0,0 +1,61 @@
import asyncio
from typing import Any, Dict, Union
from aiogram import BaseMiddleware
from aiogram.types import Message
class AlbumMiddleware(BaseMiddleware):
def __init__(self, latency: Union[int, float] = 0.1):
# Initialize latency and album_data dictionary
self.latency = latency
self.album_data = {}
#
def collect_album_messages(self, event: Message):
"""
Collect messages of the same media group.
"""
# # Check if media_group_id exists in album_data
if event.media_group_id not in self.album_data:
# # Create a new entry for the media group
self.album_data[event.media_group_id] = {"messages": []}
#
# # Append the new message to the media group
self.album_data[event.media_group_id]["messages"].append(event)
#
# # Return the total number of messages in the current media group
return len(self.album_data[event.media_group_id]["messages"])
#
async def __call__(self, handler, event: Message, data: Dict[str, Any]) -> Any:
"""
Main middleware logic.
"""
# # If the event has no media_group_id, pass it to the handler immediately
if not event.media_group_id:
return await handler(event, data)
#
# # Collect messages of the same media group
total_before = self.collect_album_messages(event)
#
# # Wait for a specified latency period
await asyncio.sleep(self.latency)
#
# # Check the total number of messages after the latency
total_after = len(self.album_data[event.media_group_id]["messages"])
#
# # If new messages were added during the latency, exit
if total_before != total_after:
return
#
# # Sort the album messages by message_id and add to data
album_messages = self.album_data[event.media_group_id]["messages"]
album_messages.sort(key=lambda x: x.message_id)
data["album"] = album_messages
#
# # Remove the media group from tracking to free up memory
del self.album_data[event.media_group_id]
# # Call the original event handler
return await handler(event, data)
#

View File

@@ -0,0 +1 @@
from .state import StateUser

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -0,0 +1,25 @@
import configparser
import os
import sys
class BaseDependencyFactory:
def __init__(self):
# Загрузка настроек из settings.ini
config_path = os.path.join(sys.path[0], 'settings.ini')
self.config = configparser.ConfigParser()
self.config.read(config_path)
self.settings = {}
for section in self.config.sections():
self.settings[section] = {}
for key in self.config[section]:
# Преобразование значений в соответствующий тип
if key == 'PREVIEW_LINK':
self.settings[section][key] = self.config.getboolean(section, key)
elif key == 'LOGS' or key == 'TEST':
self.settings[section][key] = self.config.getboolean(section, key)
else:
self.settings[section][key] = self.config.get(section, key)
def get_settings(self):
return self.settings

View File

@@ -0,0 +1,195 @@
from datetime import datetime, timedelta
from aiogram import types
from aiogram.types import InputMediaPhoto
from database.db import BotDB
BotDB = BotDB('database/tg-bot-database')
def get_first_name(message: types.Message) -> str:
return message.from_user.first_name
def get_text_message(post_text: str, first_name: str, username: str):
"""
Форматирует текст сообщения для публикации в зависимости от наличия ключевых слов "анон" и "неанон".
Args:
post_text: Текст сообщения
first_name: Имя автора поста
username: Юзернейм автора поста
Returns:
Кортеж из двух элементов:
- Сформированный текст сообщения.
- Флаг, указывающий, является ли пост анонимным (True - анонимный, False - не анонимный).
"""
if "неанон" in post_text or "не анон" in post_text:
is_anonymous = False
return f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {first_name} @{username}', is_anonymous
elif "анон" in post_text:
is_anonymous = True
return f'Пост из ТГ:\n{post_text}\n\nПост опубликован анонимно', is_anonymous
else:
is_anonymous = False
return f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {first_name} @{username}', is_anonymous
def process_photo_album(album, post_caption: str = ''):
"""
Создает список InputMediaPhoto для альбома.
Args:
album: Album объект из Telegram API.
post_caption: Текст подписи к первому фото.
Returns:
Список InputMediaPhoto.
"""
photo_media = []
for i, message in enumerate(album):
if i == 0:
photo_media.append(InputMediaPhoto(media=message.photo[-1].file_id, caption=post_caption))
else:
photo_media.append(InputMediaPhoto(media=message.photo[-1].file_id))
return photo_media
async def send_media_group_message(chat_id: int, message: types.Message, media_group: list[InputMediaPhoto]):
sent_message = await message.bot.send_media_group(
chat_id=chat_id,
media=media_group,
)
message_id = sent_message[-1].message_id
return message_id
async def send_text_message(chat_id, message: types.Message, post_text: str, markup: types.ReplyKeyboardMarkup = None):
if markup is None:
sent_message = await message.bot.send_message(
chat_id=chat_id,
text=post_text
)
message_id = sent_message.message_id
return message_id
else:
sent_message = await message.bot.send_message(
chat_id=chat_id,
text=post_text,
reply_markup=markup
)
message_id = sent_message.message_id
return message_id
async def send_photo_message(chat_id, message: types.Message, photo: str, post_text: str, markup: types.ReplyKeyboardMarkup = None):
if markup is None:
await message.bot.send_photo(
chat_id=chat_id,
caption=post_text,
photo=photo
)
else:
await message.bot.send_photo(
chat_id=chat_id,
caption=post_text,
photo=photo,
reply_markup=markup
)
def check_access(user_id: int):
"""Проверка прав на совершение действий"""
return BotDB.is_admin(user_id)
def add_days_to_date(days: int):
"""Прибавляет указанное количество дней к текущей дате и возвращает дату в формате DD-MM-YYYY."""
current_date = datetime.now()
future_date = current_date + timedelta(days=days)
formatted_date = future_date.strftime("%d-%m-%Y")
return formatted_date
def get_banned_users_list(offset: int):
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
Args:
offset: отступ для запроса в базу данных
Returns:
message - текст сообщения
user_ids - лист кортежей [(user_name: user_id)]
"""
users = BotDB.get_banned_users_from_db_with_limits(limit=7, offset=offset)
message = "Список заблокированных пользователей:\n"
for user in users:
message += f"Пользователь: {user[0]}\n"
message += f"Причина бана: {user[2]}\n"
message += f"Дата разбана: {user[3]}\n\n"
return message
def get_banned_users_buttons():
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
Args:
offset: отступ для запроса в базу данных
Returns:
message - текст сообщения
user_ids - лист кортежей [(user_name: user_id)]
"""
users = BotDB.get_banned_users_from_db()
user_ids = []
for user in users:
user_ids.append((user[0], user[1]))
return user_ids
def get_help_message_id(media_group_message_id: int, data: dict) -> int:
"""
Получает идентификатор сообщения помощи по идентификатору сообщения группы.
Args:
media_group_message_id: Идентификатор сообщения группы
data: Словарь с данными.
Returns:
Идентификатор сообщения помощи.
"""
if 'help_message_id' in data and 'media_group_message_id' in data:
return data['media_group_message_id']
else:
return 0
def delete_user_blacklist(user_id: int):
return BotDB.delete_user_blacklist(user_id=user_id)
def check_username_and_full_name(user_id: int, username: str, full_name: str):
username_db, full_name_db = BotDB.get_username_and_full_name(user_id=user_id)
return not username == username_db and full_name == full_name_db
def unban_notifier(self):
# Получение сегодняшней даты в формате DD-MM-YYYY
current_date = datetime.now()
print('Мы в функции unban_notifier')
today = current_date.strftime("%d-%m-%Y")
# Получение списка разблокированных пользователей
unblocked_users = self.BotDB.get_users_for_unblock_today(today)
message = "Разблокированные пользователи:\n"
for user_id, user_name in unblocked_users.items():
message += f"ID: {user_id}, Имя: {user_name}\n"
# Отправка сообщения в канал
self.bot.send_message(self.GROUP_FOR_MESSAGE, message)

13
helper_bot/utils/state.py Normal file
View File

@@ -0,0 +1,13 @@
from aiogram.fsm.state import StatesGroup, State
class StateUser(StatesGroup):
START = State()
SUGGEST = State()
ADMIN = State()
CHAT = State()
PRE_CHAT = State()
BAN_2 = State()
BAN_3 = State()
BAN_4 = State()
BAN_FINAL = State()

View File

@@ -1,22 +1,21 @@
import datetime
import os
from loguru import logger
import loguru
class Logger:
def __init__(self, name):
self.logger = logger.bind(name=name)
self.logger = loguru.logger.bind(name=name)
# Получение сегодняшней даты для имени файла
today = datetime.date.today().strftime('%Y-%m-%d')
# Создание папки для логов
current_dir = os.path.dirname(os.path.abspath(__file__))
logs_dir = os.path.join(current_dir, 'logs')
if not os.path.exists(logs_dir):
if not os.path.exists(current_dir):
# Если не существует, создаем ее
os.makedirs(logs_dir)
filename = f'{logs_dir}/helper_bot_{today}.log'
os.makedirs(current_dir)
filename = f'{current_dir}/helper_bot_{today}.log'
# Настройка формата логов
self.logger.add(
@@ -27,6 +26,9 @@ class Logger:
format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {name} | {line} | {message}",
)
def get_logger(self):
return self.logger
def info(self, message):
self.logger.info(message)

641
main.py
View File

@@ -1,641 +0,0 @@
import configparser
import os
import sys
from pathlib import Path
from time import sleep
from enum import Enum
from typing import Any
from apscheduler.schedulers.background import BackgroundScheduler
from db import BotDB
import telebot
import random
from datetime import datetime, timedelta
from telebot import types
from telebot.apihelper import ApiTelegramException
import messages
import traceback
#TODO: Добавить проверку можно ли отвечать пользователю? Сейчас если у него скрыто лс, ему похоже не приходят сообщения
#TODO Подумать над реализацией функционала с поступлениями в колледжи
#TODO: Покрыть все логированием и ошибками корректными. Ерроры кидать в чат.
#TODO: Покрыть все тестами
# Настройки
config_path = os.path.join(sys.path[0], 'settings.ini')
config = configparser.ConfigParser()
config.read(config_path)
# TELEGRAM
BOT_TOKEN = config.get('Telegram', 'BOT_TOKEN')
GROUP_FOR_POST = config.get('Telegram', 'group_for_posts')
GROUP_FOR_MESSAGE = config.get('Telegram', 'group_for_message')
MAIN_PUBLIC = config.get('Telegram', 'main_public')
GROUP_FOR_LOGS = config.get('Telegram', 'group_for_logs')
IMPORTANT_LOGS = config.get('Telegram', 'important_logs')
PREVIEW_LINK = config.getboolean('Telegram', 'PREVIEW_LINK')
# SETTINGS
LOGS = config.getboolean('Settings', 'logs')
TEST = config.getboolean('Settings', 'test')
# Инициализируем бота и базку
BotDB = BotDB()
class State(Enum):
START = "START"
SUGGEST = "SUGGEST"
ADMIN = "ADMIN"
CHAT = "CHAT"
PRE_CHAT = "PRE_CHAT"
class TelegramHelperBot:
def __init__(self, token):
self.bot = telebot.TeleBot(token)
self.state = State.START
# Router for user
@self.bot.message_handler(func=lambda message: True, chat_types=['private'])
def handle_message(message):
if BotDB.check_user_in_blacklist(message.from_user.id):
attribute = BotDB.get_blacklist_users_by_id(message.from_user.id)
self.bot.send_message(message.chat.id,
f'<b>Ты заблокирован\nПричина блокировки:</b> {attribute[2]}\n<b>Дата разблокировки:</b> {attribute[3]}', parse_mode='HTML')
return
if self.state == State.START:
if message.text == '/start':
self.start_message(message)
elif message.text == '📢Предложить свой пост':
self.suggest_post(message)
self.state = State.SUGGEST
elif message.text == '🤪Хочу стикеры':
self.stickers(message)
self.state = State.START
elif message.text == '📩Связаться с админами':
self.connect_with_admin(message)
self.state = State.PRE_CHAT
elif message.text == '👋🏼Сказать пока!':
self.end_message(message)
self.state = State.START
elif message.text == 'Выйти из чата':
self.end_message(message)
elif message.text == '/admin':
access = self.check_access(message.from_user.id)
if access:
self.admin_panel(message)
self.state = State.ADMIN
else:
self.bot.send_message(message.chat.id, 'Доступ запрещен, досвидания!')
elif message.text == '/state':
self.bot.send_message(message.chat.id,
f'Твой state == {self.state.value}')
else:
self.bot.send_message(message.chat.id,
#TODO: Здесь раньше был /state
"Не понимаю где ты находишься. Нажми /start, и я перезапущусь")
if self.state == State.SUGGEST:
self.bot.register_next_step_handler(message, self.send_to_suggest)
self.state = State.START
if message.text == '/start':
self.state = State.START
self.start_message(message)
if self.state == State.PRE_CHAT:
self.bot.register_next_step_handler(message, self.resend_message_in_group_for_message)
self.state = State.START
if message.text == '/start':
self.state = State.START
self.start_message(message)
if self.state == State.CHAT:
if message.text == 'Выйти из чата':
self.state = State.START
self.end_message(message)
elif message.text == '/start':
self.state = State.START
self.start_message(message)
else:
self.resend_message_in_group_for_message(message)
if self.state == State.ADMIN:
if message == '/admin' or message == '/restart' or message == 'Вернуться в админку':
access = self.check_access(message.from_user.id)
if access:
self.admin_panel(message)
else:
self.bot.send_message(message.chat.id, 'Доступ запрещен, досвидания!')
if message.text == '/start':
self.state = State.START
self.start_message(message)
@self.bot.message_handler(func=lambda message: True, chat_types=['group'])
def handle_message(message):
"""Функция ответа админа пользователю через закрытый чат"""
self.state = State.CHAT
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("Выйти из чата")
markup.add(item1)
message_id = message.reply_to_message.id
message_from_admin = message.text
chat_id = BotDB.get_user_by_message_id(message_id)
self.bot.send_message(chat_id, message_from_admin, reply_markup=markup)
# Админка
@self.bot.callback_query_handler(func=lambda call: call.data in ['publish', 'decline'])
def post_for_group(call):
if call.data == 'publish' and call.message.content_type == 'text':
try:
self.bot.send_message(chat_id=MAIN_PUBLIC, text=call.message.text)
self.bot.delete_message(chat_id=GROUP_FOR_POST, message_id=call.message.message_id)
except Exception as e:
if LOGS:
self.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
elif call.data == 'publish' and call.message.content_type == 'photo':
try:
self.bot.send_photo(
chat_id=MAIN_PUBLIC,
caption=call.message.caption,
photo=call.message.photo[-1].file_id,
)
self.bot.delete_message(chat_id=GROUP_FOR_POST, message_id=call.message.message_id)
except Exception as e:
if LOGS:
self.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
elif call.data == 'decline':
try:
self.bot.delete_message(chat_id=GROUP_FOR_POST, message_id=call.message.message_id)
except Exception as e:
if LOGS:
self.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
@self.bot.callback_query_handler(func=lambda call: True)
def pagination(call):
if call.data[:3] == 'ban':
user_id = call.data[4:]
self.ban_user(call.message, user_id)
if call.data == 'return':
self.bot.delete_message(call.message.chat.id, call.message.message_id)
self.admin_panel(call.message)
if call.data[:5] == 'unban':
self.delete_user_blacklist(call.data[6:])
msg = f'Успешно удалено.'
self.bot.send_message(chat_id=call.message.chat.id, text=msg)
elif call.data[:4] == 'page':
if call.message.text == 'Список пользователей которые последними обращались к боту':
list_users = BotDB.get_last_users_from_db()
keyboard = self.create_keyboard_with_pagination(int(call.data[5:]), len(list_users), list_users,
'ban')
self.bot.edit_message_reply_markup(call.message.chat.id, call.message.message_id,
reply_markup=keyboard)
if "Список заблокированных пользователей".lower() in call.message.text.lower():
#Готовим сообщения
message_user = self.get_banned_users_list(int(call.data[5:]) * 7 - 7)
self.bot.edit_message_text(chat_id=call.message.chat.id, message_id=call.message.message_id,
text=message_user)
#Готовим клавиатуру
buttons = self.get_banned_users_buttons()
keyboard = self.create_keyboard_with_pagination(int(call.data[5:]), len(buttons), buttons, 'unban')
self.bot.edit_message_reply_markup(call.message.chat.id, call.message.message_id,
reply_markup=keyboard)
def start(self):
while True:
try:
self.bot.polling(none_stop=True)
except (ConnectionError, Exception):
print(f"Произошла ошибка: {str(Exception)}\n\nTraceback:\n{traceback.format_exc()}")
def unban_notifier(self):
# Получение сегодняшней даты в формате DD-MM-YYYY
current_date = datetime.now()
today = current_date.strftime("%d-%m-%Y")
# Получение списка разблокированных пользователей
unblocked_users = BotDB.get_users_for_unblock_today(today)
message = "Разблокированные пользователи:\n"
for user_id, user_name in unblocked_users.items():
message += f"ID: {user_id}, Имя: {user_name}\n"
# Отправка сообщения в канал
self.bot.send_message(GROUP_FOR_MESSAGE, message)
# Черный список
def admin_panel(self, message):
try:
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("Бан (Список)")
#item2 = types.KeyboardButton("Добавить админа") #TODO: Когда-нибудь потом доделаю
#item3 = types.KeyboardButton("Удалить админа")
item4 = types.KeyboardButton("Разбан (список)")
item5 = types.KeyboardButton("Вернуться в бота")
markup.add(item1, item4, item5)
self.bot.send_message(message.chat.id, "Добро пожаловать в админку. Выбери что хочешь:",
reply_markup=markup)
self.bot.register_next_step_handler(message, self.handle_admin_message)
except Exception as e:
self.bot.register_next_step_handler(message, self.admin_panel)
def handle_admin_message(self, message):
try:
if message.text == "Бан (Список)":
self.get_last_users(message)
elif message.text == "Разбан (список)":
self.get_banned_users(message)
elif message.text == "Вернуться в бота":
self.start_message(message)
except Exception as e:
self.bot.reply_to(message, f"Ошибка\n\n {e}")
self.admin_panel(message)
def ban_user(self, message, user_id: int):
user_name = BotDB.get_username(user_id=user_id)
ban_object = {'user_id': user_id, 'user_name': user_name, 'message_for_user': None, 'date_to_unban': None}
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("Спам")
item2 = types.KeyboardButton("Заебал стикерами")
markup.add(item1, item2)
self.bot.send_message(message.chat.id,
f"Выбран пользователь: {user_id}. Выбери причину бана из списка или напиши ее в чат",
reply_markup=markup)
self.bot.register_next_step_handler(message, self.ban_user_step_2, ban_object)
def ban_user_step_2(self, message, ban_object: dict):
ban_object['message_for_user'] = message.text
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("1")
item2 = types.KeyboardButton("7")
item3 = types.KeyboardButton("30")
item4 = types.KeyboardButton("Навсегда")
markup.add(item1, item2, item3, item4)
self.bot.send_message(message.chat.id, f"Выбрана причина: {message.text}. Выбери срок бана в днях или напиши "
f"его в чат",
reply_markup=markup)
self.bot.register_next_step_handler(message, self.ban_user_step_3, ban_object)
def ban_user_step_3(self, message, ban_object: dict):
date_to_unban = None
if message.text != 'Навсегда':
date_to_unban = self.add_days_to_date(message.text)
else:
pass
ban_object['date_to_unban'] = date_to_unban
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("Подтвердить")
item2 = types.KeyboardButton("Отменить")
markup.add(item1, item2)
self.bot.send_message(message.chat.id,
f"Необходимо подтверждение:\nПользователь:{ban_object['user_id']}\nПричина бана:{ban_object['message_for_user']}.\nСрок бана:{ban_object['date_to_unban']}",
parse_mode='html',
reply_markup=markup)
self.bot.register_next_step_handler(message, self.ban_user_final_step, ban_object)
def ban_user_final_step(self, message, ban_object: dict):
if message.text == 'Подтвердить':
exists = BotDB.check_user_in_blacklist(ban_object['user_id'])
if exists:
self.bot.reply_to(message, f"Пользователь уже был заблокирован ранее.")
self.admin_panel(message)
else:
BotDB.set_user_blacklist(ban_object['user_id'],
ban_object['user_name'],
ban_object['message_for_user'],
ban_object['date_to_unban'])
self.bot.reply_to(message, f"Пользователь {ban_object['user_name']} успешно заблокирован.")
self.admin_panel(message)
def get_last_users(self, message):
list_users = BotDB.get_last_users_from_db()
keyboard = self.create_keyboard_with_pagination(1, len(list_users), list_users, 'ban')
self.bot.send_message(chat_id=message.chat.id, text="Список пользователей которые последними обращались к боту",
reply_markup=keyboard)
def get_banned_users(self, message):
message_text = self.get_banned_users_list(0)
buttons_list = self.get_banned_users_buttons()
if buttons_list:
k = self.create_keyboard_with_pagination(1, len(buttons_list), buttons_list, 'unban')
self.bot.send_message(message.chat.id, message_text, reply_markup=k)
else:
self.bot.send_message(message.chat.id, "В списке забанненых пользователей никого нет")
self.admin_panel(message)
def start_message(self, message):
try:
name_stick_hello = list(Path('Stick').rglob('Hello_*'))
random_stick_hello = open(random.choice(name_stick_hello), 'rb')
# logging
if LOGS:
self.bot.forward_message(chat_id=GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
self.bot.send_sticker(message.chat.id, random_stick_hello)
sleep(0.3)
except Exception as e:
print(f'{str(e)}')
if LOGS:
self.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
try:
user_id = message.from_user.id
first_name = message.from_user.first_name
full_name = message.from_user.full_name
is_bot = message.from_user.is_bot
username = message.from_user.username
language_code = message.from_user.language_code
current_date = datetime.now()
date = current_date.strftime("%Y-%m-%d %H:%M:%S")
if not BotDB.user_exists(user_id):
BotDB.add_new_user_in_db(user_id, first_name, full_name, username, is_bot, language_code, date,
date)
BotDB.update_date_for_user(date, user_id)
markup = self.get_reply_keyboard(message)
hello_message = messages.get_message(self.__get_first_name(message), 'HELLO_MESSAGE')
self.bot.send_message(message.chat.id, hello_message, parse_mode='html', reply_markup=markup,
disable_web_page_preview=not PREVIEW_LINK)
except Exception as e:
print(f'{str(e)}')
if LOGS:
self.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
def resend_message_in_group_for_message(self, message):
self.bot.forward_message(chat_id=GROUP_FOR_MESSAGE,
from_chat_id=message.chat.id,
message_id=message.message_id
)
current_date = datetime.now()
date = current_date.strftime("%Y-%m-%d %H:%M:%S")
BotDB.add_new_message_in_db(message.text, message.message_id + 1, message.from_user.id, date)
question = messages.get_message(self.__get_first_name(message), 'QUESTION')
markup = self.get_reply_keyboard(message)
self.bot.send_message(message.chat.id, question, parse_mode='html', disable_web_page_preview=not PREVIEW_LINK,
reply_markup=markup)
def suggest_post(self, message):
try:
markup = types.ReplyKeyboardRemove()
suggest_news = messages.get_message(self.__get_first_name(message), 'SUGGEST_NEWS')
self.bot.send_message(message.chat.id, suggest_news, parse_mode='html')
sleep(0.3)
suggest_news_2 = messages.get_message(self.__get_first_name(message), 'SUGGEST_NEWS_2')
self.bot.send_message(message.chat.id, suggest_news_2, parse_mode='html', reply_markup=markup)
except Exception as e:
self.bot.send_message(IMPORTANT_LOGS,
f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
# logging
if LOGS:
self.bot.forward_message(chat_id=GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
def stickers(self, message):
BotDB.update_info_about_stickers(user_id=message.from_user.id)
markup = self.get_reply_keyboard(message)
try:
self.bot.forward_message(chat_id=GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
self.bot.send_message(message.chat.id,
text='Хорошо, лови, добавить можно отсюда: https://t.me/addstickers/love_biysk',
reply_markup=markup)
except ApiTelegramException as e:
self.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
def connect_with_admin(self, message):
connect_with_admin = messages.get_message(self.__get_first_name(message), 'CONNECT_WITH_ADMIN')
self.bot.send_message(message.chat.id, connect_with_admin, parse_mode="html")
# logging
if LOGS:
self.bot.forward_message(chat_id=GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
def end_message(self, message):
try:
name_stick_bye = list(Path('Stick').rglob('Universal_*'))
random_stick_bye = open(random.choice(name_stick_bye), 'rb')
self.bot.send_sticker(message.chat.id, random_stick_bye)
except ApiTelegramException as e:
if LOGS:
self.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
markup = types.ReplyKeyboardRemove()
try:
bye_message = messages.get_message(self.__get_first_name(message), 'BYE_MESSAGE')
self.bot.send_message(message.chat.id, bye_message,
parse_mode='html', reply_markup=markup, disable_web_page_preview=not PREVIEW_LINK)
except Exception as e:
if LOGS:
self.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
if LOGS:
# logging
self.bot.forward_message(chat_id=GROUP_FOR_LOGS,
from_chat_id=message.chat.id,
message_id=message.message_id)
def send_to_suggest(self, message):
markup = types.InlineKeyboardMarkup(row_width=1)
item1 = types.InlineKeyboardButton("Опубликовать", callback_data='publish')
item2 = types.InlineKeyboardButton("Отклонить", callback_data='decline')
markup.add(item1, item2)
try:
if message.content_type == 'text':
post_text = message.text.lower()
if post_text.find('неанон') != -1 or post_text.find('не анон') != -1:
self.bot.send_message(
# TODO: GROUP_FOR_POST
chat_id=GROUP_FOR_POST,
text=f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {message.from_user.first_name} @{message.from_user.username}',
reply_markup=markup
)
elif post_text.find('анон') != -1:
self.bot.send_message(
# TODO: GROUP_FOR_POST
chat_id=GROUP_FOR_POST,
text=f'Пост из ТГ:\n{message.text}\n\nПост опубликован анонимно',
reply_markup=markup
)
else:
self.bot.send_message(
# TODO: GROUP_FOR_POST
chat_id=GROUP_FOR_POST,
text=f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {message.from_user.first_name} @{message.from_user.username}',
reply_markup=markup
)
elif message.content_type == 'photo' and message.media_group_id is None:
post_text_for_photo = message.caption.lower()
if post_text_for_photo.find('неанон') != -1 or post_text_for_photo.find('не анон') != -1:
self.bot.send_photo(
# TODO: GROUP_FOR_POST
chat_id=GROUP_FOR_POST,
caption=f'Пост из ТГ:\n{post_text_for_photo}\n\nАвтор поста: {message.from_user.first_name} @{message.from_user.username}',
photo=message.photo[-1].file_id,
reply_markup=markup
)
elif post_text_for_photo.find('анон') != -1 or post_text_for_photo.find('анон') != -1:
self.bot.send_photo(
# TODO: GROUP_FOR_POST
chat_id=GROUP_FOR_POST,
caption=f'Пост из ТГ:\n{post_text_for_photo}\n\nПост опубликован анонимно',
photo=message.photo[-1].file_id,
reply_markup=markup
)
else:
self.bot.send_photo(
# TODO: GROUP_FOR_POST
chat_id=GROUP_FOR_POST,
caption=f'Пост из ТГ:\n{post_text_for_photo}\n\nАвтор поста: {message.from_user.first_name} @{message.from_user.username}',
photo=message.photo[-1].file_id,
reply_markup=markup
)
# TODO: Не понятна реализация с альбомами от слова совсем
# elif message.content_type == 'photo' and message.media_group_id != None:
# bot.forward_message(chat_id=IMPORTANT_LOGS, from_chat_id=message.chat.id, message_id=message.message_id )
else:
pass
except Exception as e:
if LOGS:
self.bot.send_message(chat_id=IMPORTANT_LOGS,
text=f"Произошла ошибка: {str(e)}\n\nTraceback:\n{traceback.format_exc()}")
markup_for_user = self.get_reply_keyboard(message)
success_send_message = messages.get_message(self.__get_first_name(message), 'SUCCESS_SEND_MESSAGE')
self.bot.send_message(message.chat.id, success_send_message, parse_mode='html',
disable_web_page_preview=not PREVIEW_LINK, reply_markup=markup_for_user)
@staticmethod
def get_reply_keyboard(message):
markup = types.ReplyKeyboardMarkup(resize_keyboard=True, one_time_keyboard=True)
item1 = types.KeyboardButton("📢Предложить свой пост")
item2 = types.KeyboardButton("📩Связаться с админами")
item3 = types.KeyboardButton("👋🏼Сказать пока!")
#TODO: Есть ощущение что не совсем так работает как надо
item4 = types.KeyboardButton("🤪Хочу стикеры") if not BotDB.get_info_about_stickers(
user_id=message.from_user.id) else None
if item4:
markup.add(item1, item2, item3, item4)
else:
markup.add(item1, item2, item3)
return markup
@staticmethod
def __get_first_name(message):
return message.from_user.first_name
@staticmethod
def check_access(user_id: int):
"""Проверка прав на совершение действий"""
return BotDB.is_admin(user_id)
@staticmethod
def add_days_to_date(days):
"""Прибавляет указанное количество дней к текущей дате и возвращает дату в формате DD-MM-YYYY."""
current_date = datetime.now()
future_date = current_date + timedelta(days=int(days))
formatted_date = future_date.strftime("%d-%m-%Y")
return formatted_date
@staticmethod
def create_keyboard_with_pagination(page: int, total_items: int, array_items: list[tuple[Any, Any]], callback: str):
"""
Создает клавиатуру с пагинацией для заданного набора элементов и устанавливает необходимый callback
Args:
page: Номер текущей страницы.
total_items: Общее количество элементов.
array_items: Лист кортежей. Содержит в себе user_name: user_id
callback: Действие в коллбеке. Вернет callback вида ({callback}_{user_id})
Returns:
InlineKeyboardMarkup: Клавиатура с кнопками пагинации.
"""
# Определяем общее количество страниц
total_pages = (total_items + 7 - 1) // 7
page = page
# Создаем список кнопок
buttons = []
# Вычисляем стартовый номер для текущей страницы
start_index = (page - 1) * 7 #тут было +1, убрал, потому что на текстовом массиве выходит за пределы
# Кнопки с номерами страниц
for i in range(start_index, min(start_index + 7,
len(array_items))): #тут было len(array_items) +1, убрал, потому что на текстовом массиве выходит за пределы
buttons.append(
types.InlineKeyboardButton(f"{array_items[i][0]}", callback_data=f"{callback}_{array_items[i][1]}"))
# Добавляем кнопки "Предыдущая" и "Следующая"
if int(page) > 1:
buttons.insert(6, types.InlineKeyboardButton("⬅️ Предыдущая", callback_data=f"page_{page - 1}"))
if page < total_pages:
buttons.append(types.InlineKeyboardButton("➡️ Следующая", callback_data=f"page_{page + 1}"))
#Добавляем кнопку назад
buttons.append(types.InlineKeyboardButton("🏠 Назад", callback_data="return"))
# Формируем клавиатуру с 3 кнопками в ряд
keyboard = []
for i in range(0, len(buttons), 3):
keyboard.append(buttons[i:i + 3])
return types.InlineKeyboardMarkup(keyboard)
@staticmethod
def get_banned_users_list(offset: int):
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
Args:
offset: отступ для запроса в базу данных
Returns:
message - текст сообщения
user_ids - лист кортежей [(user_name: user_id)]
"""
users = BotDB.get_banned_users_from_db_with_limits(limit=7, offset=offset)
message = "Список заблокированных пользователей:\n"
for user in users:
message += f"Пользователь: {user[0]}\n"
message += f"Причина бана: {user[2]}\n"
message += f"Дата разбана: {user[3]}\n\n"
return message
@staticmethod
def get_banned_users_buttons():
"""
Возвращает сообщение со списком пользователей и словарь с ником + идентификатором
Args:
offset: отступ для запроса в базу данных
Returns:
message - текст сообщения
user_ids - лист кортежей [(user_name: user_id)]
"""
users = BotDB.get_banned_users_from_db()
user_ids = []
for user in users:
user_ids.append((user[0], user[1]))
return user_ids
@staticmethod
def delete_user_blacklist(user_id):
return BotDB.delete_user_blacklist(user_id=user_id)
bot = TelegramHelperBot(BOT_TOKEN)
if __name__ == "__main__":
# Запускаем бота
bot.start()
scheduler = BackgroundScheduler()
scheduler.add_job(bot.unban_notifier(), 'cron', hour=0, minute=0)
scheduler.start()

View File

@@ -1,8 +1,16 @@
import os
from db import BotDB
from database.db import BotDB
# Получаем текущий рабочий каталог
current_dir = os.path.dirname(os.path.abspath(__file__))
BotDB = BotDB()
# Переходим на уровень выше, чтобы выйти из папки migrations/
parent_dir = os.path.dirname(current_dir)
# Строим путь до файла
tg_bot_database_path = os.path.join(parent_dir, "tg-bot-database")
BotDB = BotDB(f'{tg_bot_database_path}')
def get_filename():

View File

@@ -1,7 +1,7 @@
import os
from db import BotDB
from database.db import BotDB
BotDB = BotDB()
BotDB = BotDB('tg-bot-database')
def get_filename():

8
pytest.ini Normal file
View File

@@ -0,0 +1,8 @@
[pytest]
pythonpath = .
python_files = test_*.py *_test.py
python_functions = test_*
testpaths = tests
[report]
omit = *myenv/*, custom_logger.py, *venv/*, tests/*

View File

@@ -1,3 +1,22 @@
pyTelegramBotAPI
APScheduler~=3.10.4
loguru~=0.7.2
APScheduler==3.10.4
certifi~=2024.6.2
charset-normalizer==3.3.2
coverage==7.5.4
exceptiongroup==1.2.1
idna==3.7
iniconfig==2.0.0
loguru==0.7.2
packaging==24.1
pluggy==1.5.0
pytest==8.2.2
pytz==2024.1
requests==2.32.3
six==1.16.0
tomli==2.0.1
tzlocal==5.2
urllib3~=2.2.1
pip~=23.2.1
attrs~=23.2.0
typing_extensions~=4.12.2
aiohttp~=3.9.5
aiogram~=3.10.0

7
run_helper.py Normal file
View File

@@ -0,0 +1,7 @@
import asyncio
from helper_bot.main import start_bot
from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
if __name__ == '__main__':
asyncio.run(start_bot(BaseDependencyFactory()))

825
tests/test_db.py Normal file
View File

@@ -0,0 +1,825 @@
from datetime import datetime
import os
import sqlite3
import pytest
from database.db import BotDB
@pytest.fixture
def bot():
"""Фикстура для создания объекта BotDB."""
return BotDB("test.db")
@pytest.fixture(autouse=True, )
def setup_db():
"""Фикстура для создания всей базы перед каждым тестом."""
# Mock data 1st user
user_id = 12345
first_name = "Иван"
full_name = "Иван Иванович"
username = "@iban"
message_text = 'Hello, planet'
message_id = 1
message_for_user = "LOL"
has_stickers = 0
# Mock data 2nd user
user_id_2 = 14278
first_name_2 = "Борис"
full_name_2 = "Борис Петрович"
username_2 = "@boris"
message_text_2 = 'Hello, world'
message_id_2 = 2
message_for_user_2 = "LOL2"
has_stickers_2 = 1
# Other data
date = "2024-07-10"
next_date = "2024-07-11"
conn = sqlite3.connect("test.db")
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS "admins" (
user_id INTEGER NOT NULL,
"role" TEXT
);
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS "audio_message_reference"
(
"id" INTEGER NOT NULL UNIQUE,
"file_name" TEXT NOT NULL UNIQUE,
"author_id" INTEGER NOT NULL,
"date_added" DATE NOT NULL,
"listen_count" INTEGER NOT NULL,
"file_id" INTEGER NOT NULL,
PRIMARY KEY ("id")
);
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS "blacklist"
(
"user_id" INTEGER NOT NULL UNIQUE,
"user_name" INTEGER,
"message_for_user" INTEGER,
"date_to_unban" INTEGER
);
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS "messages" (
"ID" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
"Message" TEXT NOT NULL,
"type" INTEGER
);
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS "our_users" (
"id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
"user_id" INTEGER NOT NULL UNIQUE,
"first_name" STRING,
"full_name" STRING,
"username" STRING,
"is_bot" BOOLEAN,
"language_code" STRING,
"has_stickers" INTEGER NOT NULL DEFAULT 0,
"date_added" DATE NOT NULL,
"date_changed" DATE NOT NULL
, state_user TEXT(20));
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS user_messages (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
message_text TEXT,
user_id INTEGER,
message_id INTEGER NOT NULL,
date TEXT
);
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT);
""")
cursor.execute("""
CREATE TABLE migrations (
version INTEGER PRIMARY KEY NOT NULL,
script_name TEXT NOT NULL,
created_at TEXT
);
""")
#blacklist mock data
cursor.execute("INSERT INTO blacklist (user_id, user_name, message_for_user, date_to_unban) VALUES (?, ?, ?, ?)",
(user_id, username, message_for_user, next_date))
cursor.execute("INSERT INTO blacklist (user_id, user_name, message_for_user, date_to_unban) VALUES (?, ?, ?, ?)",
(user_id_2, username_2, message_for_user_2, date))
#our_users mock data
cursor.execute(
"INSERT INTO our_users (user_id, first_name, full_name, username, date_added, date_changed, has_stickers)"
" VALUES (?, ?, ?, ?, ?, ?, ?)", (user_id, first_name, full_name, username, date, date, has_stickers)
)
cursor.execute(
"INSERT INTO our_users (user_id, first_name, full_name, username, date_added, date_changed, has_stickers)"
" VALUES (?, ?, ?, ?, ?, ?, ?)", (user_id_2, first_name_2, full_name_2, username_2, date, date, has_stickers_2)
)
#messages mock data
cursor.execute(
"INSERT INTO user_messages (message_text, user_id, message_id, date) "
"VALUES (?, ?, ?, ?)",
(message_text, user_id, message_id, date))
cursor.execute(
"INSERT INTO user_messages (message_text, user_id, message_id, date) "
"VALUES (?, ?, ?, ?)",
(message_text_2, user_id_2, message_id_2, date))
#mock admins
cursor.execute(
"INSERT INTO admins (user_id, role) "
"VALUES (?, ?)",
(user_id, 'creator'))
conn.commit()
conn.close()
yield
os.remove('test.db')
def test_bot_init(bot):
"""Проверяет, что объект BotDB инициализируется с правильным именем файла."""
assert bot.db_file == os.path.join(os.getcwd(), "test.db")
# Проверьте, что соединения с базой данных нет, так как оно не устанавливается в init
assert bot.conn is None
assert bot.cursor is None
def test_bot_connect(bot):
"""Проверяет, что метод connect создает подключение к базе данных."""
bot.connect()
assert bot.conn is not None
assert bot.cursor is not None
bot.close()
@pytest.mark.xfail
def test_bot_close(bot):
"""Проверяет, что метод close закрывает подключение к базе данных."""
bot.connect()
assert bot.conn is not None
assert bot.cursor is not None
bot.close()
assert bot.conn is None
assert bot.cursor is None
def test_create_table_success(bot):
sql_script = 'CREATE TABLE test_table (id INTEGER PRIMARY KEY);'
bot.create_table(sql_script)
# Проверяем, что таблица создана
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test_table'")
result = cursor.fetchone()
conn.close()
assert result is not None
def test_create_table_error(bot):
sql_script = 'CREATE TABLE test_table (id INTEGER PRIMARY KEY);'
bot.create_table(sql_script)
with pytest.raises(sqlite3.OperationalError):
bot.create_table(sql_script)
def test_get_current_version_success(bot):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("INSERT INTO migrations (version, script_name) VALUES (123, 'test')")
conn.commit()
conn.close()
# Вызываем функцию и проверяем результат
version = bot.get_current_version()
assert version == 123
def test_get_current_version_error(bot):
__drop_table('migrations')
with pytest.raises(sqlite3.OperationalError):
bot.get_current_version()
def test_update_version_success(bot):
# Вызываем функцию update_version
new_version = 124
script_name = "migration_script.sql"
bot.update_version(new_version, script_name)
# Проверяем, что данные записаны в таблицу
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM migrations WHERE version = ?", (new_version,))
result = cursor.fetchone()
conn.close()
assert result is not None
assert result[0] == new_version
assert result[1] == script_name
assert result[2] == datetime.now().strftime("%d-%m-%Y %H:%M:%S")
def test_update_version_integrity_error(bot):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("INSERT INTO migrations (version, script_name) VALUES (123, 'test')")
conn.commit()
conn.close()
# Пытаемся обновить версию с уже существующим значением
with pytest.raises(sqlite3.IntegrityError):
bot.update_version(123, "script_2.sql")
def test_update_version_error(bot):
__drop_table('migrations')
with pytest.raises(sqlite3.OperationalError):
bot.update_version(123, "script_2.sql")()
def test_add_new_user_in_db(bot):
"""Проверяет добавление нового пользователя в базу данных."""
user_id = 50
first_name = "Петр"
full_name = "Петр Иванов"
username = "@petr_ivanov"
is_bot = False
language_code = "ru"
date_added = "2024-07-09"
date_changed = "2024-07-09"
# Вызываем функцию add_new_user_in_db
bot.add_new_user_in_db(
user_id, first_name, full_name, username, is_bot, language_code, date_added, date_changed
)
# Проверяем наличие записи в базе данных
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM our_users WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
conn.close()
assert result is not None
assert result[1] == user_id
assert result[2] == first_name
assert result[3] == full_name
assert result[4] == username
assert result[5] == is_bot
assert result[6] == language_code
assert result[8] == date_added
assert result[9] == date_changed
def test_add_new_user_in_db_duplicate_user_id(bot, setup_db):
"""Проверяет поведение при попытке добавить пользователя с уже существующим user_id."""
user_id = 12345
# Попытка добавить пользователя с тем же user_id
with pytest.raises(sqlite3.IntegrityError):
bot.add_new_user_in_db(
user_id, "Марина", "Марина Альфредовна", "marina", False, "bg", "2024-07-09", "2024-07-09"
)
def test_add_new_user_in_db_empty_first_name(bot):
""" Проверяет добавление пользователя с пустым именем (first_name) """
user_id = 43
first_name = "" # Пустое имя
full_name = "Boris Petrov"
username = "@boris"
is_bot = False
language_code = "fr"
date_added = "2024-07-09"
date_changed = "2024-07-09"
# Вызываем функцию add_new_user_in_db
bot.add_new_user_in_db(
user_id, first_name, full_name, username, is_bot, language_code, date_added, date_changed
)
# Проверяем наличие записи в базе данных
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute(f"SELECT * FROM our_users WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
conn.close()
assert result is not None
assert result[1] == user_id
assert result[2] == first_name
assert result[3] == full_name
assert result[4] == username
assert result[5] == is_bot
assert result[6] == language_code
assert result[8] == date_added
assert result[9] == date_changed
def test_user_exists_found(bot):
"""Проверяет, что функция возвращает True, если пользователь найден."""
user_id = 12345
# Проверяем наличие записи в базе данных
assert bot.user_exists(user_id) is True
def test_user_exists_not_found(bot):
"""Проверяет, что функция возвращает False, если пользователь не найден."""
user_id = 99999
assert bot.user_exists(user_id) is False
def test_user_exists_error(bot):
"""Проверяет, что функция возвращает ошибки"""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.user_exists(12345)
def test_get_user_id_found(bot):
"""Проверяет, что функция возвращает ID пользователя, если он найден."""
user_id = 12345
# Проверяем, что возвращается правильный ID из базы
user_id_db = bot.get_user_id(user_id)
assert user_id_db == 1
def test_get_user_id_not_found(bot, setup_db):
"""Проверяет, что функция возвращает None, если пользователь не найден."""
user_id = 99999
assert bot.get_user_id(user_id) is None
def test_get_user_id_error(bot):
"""Проверяет, что функция обрабатывает некорректный user_id."""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.get_user_id(12345)
def test_get_username_found(bot):
"""Проверяет, что функция возвращает username пользователя, если он найден."""
user_id = 12345
username = "@iban"
# Проверяем, что возвращается правильный username из базы
username_db = bot.get_username(user_id)
assert username_db == username
def test_get_username_not_found(bot, setup_db):
"""Проверяет, что функция возвращает None, если пользователь не найден."""
user_id = 99999
assert bot.get_username(user_id) is None
def test_get_username_error(bot):
"""Проверяет, что функция возвращает ошибку"""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.get_username(12345)
def test_get_all_user_id_empty(bot):
"""Проверяет, что функция возвращает пустой список, если в базе нет пользователей."""
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("DELETE FROM our_users")
conn.commit()
conn.close()
# Проверяем наличие записей в базе данных
user_ids = bot.get_all_user_id()
assert user_ids == []
def test_get_all_user_id_non_empty(bot, setup_db):
"""Проверяет, что функция возвращает список всех user_id из базы данных."""
# Проверяем наличие записи в базе данных
user_ids = bot.get_all_user_id()
assert user_ids == [12345, 14278] # Проверяем, что в списке два ожидаемых user_id
def test_get_all_user_id_error(bot):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.get_all_user_id()
def test_get_user_first_name_found(bot):
"""Проверяет, что функция возвращает имя пользователя, если он найден."""
user_id = 12345
first_name = bot.get_user_first_name(user_id)
assert first_name == "Иван"
def test_get_user_first_name_not_found(bot, setup_db):
"""Проверяет, что функция возвращает None, если пользователь не найден."""
user_id = 99999
assert bot.get_user_first_name(user_id) is None
@pytest.mark.xfail
def test_get_user_first_name_invalid_user_id(bot):
"""Проверяет, что функция обрабатывает некорректный user_id."""
with pytest.raises(sqlite3.Error):
bot.get_user_first_name("invalid_user_id") # Передача строки
def test_get_user_first_name_error(bot):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.get_user_first_name(12345)
def test_get_info_about_stickers_found_received(bot):
"""Проверяет, что функция возвращает True, если пользователь получил стикеры."""
user_id = 14278
assert bot.get_info_about_stickers(user_id) is True
def test_get_info_about_stickers_found_not_received(bot, setup_db):
"""Проверяет, что функция возвращает False, если пользователь не получил стикеры."""
user_id = 12345
assert bot.get_info_about_stickers(user_id) is False
@pytest.mark.xfail
def test_get_info_about_stickers_not_found(bot, setup_db):
"""Проверяет, что функция возвращает None, если пользователь не найден."""
user_id = 99999
assert bot.get_info_about_stickers(user_id) is None
@pytest.mark.xfail
def test_get_info_about_stickers_invalid_user_id(bot):
"""Проверяет, что функция обрабатывает некорректный user_id."""
with pytest.raises(sqlite3.Error):
bot.get_info_about_stickers("invalid_user_id")
def test_get_info_about_stickers_error(bot):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.get_info_about_stickers(12345)
def test_update_info_about_stickers_success(bot):
"""Проверяет, что функция успешно обновляет информацию о получении стикеров."""
user_id = 12345
bot.update_info_about_stickers(user_id)
# Проверяем, что информация обновлена
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT has_stickers FROM our_users WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
conn.close()
assert result[0] == 1
def test_update_info_about_stickers_not_found(bot):
"""Проверяет, что функция не вызывает ошибки, если пользователь не найден."""
user_id = 99999
bot.update_info_about_stickers(user_id)
# Проверяем, что база данных не изменилась
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM our_users WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
conn.close()
assert result[0] == 0
def test_update_info_about_stickers_error(bot):
"""Проверяет, что функция вызывает ошибки"""
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.update_info_about_stickers(12345)
def test_get_users_blacklist_empty(bot):
"""Проверяет, что функция возвращает пустой словарь, если в черном списке нет пользователей."""
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("DELETE FROM blacklist")
conn.commit()
conn.close()
blacklist = bot.get_users_blacklist()
assert blacklist == {}
def test_get_users_blacklist_non_empty(bot):
"""Проверяет, что функция возвращает словарь с пользователями из черного списка."""
blacklist = bot.get_users_blacklist()
assert blacklist == {12345: "@iban", 14278: "@boris"}
def test_get_users_blacklist_error(bot):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.get_users_blacklist()
def test_get_blacklist_users_by_id_found(bot, setup_db):
"""Проверяет, что функция возвращает информацию о пользователе, если он найден в черном списке."""
user_id = 12345
result = bot.get_blacklist_users_by_id(user_id)
assert result == (12345, "@iban", "LOL", "2024-07-11")
def test_get_blacklist_users_by_id_not_found(bot, setup_db):
"""Проверяет, что функция возвращает None, если пользователь не найден в черном списке."""
user_id = 99999
assert bot.get_blacklist_users_by_id(user_id) is None
@pytest.mark.xfail
def test_get_blacklist_users_by_id_invalid_user_id(bot):
"""Проверяет, что функция обрабатывает некорректный user_id."""
with pytest.raises(sqlite3.Error):
bot.get_blacklist_users_by_id("invalid_user_id") # Передача строки
def test_get_blacklist_users_by_id_error(bot):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.get_blacklist_users_by_id(12345)
def test_get_users_for_unblock_today_found(bot):
"""Проверяет, что функция возвращает словарь с пользователями, у которых истекает блокировка сегодня."""
date_to_unban = "2024-07-11"
result = bot.get_users_for_unblock_today(date_to_unban)
assert result == {12345: "@iban"}
def test_get_users_for_unblock_today_not_found(bot, setup_db):
"""Проверяет, что функция возвращает пустой словарь, если сегодня нет пользователей, у которых истекает блокировка."""
date_to_unban = "2024-07-12"
result = bot.get_users_for_unblock_today(date_to_unban)
assert result == {}
def test_get_users_for_unblock_today_error(bot):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.get_users_for_unblock_today("2023-12-26")
def test_check_user_in_blacklist_found(bot, setup_db):
"""Проверяет, что функция возвращает True, если пользователь найден в черном списке."""
user_id = 12345
bot.set_user_blacklist(user_id, "JohnDoe") # Добавляем пользователя в черный список
assert bot.check_user_in_blacklist(user_id) is True
def test_check_user_in_blacklist_not_found(bot, setup_db):
"""Проверяет, что функция возвращает False, если пользователь не найден в черном списке."""
user_id = 99999
assert bot.check_user_in_blacklist(user_id) is False
def test_check_user_in_blacklist_error(bot, setup_db):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.check_user_in_blacklist(12345)
def test_set_user_blacklist_success(bot):
"""Проверяет, что функция успешно добавляет пользователя в черный список."""
user_id = 11
user_name = "Гриша"
message_for_user = "Лови бан!"
date_to_unban = datetime.now().strftime("%Y-%m-%d") # Текущая дата
assert bot.set_user_blacklist(user_id, user_name, message_for_user, date_to_unban) is None
# Проверяем, что запись добавлена в базу
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM blacklist WHERE user_id = ?", (user_id,))
result = cursor.fetchone()
conn.commit()
conn.close()
assert result is not None
assert result[1] == user_name
assert result[2] == message_for_user
assert result[3] == date_to_unban
@pytest.mark.xfail
def test_set_user_blacklist_duplicate_user_id(bot, setup_db):
"""Проверяет, что функция не добавляет дубликат user_id в черный список."""
user_id = 12345
bot.set_user_blacklist(user_id, "JohnDoe")
with pytest.raises(sqlite3.IntegrityError):
bot.set_user_blacklist(user_id, "JaneSmith") # Попытка добавить дубликат
@pytest.mark.xfail
def test_set_user_blacklist_error(bot, setup_db):
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.set_user_blacklist(12345, "JohnDoe", "You are banned!", "2024-01-01")
def test_delete_user_blacklist_success(bot):
bot.delete_user_blacklist(12345)
assert bot.check_user_in_blacklist(12345) is False
@pytest.mark.xfail
def test_delete_user_blacklist_not_found(bot):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("INSERT INTO blacklist (user_id, user_name, date_to_unban) VALUES (?, ?, ?)",
(12345, "JohnDoe", "2023-12-26"))
conn.commit()
conn.close()
result = bot.delete_user_blacklist(514)
assert result is False
@pytest.mark.xfail
def test_delete_user_blacklist_error(bot):
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.delete_user_blacklist(12345)
def test_add_new_message_in_db_success(bot):
result = bot.add_new_message_in_db('hello', 4232187, 5, '2024-01-01')
assert result is None
def test_add_new_message_in_db_error(bot):
__drop_table('user_messages')
with pytest.raises(sqlite3.Error):
bot.add_new_message_in_db('hello', 12345, 1, '2024-01-01')
def test_update_date_for_user_success(bot):
bot.update_date_for_user('2024-07-15', 12345)
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("SELECT date_changed FROM our_users WHERE user_id = ?", (12345,))
new_date = cursor.fetchone()[0]
conn.close()
assert new_date == '2024-07-15'
@pytest.mark.xfail
def test_update_date_for_user_error(bot):
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.update_date_for_user('2024-07-15', 12345)
def test_is_admin_success(bot):
assert bot.is_admin(12345) is True
def test_is_admin_not_found(bot):
assert bot.is_admin(1) is False
def test_is_admin_error(bot):
__drop_table('admins')
assert bot.is_admin(1) is None
def test_get_user_by_message_id_success(bot):
assert bot.get_user_by_message_id(1) == 12345
@pytest.mark.xfail
def test_get_user_by_message_id_not_found(bot):
assert bot.get_user_by_message_id(124) == None
def test_get_user_by_message_id_error(bot):
__drop_table('user_messages')
with pytest.raises(sqlite3.Error):
bot.get_user_by_message_id(14)
def test_get_last_users_from_db_success(bot):
users = bot.get_last_users_from_db()
assert users is not None
assert len(users) == 2
def test_get_last_users_from_db_empty(bot):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("DELETE FROM our_users")
conn.commit()
conn.close()
users = bot.get_last_users_from_db()
assert users == []
assert len(users) == 0
def test_get_user_by_message_id_error(bot):
__drop_table('our_users')
with pytest.raises(sqlite3.Error):
bot.get_last_users_from_db()
def test_get_banned_users_from_db_success(bot):
users = bot.get_banned_users_from_db()
assert users[0][0] == '@iban'
assert users[0][1] == 12345
assert users[0][2] == 'LOL'
assert users[1][0] == '@boris'
assert users[1][1] == 14278
assert users[1][2] == 'LOL2'
def test_get_banned_users_from_db_empty(bot):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("DELETE FROM blacklist")
conn.commit()
conn.close()
users = bot.get_banned_users_from_db()
assert users == []
assert len(users) == 0
def test_get_banned_users_from_db_error(bot):
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.get_banned_users_from_db()
def test_get_banned_users_from_db_with_limits_success_limit(bot):
users = bot.get_banned_users_from_db_with_limits(0, 1)
assert users[0][0] == '@iban'
assert users[0][1] == 12345
assert users[0][2] == 'LOL'
assert len(users) == 1
def test_get_banned_users_from_db_with_limits_success_offset(bot):
users = bot.get_banned_users_from_db_with_limits(1, 2)
assert users[0][0] == '@boris'
assert users[0][1] == 14278
assert users[0][2] == 'LOL2'
assert len(users) == 1
def test_get_banned_users_from_db_with_limits_empty(bot):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute("DELETE FROM blacklist")
conn.commit()
conn.close()
users = bot.get_banned_users_from_db_with_limits(0, 2)
assert users == []
assert len(users) == 0
def test_get_banned_users_from_db_with_limits_error(bot):
__drop_table('blacklist')
with pytest.raises(sqlite3.Error):
bot.get_banned_users_from_db_with_limits(0, 2)
def __drop_table(table_name: str):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
cursor.execute(f"DROP TABLE {table_name}")
conn.commit()
conn.close()
if __name__ == "__main__":
pytest.main()

View File

@@ -4,8 +4,7 @@ import sys
from pathlib import Path
from time import sleep
import db
from db import BotDB
from database.db import BotDB
import telebot
import random
from datetime import datetime
@@ -31,7 +30,7 @@ TEST = config.getboolean('Settings', 'test')
#Инициализируем бота и базку
bot = telebot.TeleBot(BOT_TOKEN, parse_mode=None)
BotDB = BotDB('tg-bot-database')
BotDB = BotDB('database/tg-bot-database')