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)