390 lines
16 KiB
Python
390 lines
16 KiB
Python
"""
|
||
Middleware для проверки сообщений на запрещённые слова (банворды).
|
||
|
||
Pipeline проверки:
|
||
1. Пропускаем админов и служебные сообщения
|
||
2. Проверяем whitelist (исключения)
|
||
3. Проверяем режим silence (удаляем всё)
|
||
4. Проверяем режим conflict (конфликтные слова)
|
||
5. Проверяем постоянные банворды (substring, lemma, part)
|
||
6. Проверяем временные банворды
|
||
7. Если найдено - удаляем, логируем, уведомляем админов
|
||
|
||
НОВОЕ: Все проверки работают с нормализацией повторяющихся букв (3+ → 1).
|
||
"""
|
||
from typing import Callable, Dict, Any, Awaitable, Optional
|
||
import re
|
||
|
||
from aiogram import BaseMiddleware
|
||
from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton
|
||
from aiogram.exceptions import TelegramBadRequest
|
||
|
||
from configs import settings
|
||
from database import get_manager, BanWordType
|
||
from bot.special import process_text, extract_words, get_lemma
|
||
from middleware.loggers import logger
|
||
|
||
__all__ = ("BanWordsMiddleware",)
|
||
|
||
|
||
class BanWordsMiddleware(BaseMiddleware):
|
||
"""
|
||
Middleware для фильтрации сообщений с банвордами.
|
||
|
||
Проверяет каждое текстовое сообщение на наличие запрещённых слов,
|
||
удаляет спам и уведомляет администраторов.
|
||
"""
|
||
|
||
def __init__(self):
|
||
"""Инициализирует middleware"""
|
||
super().__init__()
|
||
self.manager = get_manager()
|
||
|
||
async def __call__(
|
||
self,
|
||
handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
|
||
event: Message,
|
||
data: Dict[str, Any]
|
||
) -> Any:
|
||
"""
|
||
Обрабатывает входящие сообщения.
|
||
|
||
Args:
|
||
handler: Следующий обработчик в цепочке
|
||
event: Сообщение от пользователя
|
||
data: Данные из диспетчера
|
||
|
||
Returns:
|
||
Any: Результат обработчика или None (если сообщение удалено)
|
||
"""
|
||
# Пропускаем не-текстовые сообщения
|
||
if not event.text and not event.caption:
|
||
return await handler(event, data)
|
||
|
||
# Получаем текст (из text или caption)
|
||
message_text = event.text or event.caption
|
||
|
||
# Пропускаем команды (начинаются с /)
|
||
if message_text.startswith('/'):
|
||
return await handler(event, data)
|
||
|
||
# Проверяем, является ли пользователь админом
|
||
user_id = event.from_user.id
|
||
is_super_admin = user_id in settings.OWNER_ID
|
||
is_admin = is_super_admin or self.manager.is_admin_cached(user_id)
|
||
|
||
# Админы пропускаются
|
||
if is_admin:
|
||
return await handler(event, data)
|
||
|
||
# Проверяем сообщение на банворды
|
||
spam_result = await self._check_message(message_text)
|
||
|
||
if spam_result:
|
||
# Найден спам - удаляем и уведомляем
|
||
await self._handle_spam(event, spam_result)
|
||
return None # Не продолжаем обработку
|
||
|
||
# Сообщение чистое - пропускаем дальше
|
||
return await handler(event, data)
|
||
|
||
@staticmethod
|
||
def _normalize_for_part_check(text: str) -> str:
|
||
"""
|
||
Нормализует текст для проверки частей слов.
|
||
Удаляет ВСЕ символы кроме букв и цифр, приводит к нижнему регистру.
|
||
|
||
Args:
|
||
text: Исходный текст
|
||
|
||
Returns:
|
||
str: Нормализованный текст (только буквы и цифры, нижний регистр)
|
||
|
||
Examples:
|
||
"@Astrixkeepbot" -> "astrixkeepbot"
|
||
"hello@world.com" -> "helloworldcom"
|
||
"test_123-456" -> "test123456"
|
||
"""
|
||
return re.sub(r'[^a-zA-Zа-яА-ЯёЁ0-9]', '', text.lower())
|
||
|
||
@staticmethod
|
||
def _normalize_repeated_chars(text: str, max_repeats: int = 1) -> str:
|
||
"""
|
||
Убирает повторяющиеся буквы (обход "лееейн" -> "лейн", "телееелооог" -> "телелог").
|
||
|
||
Args:
|
||
text: Исходное слово
|
||
max_repeats: Максимальное количество повторов одной буквы (1 = убрать все повторы)
|
||
|
||
Returns:
|
||
str: Нормализованное слово
|
||
|
||
Examples:
|
||
("лееейн", 1) -> "лейн"
|
||
("телееелооог", 1) -> "телелог"
|
||
("хеееелооооу", 1) -> "хелоу"
|
||
("аааааа", 1) -> "а"
|
||
("привеееет", 2) -> "приввеет" (если max_repeats=2)
|
||
"""
|
||
if max_repeats == 1:
|
||
# Заменяем 2+ одинаковых букв подряд на 1 такую же букву
|
||
return re.sub(r'([а-яёa-z])\1+', r'\1', text, flags=re.IGNORECASE)
|
||
else:
|
||
# Заменяем (max_repeats+1)+ одинаковых букв на max_repeats таких букв
|
||
pattern = f'([а-яёa-z])\\1{{{max_repeats},}}'
|
||
replacement = '\\1' * max_repeats
|
||
return re.sub(pattern, replacement, text, flags=re.IGNORECASE)
|
||
|
||
async def _check_message(self, text: str) -> Optional[Dict[str, str]]:
|
||
"""
|
||
Проверяет сообщение на наличие банвордов.
|
||
|
||
Args:
|
||
text: Текст сообщения
|
||
|
||
Returns:
|
||
Optional[Dict]: {"word": "найденное_слово", "type": "тип_проверки"} или None
|
||
"""
|
||
# Нормализуем текст для проверки
|
||
text_lower = text.lower()
|
||
text_processed = process_text(text_lower)
|
||
|
||
# Дополнительно нормализуем повторяющиеся буквы для всех проверок
|
||
text_normalized = self._normalize_repeated_chars(text_processed, max_repeats=1)
|
||
|
||
logger.debug(
|
||
f"Проверка текста: исходный='{text[:50]}', обработанный='{text_processed[:50]}', "
|
||
f"нормализованный='{text_normalized[:50]}'",
|
||
log_type="BANWORDS"
|
||
)
|
||
|
||
# === 1. WHITELIST (исключения) ===
|
||
# Проверяем оба варианта: с повторами и без
|
||
if self.manager.is_whitelisted(text_processed) or self.manager.is_whitelisted(text_normalized):
|
||
logger.debug(
|
||
f"Сообщение содержит whitelist слово",
|
||
log_type="BANWORDS"
|
||
)
|
||
return None
|
||
|
||
# === 2. SILENCE MODE (удаляем всё) ===
|
||
if await self.manager.is_silence_active():
|
||
return {
|
||
"word": "[режим тишины]",
|
||
"type": "silence"
|
||
}
|
||
|
||
# === 3. CONFLICT MODE (конфликтные слова) ===
|
||
if await self.manager.is_conflict_active():
|
||
# Проверяем конфликтные подстроки (с нормализацией)
|
||
conflict_substring = self.manager.get_banwords_cached(
|
||
BanWordType.CONFLICT_SUBSTRING
|
||
)
|
||
for word in conflict_substring:
|
||
word_normalized = self._normalize_repeated_chars(word, max_repeats=1)
|
||
if word_normalized in text_normalized:
|
||
return {"word": word, "type": "conflict_substring"}
|
||
|
||
# Проверяем конфликтные леммы
|
||
conflict_lemma = self.manager.get_banwords_cached(
|
||
BanWordType.CONFLICT_LEMMA
|
||
)
|
||
words_in_text = extract_words(text_processed)
|
||
for word_text in words_in_text:
|
||
word_normalized = self._normalize_repeated_chars(word_text, max_repeats=1)
|
||
lemma = get_lemma(word_normalized)
|
||
|
||
if lemma in conflict_lemma:
|
||
return {"word": lemma, "type": "conflict_lemma"}
|
||
|
||
# === 4. SUBSTRING (подстроки) с нормализацией ===
|
||
substring_words = self.manager.get_banwords_cached(BanWordType.SUBSTRING)
|
||
for word in substring_words:
|
||
# Нормализуем и банворд, и текст
|
||
word_normalized = self._normalize_repeated_chars(word, max_repeats=1)
|
||
|
||
if word_normalized in text_normalized:
|
||
logger.info(
|
||
f"Найдена подстрока: '{word}' (норм: '{word_normalized}') в '{text_normalized[:100]}'",
|
||
log_type="BANWORDS"
|
||
)
|
||
return {"word": word, "type": "substring"}
|
||
|
||
# === 5. PART (части слов без пробелов и спецсимволов) ===
|
||
part_words = self.manager.get_banwords_cached(BanWordType.PART)
|
||
if part_words:
|
||
# Специальная нормализация для PART: удаляем ВСЁ кроме букв и цифр
|
||
text_part_normalized = self._normalize_for_part_check(text)
|
||
text_part_normalized = self._normalize_repeated_chars(text_part_normalized, max_repeats=1)
|
||
|
||
for part in part_words:
|
||
part_normalized = self._normalize_for_part_check(part)
|
||
part_normalized = self._normalize_repeated_chars(part_normalized, max_repeats=1)
|
||
|
||
if part_normalized in text_part_normalized:
|
||
logger.info(
|
||
f"Найдена запрещенная часть: '{part}' (норм: '{part_normalized}') "
|
||
f"в '{text_part_normalized[:100]}'",
|
||
log_type="BANWORDS"
|
||
)
|
||
return {"word": part, "type": "part"}
|
||
|
||
# === 6. LEMMA (нормальные формы слов) ===
|
||
lemma_words = self.manager.get_banwords_cached(BanWordType.LEMMA)
|
||
if lemma_words:
|
||
words_in_text = extract_words(text_processed)
|
||
for word_text in words_in_text:
|
||
# Убираем повторяющиеся буквы ПЕРЕД лемматизацией
|
||
word_normalized = self._normalize_repeated_chars(word_text, max_repeats=1)
|
||
lemma = get_lemma(word_normalized)
|
||
|
||
if lemma in lemma_words:
|
||
logger.info(
|
||
f"Найдена лемма: '{lemma}' из слова '{word_text}' (норм: '{word_normalized}')",
|
||
log_type="BANWORDS"
|
||
)
|
||
return {"word": lemma, "type": "lemma"}
|
||
|
||
# Банворды не найдены
|
||
return None
|
||
|
||
async def _handle_spam(
|
||
self,
|
||
message: Message,
|
||
spam_result: Dict[str, str]
|
||
) -> None:
|
||
"""
|
||
Обрабатывает найденный спам: удаляет, логирует, уведомляет.
|
||
|
||
Args:
|
||
message: Сообщение со спамом
|
||
spam_result: Результат проверки (слово + тип)
|
||
"""
|
||
user = message.from_user
|
||
matched_word = spam_result["word"]
|
||
match_type = spam_result["type"]
|
||
|
||
# Получаем текст сообщения
|
||
message_text = message.text or message.caption or "[нет текста]"
|
||
|
||
# === 1. УДАЛЯЕМ СООБЩЕНИЕ ===
|
||
try:
|
||
await message.delete()
|
||
logger.info(
|
||
f"Удалено сообщение от @{user.username or user.id} "
|
||
f"(слово: '{matched_word}', тип: {match_type})",
|
||
log_type="BANWORDS",
|
||
user=f"@{user.username}" if user.username else f"id{user.id}"
|
||
)
|
||
except TelegramBadRequest as e:
|
||
logger.error(
|
||
f"Не удалось удалить сообщение: {e}",
|
||
log_type="BANWORDS",
|
||
user=f"@{user.username}" if user.username else f"id{user.id}"
|
||
)
|
||
return
|
||
|
||
# === 2. ЛОГИРУЕМ В БД ===
|
||
await self.manager.log_spam(
|
||
user_id=user.id,
|
||
username=user.username or f"id{user.id}",
|
||
chat_id=message.chat.id,
|
||
message_text=message_text,
|
||
matched_word=matched_word,
|
||
match_type=match_type
|
||
)
|
||
|
||
# === 3. УВЕДОМЛЯЕМ АДМИНОВ ===
|
||
await self._notify_admins(message, matched_word, match_type, message_text)
|
||
|
||
async def _notify_admins(
|
||
self,
|
||
message: Message,
|
||
matched_word: str,
|
||
match_type: str,
|
||
message_text: str
|
||
) -> None:
|
||
"""
|
||
Отправляет уведомление в админский чат с кнопками.
|
||
|
||
Args:
|
||
message: Удалённое сообщение
|
||
matched_word: Слово, по которому сработал фильтр
|
||
match_type: Тип проверки
|
||
message_text: Текст сообщения
|
||
"""
|
||
user = message.from_user
|
||
username = f"@{user.username}" if user.username else f"ID: {user.id}"
|
||
|
||
# Получаем количество предыдущих нарушений
|
||
spam_count = await self.manager.get_user_spam_count(user.id)
|
||
|
||
# Формируем текст уведомления
|
||
notification_text = (
|
||
f"🚫 <b>Удалено сообщение</b>\n\n"
|
||
f"👤 <b>Пользователь:</b> {username}\n"
|
||
f"🆔 <b>ID:</b> <code>{user.id}</code>\n"
|
||
f"📊 <b>Нарушений:</b> {spam_count}\n\n"
|
||
f"🔍 <b>Триггер:</b> <code>{matched_word}</code>\n"
|
||
f"📝 <b>Тип:</b> {self._get_type_emoji(match_type)} {match_type}\n\n"
|
||
f"💬 <b>Текст:</b>\n"
|
||
f"<code>{self._escape_html(message_text[:500])}</code>"
|
||
)
|
||
|
||
# Создаём клавиатуру с действиями
|
||
keyboard = InlineKeyboardMarkup(inline_keyboard=[
|
||
[
|
||
InlineKeyboardButton(
|
||
text="🔨 Забанить",
|
||
callback_data=f"spam_ban:{user.id}:{message.chat.id}"
|
||
),
|
||
InlineKeyboardButton(
|
||
text="✅ Закрыть",
|
||
callback_data="spam_close"
|
||
)
|
||
],
|
||
[
|
||
InlineKeyboardButton(
|
||
text="📊 Статистика",
|
||
callback_data=f"spam_stats:{user.id}"
|
||
)
|
||
]
|
||
])
|
||
|
||
# Отправляем уведомление
|
||
try:
|
||
bot = message.bot
|
||
await bot.send_message(
|
||
chat_id=settings.ADMIN_CHAT_ID,
|
||
text=notification_text,
|
||
reply_markup=keyboard,
|
||
parse_mode="HTML"
|
||
)
|
||
except Exception as e:
|
||
logger.error(
|
||
f"Ошибка отправки уведомления админам: {e}",
|
||
log_type="BANWORDS"
|
||
)
|
||
|
||
@staticmethod
|
||
def _get_type_emoji(match_type: str) -> str:
|
||
"""Возвращает эмодзи для типа проверки"""
|
||
emoji_map = {
|
||
"substring": "🔤",
|
||
"lemma": "📖",
|
||
"part": "🧩",
|
||
"silence": "🔇",
|
||
"conflict_substring": "⚔️",
|
||
"conflict_lemma": "⚔️"
|
||
}
|
||
return emoji_map.get(match_type, "❓")
|
||
|
||
@staticmethod
|
||
def _escape_html(text: str) -> str:
|
||
"""Экранирует HTML символы для безопасного отображения"""
|
||
return (
|
||
text.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">")
|
||
)
|