Бот для защиты от спама по сигнатуре

В марте 2024 года мы уже публиковали бот для защиты от спама, суть которого сводилась к автоматическому удалению, удалению и блоку или блоку без удаления всех тех, кого нет в вашем списке контактов. И хотя принцип его работы все еще не утратил актуальности и, по мнению автора, является оптимальным решением, для многих представителей сферы он не является удобным, так как специфика их работы подразумевает постоянные переписки с теми, кого нет в контактах.

Спам между тем никуда не девается — и даже платные функции Telegram Premium от него не спасают, ведь спамеры давно внесли премак в список расходников. Решить эту проблему призван новый бот, принцип которого основан на анализе сигнатур. О том, как его создать, — рассказываем ниже.

Какие задачи решает бот для защиты от спама по сигнатуре

Данный бот закрывает следующие задачи:

• Защита от спама без ущерба коммуникации.

• Определение спама по вручную заданной или автоматически идентифицированной сигнатуре.

• Автоматическое удаление спам-сообщений и блокировка их отправителей.

• Предупреждение повторного спама за счет заранее заданных сигнатур.

Кроме того, бот может стать базой для более сложной системы защиты, основанной как на примитивном fuzzy-сравнении, так и на интеллектуальном анализе. Но в рамках данной рубрики это было бы слишком сложным, посему такие изощрения оставим на откуп искушенной части читателей.

Принцип работы бота для защиты от спама по сигнатуре

Алгоритмически бот довольно прост, что, впрочем, не умаляет его полезности, делает он следующее:

• Слушает эфир и реагирует на события.

• Сравнивает входящие сообщения с базой сигнатур для выявления спама.

• Реагирует на команды в избранном.

• Выводит список команд при вводе /help.

• Выводит список сигнатур при вводе /list.

• Добавляет сигнатуру при вводе /add <сигнатура>.

• Удаляет сигнатуру при вводе /del <сигнатура>.

• Удаляет все сигнатуры при вводе /clear.

• Добавляет юзера в белый список при вводе /white <ID>.

• Удаляет юзера из белого списка при вводе /unwhite <ID>.

• Выводит белый список при вводе /whitelist.

• Задает число повторов для автоматического добавления в список сигнатур при вводе /set <число>.

• Переключает реакцию бота в режим «только удаление» при вводе /mode delete — только удалять спам.

• Переключает реакцию бота в режим «только удаление и ЧС» при вводе/mode delete_ban.

Как видите — все просто. Но при этом достаточно надежно защищает от большей части спама.

Пошаговая инструкция, как сделать бот для защиты от спама по сигнатуре

Традиционно, сначала нужно развернуть сервер. Подразумевается, что вы это уже сделали. Однако если нет и не знаете, как это делать, — держите гайд, как развернуть сервер на домашнем ПК. Вообще, домашний ПК — не лучшее решение, но для тестов он идеален. Впоследствии же советуем вам арендовать себе полноценный VPS. Иначе бот банально не будет работать, когда ваш ПК выключен.

После развертывания сервера нужно его настроить, для этого пишем в консоль:

pip install telethon

Затем нам нужно получить токены. Это делается так:

1. Открываем Telegram Web.
2. Находим пункт «API development tools».
3. Жмем «Создать приложение» и копируем сгенерированные api_id и api_hash.
4. Сохраняем их.

Далее создаем наш бот:

1. В текстовом редакторе создаем файл. Переименуем его в bot.py и вставляем код:

import asyncio

import json

import os

import re

from urllib.parse import urlsplit, urlunsplit

from telethon import TelegramClient, events, functions

from telethon.tl.functions.contacts import BlockRequest

API_ID = ваш API_ID

API_HASH = "ваш API_HASH"

SESSION_NAME = "antispam_userbot"

SIGNATURES_FILE = "signatures.json"

WHITELIST_FILE = "whitelist.json"

SETTINGS_FILE = "settings.json"

DEFAULT_REPEAT_THRESHOLD = 3

DEFAULT_ACTION = "delete"

client = TelegramClient(

SESSION_NAME,

API_ID,

API_HASH

)

def load_json(filename, default):

if not os.path.exists(filename):

save_json(filename, default)

return default

try:

with open(filename, "r", encoding="utf-8") as f:

return json.load(f)

except Exception:

return default

def save_json(filename, data):

with open(filename, "w", encoding="utf-8") as f:

json.dump(

data,

f,

ensure_ascii=False,

indent=2

)

signatures = load_json(

SIGNATURES_FILE,

[]

)

whitelist = load_json(

WHITELIST_FILE,

[]

)

settings = load_json(

SETTINGS_FILE,

{

"repeat_threshold": DEFAULT_REPEAT_THRESHOLD,

"action": DEFAULT_ACTION

}

)

repeat_tracker = {}

def normalize_url(url):

url = url.strip()

try:

parts = urlsplit(url)

scheme = parts.scheme.lower()

netloc = parts.netloc.lower()

path = parts.path.rstrip("/")

return urlunsplit((

scheme,

netloc,

path,

parts.query,

parts.fragment

))

except Exception:

return url.lower().rstrip("/")

def extract_urls(text):

if not text:

return []

pattern = re.compile(

r'https?://[^\s<>"\'\]\)]+',

re.IGNORECASE

)

urls = pattern.findall(text)

result = []

for url in urls:

url = url.rstrip(".,!?;:")

normalized = normalize_url(url)

if normalized:

result.append(normalized)

return result

def normalize_text(text):

if not text:

return ""

text = text.replace("\r", " ")

text = text.replace("\n", " ")

text = re.sub(r"\s+", " ", text)

return text.strip().casefold()

def is_signature(signature):

return signature in signatures

def add_signature(signature):

if signature not in signatures:

signatures.append(signature)

save_json(

SIGNATURES_FILE,

signatures

)

return True

return False

def remove_signature(signature):

if signature in signatures:

signatures.remove(signature)

save_json(

SIGNATURES_FILE,

signatures

)

return True

return False

def is_whitelisted(user_id):

return str(user_id) in whitelist

def add_whitelist(user_id):

user_id = str(user_id)

if user_id not in whitelist:

whitelist.append(user_id)

save_json(

WHITELIST_FILE,

whitelist

)

return True

return False

def remove_whitelist(user_id):

user_id = str(user_id)

if user_id in whitelist:

whitelist.remove(user_id)

save_json(

WHITELIST_FILE,

whitelist

)

return True

return False

async def notify(text):

await client.send_message(

"me",

text

)

async def get_contacts_ids():

result = await client(

functions.contacts.GetContactsRequest(

hash=0

)

)

return {

str(user.id)

for user in result.users

}

async def block_user(user_id):

try:

await client(

BlockRequest(

id=user_id

)

)

return True

except Exception as e:

await notify(

"⚠️ Не удалось заблокировать пользователя.\n\n"

f"ID: {user_id}\n"

f"Ошибка: {e}"

)

return False

@client.on(events.NewMessage(outgoing=True))

async def commands_handler(event):

me = await client.get_me()

if event.chat_id != me.id:

return

text = (event.raw_text or "").strip()

if not text.startswith("/"):

return

parts = text.split(maxsplit=1)

command = parts[0].lower()

argument = ""

if len(parts) > 1:

argument = parts[1].strip()

if command == "/help":

await notify(

"🛡 АНТИСПАМ\n\n"

"/help — список команд\n"

"/list — список сигнатур\n"

"/add <сигнатура> — добавить текст или ссылку\n"

"/del <сигнатура> — удалить сигнатуру\n"

"/clear — удалить все сигнатуры\n\n"

"/white <ID> — добавить пользователя в whitelist\n"

"/unwhite <ID> — убрать пользователя из whitelist\n"

"/whitelist — показать whitelist\n\n"

"/set <число> — количество разных отправителей "

"для автодобавления сигнатуры\n"

"/mode delete — только удалять спам\n"

"/mode delete_ban — удалять и блокировать\n\n"

f"Текущий порог: "

f"{settings['repeat_threshold']}\n"

f"Текущий режим: "

f"{settings['action']}"

)

return

if command == "/list":

if not signatures:

await notify(

"📋 Сигнатур пока нет."

)

return

text_out = "📋 СИГНАТУРЫ:\n\n"

for i, signature in enumerate(

signatures,

1

):

text_out += (

f"{i}. {signature}\n"

)

await notify(text_out)

return

if command == "/add":

if not argument:

await notify(

"Использование:\n"

"/add <сигнатура>\n\n"

"Сигнатурой может быть текст или ссылка."

)

return

if re.match(

r"^https?://",

argument,

re.IGNORECASE

):

signature = normalize_url(argument)

else:

signature = normalize_text(argument)

if not signature:

await notify(

"❌ Пустая сигнатура."

)

return

if add_signature(signature):

await notify(

"✅ Сигнатура добавлена:\n\n"

f"{signature}"

)

else:

await notify(

"ℹ️ Такая сигнатура уже существует:\n\n"

f"{signature}"

)

return

if command == "/del":

if not argument:

await notify(

"Использование:\n"

"/del <сигнатура>"

)

return

if re.match(

r"^https?://",

argument,

re.IGNORECASE

):

signature = normalize_url(argument)

else:

signature = normalize_text(argument)

if remove_signature(signature):

repeat_tracker.pop(

signature,

None

)

await notify(

"🗑 Сигнатура удалена:\n\n"

f"{signature}"

)

else:

await notify(

"ℹ️ Такой сигнатуры нет:\n\n"

f"{signature}"

)

return

if command == "/clear":

signatures.clear()

repeat_tracker.clear()

save_json(

SIGNATURES_FILE,

signatures

)

await notify(

"🗑 Все сигнатуры удалены."

)

return

if command == "/white":

if not argument or not argument.isdigit():

await notify(

"Использование:\n"

"/white 123456789"

)

return

if add_whitelist(argument):

await notify(

f"✅ Пользователь {argument} "

"добавлен в whitelist."

)

else:

await notify(

f"ℹ️ Пользователь {argument} "

"уже находится в whitelist."

)

return

if command == "/unwhite":

if not argument or not argument.isdigit():

await notify(

"Использование:\n"

"/unwhite 123456789"

)

return

if remove_whitelist(argument):

await notify(

f"🗑 Пользователь {argument} "

"удалён из whitelist."

)

else:

await notify(

f"ℹ️ Пользователя {argument} "

"нет в whitelist."

)

return

if command == "/whitelist":

if not whitelist:

await notify(

"📋 Whitelist пуст."

)

return

text_out = "📋 WHITELIST:\n\n"

for i, user_id in enumerate(

whitelist,

1

):

text_out += (

f"{i}. {user_id}\n"

)

await notify(text_out)

return

if command == "/set":

if not argument.isdigit():

await notify(

"Использование:\n"

"/set 3"

)

return

value = int(argument)

if value < 1:

await notify(

"❌ Число должно быть не меньше 1."

)

return

settings["repeat_threshold"] = value

save_json(

SETTINGS_FILE,

settings

)

await notify(

"✅ Порог установлен: "

f"{value}\n\n"

"Одна и та же ссылка или текст "

f"от {value} разных отправителей "

"будет автоматически добавлена "

"в сигнатуры."

)

return

if command == "/mode":

if argument not in (

"delete",

"delete_ban"

):

await notify(

"Использование:\n\n"

"/mode delete — только удалять\n"

"/mode delete_ban — удалять + блокировать"

)

return

settings["action"] = argument

save_json(

SETTINGS_FILE,

settings

)

if argument == "delete":

description = (

"сообщения будут только удаляться"

)

else:

description = (

"сообщения будут удаляться, "

"а отправители блокироваться"

)

await notify(

"✅ Режим изменён.\n\n"

f"{description}."

)

return

@client.on(events.NewMessage(incoming=True))

async def incoming_handler(event):

if not event.is_private:

return

me = await client.get_me()

if event.chat_id == me.id:

return

sender = await event.get_sender()

if sender is None:

return

user_id = str(sender.id)

if is_whitelisted(user_id):

return

contact_ids = await get_contacts_ids()

if user_id in contact_ids:

return

text = event.raw_text or ""

if not text:

return

normalized_text = normalize_text(text)

urls = extract_urls(text)

matched_signature = None

for url in urls:

if is_signature(url):

matched_signature = url

break

if matched_signature is None:

for signature in signatures:

if re.match(

r"^https?://",

signature,

re.IGNORECASE

):

continue

if signature in normalized_text:

matched_signature = signature

break

if matched_signature:

try:

await event.delete()

except Exception as e:

await notify(

"⚠️ Не удалось удалить сообщение.\n\n"

f"Сигнатура: {matched_signature}\n"

f"Ошибка: {e}"

)

return

if settings["action"] == "delete_ban":

await block_user(

sender.id

)

await notify(

"🚫 СПАМ УДАЛЁН + ПОЛЬЗОВАТЕЛЬ "

"ЗАБЛОКИРОВАН\n\n"

f"Пользователь: {user_id}\n"

f"Сигнатура: {matched_signature}"

)

return

candidates = set()

for url in urls:

candidates.add(url)

if normalized_text:

candidates.add(

normalized_text

)

for candidate in candidates:

if is_signature(candidate):

continue

users = repeat_tracker.setdefault(

candidate,

set()

)

users.add(user_id)

threshold = settings[

"repeat_threshold"

]

if len(users) >= threshold:

added = add_signature(

candidate

)

if added:

repeat_count = len(users)

repeat_tracker.pop(

candidate,

None

)

try:

await event.delete()

except Exception:

pass

if (

candidate.startswith("http://")

or candidate.startswith("https://")

):

signature_type = "ссылка"

else:

signature_type = "текст"

await notify(

"🚨 АВТОМАТИЧЕСКИ ОБНАРУЖЕН СПАМ\n\n"

f"Тип: {signature_type}\n\n"

f"Сигнатура:\n"

f"{candidate}\n\n"

f"Разных отправителей: "

f"{repeat_count}\n"

f"Порог: {threshold}\n\n"

"Сигнатура добавлена автоматически."

)

async def main():

await client.start()

me = await client.get_me()

await notify(

"🛡 Антиспам запущен.\n\n"

"Команды доступны только в Избранном.\n"

"Мониторятся только входящие ЛС."

)

await client.run_until_disconnected()

if __name__ == "__main__":

asyncio.run(main())

2. Находим строки API_ID = ваш API_ID и API_HASH = "ваш API_HASH" и подставляем свои значения.

3. Сохраняем текстовый файл.

4. Запускаем бот командой:

python bot.py

5. Проверяем, все ли корректно работает.

Демонстрация работы

Подводя итоги

Как видите, защититься от спама не так уж и сложно. И для этого даже не обязательно игнорировать всех тех, кого нет в контактах. Достаточно просто запустить бот для защиты от спама по сигнатуре, создать который своими руками теперь можете и вы.