Skip to content

T
telegram_bot_antispammer
Commit af057b09 authored by Иванова Мария Кирилловна's avatar Иванова Мария Кирилловна
Browse files
Options

Update bot.py

parent cd5806ce
Hide whitespace changes
Inline Side-by-side
Showing with 58 additions and 29 deletions
bot.py
View file @ af057b09
......@@ -13,6 +13,8 @@ import os
from dotenv import load_dotenv
import unicodedata
from aiogram.filters import Command
import logging
from aiogram import types
# Список слов фраз - указателей спама
SPAM_KEYWORDS = [
......@@ -38,7 +40,7 @@ SPAM_KEYWORDS = [
"лотерея", "быстрый выигрыш", "выигрыш", "приз", "деньги", "халява", "прибыль", "заработок", "зп", "подработка", "легкий заработок",
"мгновенно", "доход", "доход онлайн", "как заработать", "схема заработка",
"пассивный доход", "быстро заработать", "богатство", "всего за", "инвестировать", "инвестиция",
"биткоин", "криптовалюта", "forex", "деньги на карта", "18+", "дивиденд", "процент", "депозит", "выплата", "невероятный доход", "статус","гарантированный доход", "секрет богатство", "без риск",
"биткоин", "криптовалюта", "forex", "деньги на карта", "СБП", "18+", "дивиденд", "процент", "депозит", "выплата", "невероятный доход", "статус","гарантированный доход", "секрет богатство", "без риск",
# Психология, манипуляции и самопомощь
"гипноз", "секрет уверенность", "как стать успешный", "формула успех", "психотренинг", "управление человек", "контроль над разум", "манипуляция", "техника влияния", "мгновенный успех",
......@@ -81,8 +83,14 @@ SPAM_KEYWORDS = [
"🎁", "🔥", "💸", "💰", "💲", "💎", "🤑", "🤩", "⚡", "⭐", "💥", "✅", "🏆", "✋", "💖", "💝", "⚔️", "🥇", "💯", "💳", "🔞", "🥳", "😱"
]
# Настройка логирования
log_level = os.getenv('LOG_LEVEL', 'DEBUG').upper() # Считываем уровень логирования из .env (по умолчанию DEBUG)
logging.basicConfig(level=log_level, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def preprocess_spam_keywords(keywords):
"""Приводит ключевые слова к нормализованной форме."""
logger.debug("preprocess_spam_keywords: keyword conversion")
return set(' '.join(normalize_text_to_infinitive(keyword)) for keyword in keywords)
# Загрузка переменных из файла .env
......@@ -102,22 +110,31 @@ dp.include_router(router)
active_users = {}
morph = MorphAnalyzer()
def has_mixed_layout(text):
"""
Проверяет, содержит ли текст слова, написанные на смешанной раскладке.
Например: "пр1в3т", "teст", "нeт".
"""
cyrillic_letters = "абвгдеёжзийклмнопрстуфхцчшщъыьэюя"
latin_letters = "abcdefghijklmnopqrstuvwxyz"
try:
words = text.split()
for word in words:
has_cyrillic = any(char in cyrillic_letters for char in word.lower())
has_latin = any(char in latin_letters for char in word.lower())
if has_cyrillic and has_latin:
return True # Найдено слово со смешанной раскладкой
return False
cyrillic_letters = "абвгдеёжзийклмнопрстуфхцчшщъыьэюя"
latin_letters = "abcdefghijklmnopqrstuvwxyz"
words = text.split()
for word in words:
has_cyrillic = any(char in cyrillic_letters for char in word.lower())
has_latin = any(char in latin_letters for char in word.lower())
if has_cyrillic and has_latin:
logger.info("Word with mixed layout: %s", word)
return True # Найдено слово со смешанной раскладкой
logger.info("No word with mixed layout.")
return False
except Exception as e:
logger.error("Checking error: %s", e)
return False
# Регулярное выражение для распознавания эмодзи
emoji_pattern = re.compile("["
......@@ -141,6 +158,7 @@ def normalize_text_to_infinitive(text):
"""
if not text: # Проверяем, что текст не None или пустой
logger.warning("No text for normalisation")
return []
# Приведение текста к единой форме Unicode (NFKC)
......@@ -166,17 +184,21 @@ def normalize_text_to_infinitive(text):
normalized_words.append(morph.parse(word)[0].normal_form)
except Exception as e:
# Если слово не удалось обработать, добавляем его как есть
logger.error("Normalisation error '%s': %s", word, e)
normalized_words.append(word)
return normalized_words
def extract_emojis(text):
#return text.join(c for c in text if unicodedata.category(c).startswith('So'))
return ''.join(c for c in text if emoji.is_emoji(c))
try:
return ''.join(c for c in text if emoji.is_emoji(c))
except Exception as e:
logger.error("Extracting emojis error: %s", e)
return ''
def is_spam(text):
if text is None: # Проверяем, что текст не None
logger.warning("No text")
return False
# Нормализуем текст
......@@ -208,6 +230,8 @@ def is_spam(text):
spam_count += 1
matched_keywords.append(keyword) # Добавляем совпавшее слово в список
logger.debug(f"Spam-words found in text: {', '.join(matched_keywords)}")
spam_count += emoji_count # Учитываем только спам-эмодзи
# Возвращаем True, если найдено два или более совпадений
......@@ -230,6 +254,7 @@ async def welcome_new_user(event: ChatMemberUpdated):
try:
# Отправляем приветственное сообщение
logger.info(f"Sending welcome message to {new_user.first_name} ({new_user.id}) in chat {event.chat.id}")
welcome_message = await bot.send_message(
chat_id=event.chat.id,
text=f"Здравствуйте, {new_user.first_name}! Добро пожаловать в нашу группу!\n\n"
......@@ -246,9 +271,10 @@ async def welcome_new_user(event: ChatMemberUpdated):
"user_answered": False,
"chat_id": event.chat.id,
}
logger.info(f"User {new_user.first_name} ({new_user.id}) added to active users.")
except Exception as e:
print(f"Failed sending a message to the group {event.chat.id}. Ошибка: {e}")
logger.error(f"Failed sending a message to the group {event.chat.id}. Error: {e}")
# Ожидание ответа пользователя
await check_answer(new_user.id)
......@@ -265,10 +291,11 @@ async def check_answer(user_id):
# Удаляем пользователя из группы
await bot.ban_chat_member(chat_id=user_data["chat_id"], user_id=user_id)
logger.info(f"User {user_id} has been banned due to not answering in time.")
except AiogramError as e:
print(f"User deletion error {user_id}. Error: {e}")
logger.error(f"User deletion error {user_id}. Error: {e}")
except Exception as e:
print(f"User deletion error {user_id}. Error: {e}")
logger.error(f"User deletion error {user_id}. Error: {e}")
async def cleanup_messages(user_data):
"""Удаление всех сообщений пользователя и бота."""
......@@ -278,7 +305,7 @@ async def cleanup_messages(user_data):
try:
await bot.delete_message(chat_id=user_data["chat_id"], message_id=message_id)
except Exception as e:
print(f"Bot message deletion error. Error: {e}")
logger.error(f"Bot message deletion error. Error: {e}")
# Удаляем сообщения пользователя
for user_message in user_data["user_messages"]:
......@@ -286,9 +313,9 @@ async def cleanup_messages(user_data):
if user_message:
await bot.delete_message(chat_id=user_data["chat_id"], message_id=user_message.message_id)
except Exception as e:
print(f"User message deletion error. Error: {e}")
logger.error(f"User message deletion error. Error: {e}")
except Exception as e:
print(f"Message deletion error. Error: {e}")
logger.error(f"Message deletion error. Error: {e}")
@router.message()
async def handle_message(message: Message):
......@@ -302,7 +329,7 @@ async def handle_message(message: Message):
await message.delete()
return
except AiogramError as e:
print(f"User status verification error {message.from_user.id}. Error: {e}")
logger.error(f"User status verification error {message.from_user.id}. Error: {e}")
text = message.text or message.caption or ""
text = " ".join(normalize_text_to_infinitive(text)) # Убедимся, что это строка
......@@ -313,7 +340,7 @@ async def handle_message(message: Message):
await message.delete()
return
except Exception as e:
print(f"Error deleting a message with a mixed layout. Error: {e}")
logger.error(f"Error deleting a message with a mixed layout. Error: {e}")
# Проверка на мультимедиа (фото/видео) и его подпись
if (message.photo or message.video or message.document) and text.strip() and is_spam(text):
......@@ -321,7 +348,7 @@ async def handle_message(message: Message):
await message.delete()
return
except Exception as e:
print(f"Error deleting a multimedia message. Error: {e}")
logger.error(f"Error deleting a multimedia message. Error: {e}")
# Проверка на кнопки/ссылки
if message.reply_markup and hasattr(message.reply_markup, 'inline_keyboard'):
......@@ -331,7 +358,7 @@ async def handle_message(message: Message):
await message.delete()
return
except Exception as e:
print(f"Error when deleting a message with a button. Error: {e}")
logger.error(f"Error when deleting a message with a button. Error: {e}")
# Проверка на спам
if is_spam(message.text):
......@@ -339,7 +366,7 @@ async def handle_message(message: Message):
await message.delete()
return
except Exception as e:
print(f"Error when trying to delete spam messages. Error: {e}")
logger.error(f"Error when trying to delete spam messages. Error: {e}")
user_data = active_users.get(message.from_user.id)
......@@ -356,18 +383,19 @@ async def handle_message(message: Message):
user_data["message_ids"].append(correct_message.message_id) # Добавляем в список ID сообщений
# Удаляем сообщения после правильного ответа
await cleanup_messages(user_data)
logger.info(f"User {message.from_user.id} answered correctly with {user_answer}. Correct answer.")
else:
# Ответ неправильный, добавляем новое сообщение бота
incorrect_message = await message.reply("Неправильный ответ. Попробуйте еще раз.")
user_data["message_ids"].append(incorrect_message.message_id)
logger.warning(f"User {message.from_user.id} answered incorrectly with {user_answer}. Correct answer was {user_data['correct_answer']}.")
else:
# Если ответ не число
invalid_message = await message.reply("Пожалуйста, отправьте числовой ответ.")
user_data["message_ids"].append(invalid_message.message_id)
logger.warning(f"User {message.from_user.id} sent an invalid answer: {message.text}")
from aiogram import types
# Обработчик сообщений
@dp.message(lambda message: message.reply_to_message and '/spam' in message.text.lower())
async def delete_spam(message: types.Message):
......@@ -377,14 +405,15 @@ async def delete_spam(message: types.Message):
# Удаляем сообщение, на которое был дан ответ
await message.reply_to_message.delete()
except Exception as e:
print(f"Deleting error: {e}")
logger.error("Deleting error: %s", e)
async def main():
"""Запуск бота."""
print("bot is running")
logger.info("bot is running")
await dp.start_polling(bot)
if __name__ == "__main__":
SPAM_KEYWORDS = preprocess_spam_keywords(SPAM_KEYWORDS)
asyncio.run(main())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment