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.keyboards import get_reply_keyboard_admin, create_keyboard_with_pagination, \
create_keyboard_for_ban_days, create_keyboard_for_approve_ban, create_keyboard_for_ban_reason
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
admin_router = Router()
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 = bdf.get_db()
@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 ban_by_nickname(message: types.Message, state: FSMContext):
await message.answer('Пришли мне username блокируемого пользователя')
await state.set_state('PRE_BAN')
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
F.text == 'Отменить'
)
async def decline_ban(message: types.Message, state: FSMContext):
await state.set_data({})
await state.set_state("ADMIN")
logger.info(f"Отмена процедуры блокировки")
markup = get_reply_keyboard_admin()
await message.answer('Вернулись в меню', reply_markup=markup)
@admin_router.message(
ChatTypeFilter(chat_type=["private"]),
StateFilter("PRE_BAN")
)
async def ban_by_nickname_step_2(message: types.Message, state: FSMContext):
logger.info(
f"Функция ban_by_nickname_2. Получен никнейм пользователя: {message.text}")
user_name = message.text
user_id = BotDB.get_user_id_by_username(user_name)
await state.update_data(user_id=user_id, user_name=user_name, message_for_user=None,
date_to_unban=None)
full_name = BotDB.get_full_name_by_id(user_id)
markup = create_keyboard_for_ban_reason()
await message.answer(
text=f"Выбран пользователь:\nid: {user_id}\nusername: {user_name}\n"
f"Имя:{full_name}\nВыбери причину бана из списка или напиши ее в чат",
reply_markup=markup)
await state.set_state('BAN_2')
@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"),
F.text == 'Подтвердить'
)
async def approve_ban(message: types.Message, state: FSMContext):
user_data = await state.get_data()
logger.info(f"Переход на финальный шаг бана пользователя. Словарь с данными для бана: {user_data})")
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)