Бот для пользователей у которых спам-блок | Bot for users who have a spam block

Приветствую всех, представляю вашему вниманию бота, который будет перенаправлять сообщения от юзеров со спам-блоком, прямиком к вам в чат с ним!​

Что он умеет?

Поддержка типов контента: текст, стикеры, фото, видео, *********, голосовые сообщения, медиа-альбомы.

Высокая производительность и отзывчивость.

Минимальные зависимости и настройки.

История всех обращений с метаданными записываются в БД SQLite.

Важно установить все библиотеки, для просмотра файлов типа SQLite, вам требуется установить SQLiteStudio.

Количество строк кода - 169.

Что требуется для запуска Бота?

В созданную папку с ботом, создайте текстовый файл requirements.txt, для установки зависимостей.

Установите зависимости - pip install -r requirements.txt

Добавьте токены в соответствующие поля.

Запустите бота - python main.py

Будут вопросы, пишите в лс форума или под темой - помогу.

import logging
import sqlite3
from datetime import datetime
from aiogram import Bot, Dispatcher, types, F
from aiogram.enums import ContentType
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
from aiogram.types import Message

TOKEN = "ТОКЕ СЮДА" # А это нужно чтобы бот работал)
ADMIN_ID = "А СЮДА ТОКЕН АДМИНА" # Токен админа требуется для перенаправления сообщения

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)

bot = Bot(token=TOKEN)
storage = MemoryStorage()
dp = Dispatcher(storage=storage)

class Form(StatesGroup):
waiting_message = State()

def init_db():
with sqlite3.connect('spam_block.db') as conn:
conn.execute('''CREATE TABLE IF NOT EXISTS requests
(id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER,
username TEXT,
message TEXT,
media_type TEXT,
media_id TEXT,
timestamp DATETIME)''')

def save_request(user_id: int, username: str, message: str,
media_type: str | None = None, media_id: str | None = None):
with sqlite3.connect('spam_block.db') as conn:
conn.execute('''INSERT INTO requests
(user_id, username, message, media_type, media_id, timestamp)
VALUES (?, ?, ?, ?, ?, ?)''',
(user_id, username, message, media_type, media_id, datetime.now()))

@dp.message(Command('start'))
async def start(message: Message, state: FSMContext):
await state.set_state(Form.waiting_message)
await message.answer(
f"Привет, {message.from_user.first_name}!\n"
"Отправь сообщение или стикер администратору.\n"
"Можешь отправлять несколько сообщений подряд.\n"
"Для завершения используй /cancel",
reply_markup=types.ReplyKeyboardRemove()
)

@dp.message(Command('cancel'))
@dp.message(F.text.casefold() == "cancel")
async def cancel(message: Message, state: FSMContext):
await state.clear()
await message.answer(
"Действие отменено. Чтобы начать заново, используй /start",
reply_markup=types.ReplyKeyboardRemove()
)

async def forward_to_admin(content: dict, user: types.User, caption: str):
admin_text = (
f"Новое сообщение от пользователя:\n"
f"ID: {user.id}\n"
f"Username: @{user.username}\n"
f"Имя: {user.full_name}\n"
f"Сообщение:\n{caption}"
)

try:
if content['method'] == bot.send_sticker:
await bot.send_sticker(ADMIN_ID, sticker=content['kwargs']['sticker'])
await bot.send_message(ADMIN_ID, admin_text)
elif 'media' in content:
await bot.send_media_group(ADMIN_ID, [content['media']])
await bot.send_message(ADMIN_ID, admin_text)
else:
method = content['method']
if method == bot.send_message:
await method(ADMIN_ID, text=admin_text)
else:
await method(ADMIN_ID, **content['kwargs'], caption=admin_text)
except Exception as e:
logger.error(f"Error sending to admin: {e}")

@dp.message(Form.waiting_message)
async def handle_message(message: Message, state: FSMContext):
try:
user = message.from_user
content = {'method': bot.send_message, 'kwargs': {}}
media_type = None
media_id = None
caption = message.caption or message.text or "Медиа-сообщение"

if message.sticker:
media_type = 'sticker'
media_id = message.sticker.file_id
content.update({
'method': bot.send_sticker,
'kwargs': {'sticker': media_id}
})
caption = "Стикер"
elif message.photo:
media_type = 'photo'
media_id = message.photo[-1].file_id
content.update({
'method': bot.send_photo,
'kwargs': {'photo': media_id}
})
elif message.video:
media_type = 'video'
media_id = message.video.file_id
content.update({
'method': bot.send_video,
'kwargs': {'video': media_id}
})
elif message.document:
media_type = 'document'
media_id = message.document.file_id
content.update({
'method': bot.send_document,
'kwargs': {'document': media_id}
})
elif message.audio:
media_type = 'audio'
media_id = message.audio.file_id
content.update({
'method': bot.send_audio,
'kwargs': {'audio': media_id}
})
elif message.voice:
media_type = 'voice'
media_id = message.voice.file_id
content.update({
'method': bot.send_voice,
'kwargs': {'voice': media_id}
})
elif message.media_group_id:
media_type = 'media_group'
content.update({
'media': message.media_group_id,
'method': bot.send_media_group
})

save_request(
user_id=user.id,
username=user.username or user.full_name,
message=caption,
media_type=media_type,
media_id=media_id
)

await forward_to_admin(content, user, caption)
await message.answer("✅ Сообщение доставлено! Можете отправить следующее.")

except Exception as e:
logger.error(f"Error handling message: {e}")
await message.answer("Произошла ошибка при обработке сообщения")
await state.clear()

if __name__ == '__main__':
init_db()
dp.run_polling(bot)

Python

import logging

import sqlite3

from datetime import datetime

from aiogram import Bot, Dispatcher, types, F

from aiogram.enums import ContentType

from aiogram.filters import Command

from aiogram.fsm.context import FSMContext

from aiogram.fsm.state import State, StatesGroup

from aiogram.fsm.storage.memory import MemoryStorage

from aiogram.types import Message



TOKEN = "ТОКЕ СЮДА" # А это нужно чтобы бот работал)

ADMIN_ID = "А СЮДА ТОКЕН АДМИНА" # Токен админа требуется для перенаправления сообщения



logging.basicConfig(

    level=logging.INFO,

    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",

)

logger = logging.getLogger(__name__)



bot = Bot(token=TOKEN)

storage = MemoryStorage()

dp = Dispatcher(storage=storage)



class Form(StatesGroup):

    waiting_message = State()



def init_db():

    with sqlite3.connect('spam_block.db') as conn:

        conn.execute('''CREATE TABLE IF NOT EXISTS requests

                     (id INTEGER PRIMARY KEY AUTOINCREMENT,

                      user_id INTEGER,

                      username TEXT,

                      message TEXT,

                      media_type TEXT,

                      media_id TEXT,

                      timestamp DATETIME)''')



def save_request(user_id: int, username: str, message: str,

                media_type: str | None = None, media_id: str | None = None):

    with sqlite3.connect('spam_block.db') as conn:

        conn.execute('''INSERT INTO requests

                     (user_id, username, message, media_type, media_id, timestamp)

                     VALUES (?, ?, ?, ?, ?, ?)''',

                  (user_id, username, message, media_type, media_id, datetime.now()))



@dp.message(Command('start'))

async def start(message: Message, state: FSMContext):

    await state.set_state(Form.waiting_message)

    await message.answer(

        f"Привет, {message.from_user.first_name}!\n"

        "Отправь сообщение или стикер администратору.\n"

        "Можешь отправлять несколько сообщений подряд.\n"

        "Для завершения используй /cancel",

        reply_markup=types.ReplyKeyboardRemove()

    )



@dp.message(Command('cancel'))

@dp.message(F.text.casefold() == "cancel")

async def cancel(message: Message, state: FSMContext):

    await state.clear()

    await message.answer(

        "Действие отменено. Чтобы начать заново, используй /start",

        reply_markup=types.ReplyKeyboardRemove()

    )



async def forward_to_admin(content: dict, user: types.User, caption: str):

    admin_text = (

        f"Новое сообщение от пользователя:\n"

        f"ID: {user.id}\n"

        f"Username: @{user.username}\n"

        f"Имя: {user.full_name}\n"

        f"Сообщение:\n{caption}"

    )

    

    try:

        if content['method'] == bot.send_sticker:

            await bot.send_sticker(ADMIN_ID, sticker=content['kwargs']['sticker'])

            await bot.send_message(ADMIN_ID, admin_text)

        elif 'media' in content:

            await bot.send_media_group(ADMIN_ID, [content['media']])

            await bot.send_message(ADMIN_ID, admin_text)

        else:

            method = content['method']

            if method == bot.send_message:

                await method(ADMIN_ID, text=admin_text)

            else:

                await method(ADMIN_ID, **content['kwargs'], caption=admin_text)

    except Exception as e:

        logger.error(f"Error sending to admin: {e}")



@dp.message(Form.waiting_message)

async def handle_message(message: Message, state: FSMContext):

    try:

        user = message.from_user

        content = {'method': bot.send_message, 'kwargs': {}}

        media_type = None

        media_id = None

        caption = message.caption or message.text or "Медиа-сообщение"



        if message.sticker:

            media_type = 'sticker'

            media_id = message.sticker.file_id

            content.update({

                'method': bot.send_sticker,

                'kwargs': {'sticker': media_id}

            })

            caption = "Стикер"

        elif message.photo:

            media_type = 'photo'

            media_id = message.photo[-1].file_id

            content.update({

                'method': bot.send_photo,

                'kwargs': {'photo': media_id}

            })

        elif message.video:

            media_type = 'video'

            media_id = message.video.file_id

            content.update({

                'method': bot.send_video,

                'kwargs': {'video': media_id}

            })

        elif message.document:

            media_type = 'document'

            media_id = message.document.file_id

            content.update({

                'method': bot.send_document,

                'kwargs': {'document': media_id}

            })

        elif message.audio:

            media_type = 'audio'

            media_id = message.audio.file_id

            content.update({

                'method': bot.send_audio,

                'kwargs': {'audio': media_id}

            })

        elif message.voice:

            media_type = 'voice'

            media_id = message.voice.file_id

            content.update({

                'method': bot.send_voice,

                'kwargs': {'voice': media_id}

            })

        elif message.media_group_id:

            media_type = 'media_group'

            content.update({

                'media': message.media_group_id,

                'method': bot.send_media_group

            })



        save_request(

            user_id=user.id,

            username=user.username or user.full_name,

            message=caption,

            media_type=media_type,

            media_id=media_id

        )



        await forward_to_admin(content, user, caption)

        await message.answer("✅ Сообщение доставлено! Можете отправить следующее.")



    except Exception as e:

        logger.error(f"Error handling message: {e}")

        await message.answer("Произошла ошибка при обработке сообщения")

        await state.clear()



if __name__ == '__main__':

    init_db()

    dp.run_polling(bot)

requirements.txt

aiogram==3.3.0
python-dotenv==1.0.0

P,S

В данной ветке я не нашел такого бота. Буду рад получить ОС, от вас.