109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
from typing import Callable, Awaitable, Any, Dict
|
||
|
||
from aiogram import BaseMiddleware
|
||
from aiogram.types import Message, CallbackQuery, TelegramObject
|
||
|
||
from database import db
|
||
|
||
__all__ = ("BanCheckMiddleware",)
|
||
|
||
class BanCheckMiddleware(BaseMiddleware):
|
||
"""
|
||
Middleware для проверки забанен ли пользователь.
|
||
Если пользователь забанен в боте - блокирует все его действия.
|
||
"""
|
||
|
||
async def __call__(
|
||
self,
|
||
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
|
||
event: TelegramObject,
|
||
data: Dict[str, Any]
|
||
) -> Any:
|
||
"""
|
||
Проверяет каждый входящий запрос на наличие пользователя в черном списке.
|
||
|
||
Args:
|
||
handler: Следующий обработчик
|
||
event: Событие (сообщение, callback и т.д.)
|
||
data: Данные контекста
|
||
|
||
Returns:
|
||
Результат обработчика или None если пользователь забанен
|
||
"""
|
||
# Извлекаем информацию о пользователе из события
|
||
user = await self._extract_user(event)
|
||
|
||
if user is None:
|
||
# Не смогли определить пользователя - пропускаем
|
||
return await handler(event, data)
|
||
|
||
# Проверяем в базе данных статус пользователя
|
||
user_db = await db.get_user(user.id)
|
||
|
||
if user_db and user_db.status == "banned":
|
||
# Пользователь забанен - блокируем запрос
|
||
await self._send_ban_message(event, data)
|
||
return None
|
||
|
||
# Пользователь не забанен - пропускаем запрос дальше
|
||
return await handler(event, data)
|
||
|
||
@staticmethod
|
||
async def _extract_user(event: TelegramObject) -> Any:
|
||
"""
|
||
Извлекает пользователя из разных типов событий.
|
||
|
||
Args:
|
||
event: Событие Telegram
|
||
|
||
Returns:
|
||
Объект пользователя или None
|
||
"""
|
||
if isinstance(event, Message):
|
||
return event.from_user
|
||
elif isinstance(event, CallbackQuery):
|
||
return event.from_user
|
||
# Можно добавить другие типы событий при необходимости
|
||
return None
|
||
|
||
@staticmethod
|
||
async def _send_ban_message(event: TelegramObject, data: Dict[str, Any]) -> None:
|
||
"""
|
||
Отправляет сообщение о бане пользователю.
|
||
|
||
Args:
|
||
event: Событие которое triggered проверку
|
||
data: Данные контекста с ботом
|
||
"""
|
||
bot = data.get('bot')
|
||
|
||
if not bot:
|
||
return
|
||
|
||
chat_id = None
|
||
message_id = None
|
||
|
||
# Определяем куда отправлять сообщение в зависимости от типа события
|
||
if isinstance(event, Message):
|
||
chat_id = event.chat.id
|
||
message_id = event.message_id
|
||
elif isinstance(event, CallbackQuery) and event.message:
|
||
chat_id = event.message.chat.id
|
||
message_id = event.message.message_id
|
||
|
||
if chat_id:
|
||
try:
|
||
if isinstance(event, CallbackQuery):
|
||
# Для callback запросов отвечаем уведомлением
|
||
await event.answer("🚫 Вы заблокированы в боте!", show_alert=True)
|
||
else:
|
||
# Для сообщений отправляем новое сообщение
|
||
await bot.send_message(
|
||
chat_id=chat_id,
|
||
text="🚫 Вы заблокированы в боте!",
|
||
reply_to_message_id=message_id
|
||
)
|
||
except Exception:
|
||
# Игнорируем ошибки отправки (пользователь мог заблокировать бота)
|
||
pass
|