140 lines
6.8 KiB
Python
140 lines
6.8 KiB
Python
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
|
||
|
||
|
||
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 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)
|