AIOgram3 19. Капча для вступающих в чат

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

В начале июля мы проводили конкурс, в котором была задача разработать этот модуль для бота в нашем чате. Ознакомиться с вариантами реализации от конкурсантов можно на нашем GitHub.

В этом посте мы разберем нашу первоначальную реализацию капчи для бота со следующим функционалом:

• Генерация изображения с математическим выражением.
• Обработка ответов пользователя с тремя попытками на ввод верного ответа.
• Таймер для принудительного исключения пользователя, если не было ответа или попытки исчерпаны за указанное время.

Установка необходимых библиотек.

В реализации будут использоваться две дополнительные библиотеки:

1. pillow — одна из самых известных Python-библиотек для работы с изображениями. Именно она будет создавать изображение капчи.
2. apscheduler — мощная библиотека для создания планировщиков задач, выполняемых по расписанию.

Установим их:

pip install apscheduler pillow

Не забывайте добавить их в requirements.txt!

Функция генерации изображения.

Нам нужна функция, которая будет генерировать изображение с текстом капчи.

В пакете utils создадим файл captcha_utils.py. В нем создадим асинхронную функцию generate_image, принимающую строку с выражением и возвращающую изображение в виде байтового представления.

Код функции:

from io import BytesIO  

from PIL import Image, ImageDraw, ImageFont  


async def generate_image(expression: str) -> BytesIO:  
    image = Image.new("RGB", (800, 600), "white")  
    draw = ImageDraw.Draw(image)  
    text = expression + " = ?"  
    font_size = 64  
    try:  
        font = ImageFont.truetype("pressanybutton_bot/res/Gilroy-Medium.ttf", font_size)  
    except OSError:  
        font = ImageFont.load_default()  
    text_box = draw.textbbox((0, 0), text, font=font)  
    text_width = text_box[2] - text_box[0]  
    text_height = text_box[3] - text_box[1]  
    text_x = (800 - text_width) // 2  
    text_y = (600 - text_height) // 2  
    draw.text(xy=(text_x, text_y), text=text, font=font, fill="black")  

    captcha_result = BytesIO()  
    image.save(captcha_result, format="png")  
    captcha_result.seek(0)  

    return captcha_result

Разберём код:

Сначала в переменной image создается новое изображение размером 800x600 пикселей с белым фоном в цветовой модели RGB. В переменной draw инициализируется объект для рисования на основе созданного ранее изображения.

Далее в переменной text добавляем к выражению строку с вопросом, а в переменной font_size указываем размер шрифта.

В блоке try-except пытаемся подключить собственный шрифт. Если он не найден, используется шрифт по умолчанию.

В переменной text_box создается текстовое поле с указанным текстом и шрифтом, а в text_width и text_height вычисляется высота и ширина текста.

Затем в переменных text_x и text_y вычисляем координаты для размещения текста по центру и рисуем текст на изображении.

После того, как изображение готово, создается переменная captcha_result, представляющая собой объект в памяти, в который будет записано изображение без сохранения его локально в файл. Этот способ был подсмотрен в решении участника конкурса — Виктора Королева.

Сохраняем изображение в памяти и сбрасываем указатель позиции в буфере на начало.

Возвращаем полученный объект изображения.

Функция бана пользователя.

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

Почему бан пользователя, если нам нужно его просто "кикнуть" из чата? Дело в том, что в Telegram Bot API больше нет возможности просто исключить пользователя — даже ручное исключение пользователя в чате банит его. Однако, можно указать время бана, например, одна минута или неделя.

В теле функции у объекта сообщения/события вызываем метод .bot, предоставляющий доступ к экземпляру запущенного бота, а у него метод .ban_chat_member() для непосредственного бана. В метод передаем три аргумента:

1. chat_id — идентификатор чата, в котором необходимо забанить пользователя. В нашем случае это идентификатор группы.
2. user_id — идентификатор пользователя, которого следует забанить.
3. until_date — срок бана в виде даты и времени.

Затем в блоке if проверяем, передан ли в функцию аргумент state, и если он передан, вызываем сброс состояния.

Код функции:

async def ban_user(message: ChatMemberUpdated | Message, state: FSMContext = None) -> None:  
    await message.bot.ban_chat_member(  
        chat_id=secrets.group_id,  
        user_id=message.from_user.id,  
        until_date=timedelta(seconds=35),  
    )  

    if state:  
        await state.clear()

Экземпляр класса планировщика.

Для того, чтобы реализовать таймер бана пользователя, нам нужно определить экземпляр класса планировщика. Откроем файл settings.py. В самом низу создадим переменную scheduler и присвоим ей экземпляр класса AsyncIOScheduler, в который передадим один аргумент timezone с часовым поясом.

scheduler = AsyncIOScheduler(timezone="Europe/Moscow")

Рекомендую изучить эту библиотеку, поскольку она весьма функциональна и может применяться в разных проектах.

Класс состояния для обработчика.

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

Откроем файл statesform.py в пакете utils и добавим новый класс CheckJoin, унаследованный от StatesGroup. В нем пропишем одно состояние WAIT_ANSWER — объект класса State.

class CheckJoin(StatesGroup):  
    WAIT_ANSWER = State()

Обработчик вступления в чат.

Мы уже писали обработчик в посте "AIOgram3 15. Обработка события вступления или покидания чата", но там было просто приветствие пользователя. Теперь нам необходимо изменить функцию так, чтобы она при вступлении нового пользователя выдавала ему изображение капчи с сообщением.

Откроем файл events.py в пакете handlers и найдем функцию on_user_join. Удалим все, что в ней есть.

В самом начале создаем переменную ban_job, в ней вызываем метод .add_job() у экземпляра планировщика. В метод передаем три аргумента:

1. func — объект функции бана пользователя. Обратите внимание, нужна именно ссылка на функцию, а не ее вызов, то есть без скобок ()!
2. trigger — apcheduler позволяет использовать разные триггеры, такие как date, interval и cron. В нашем случае необходим date-триггер, который будет срабатывать в указанное время.
3. kwargs — словарь с ключевыми аргументами для передачи в вызываемую функцию.

После создания задачи в переменную ban_job вернется объект задачи, из которого нам нужен будет job_id.

ban_job = scheduler.add_job(  
    func=ban_user,  
    trigger=DateTrigger(run_date=datetime.now() + timedelta(minutes=1)),  
    kwargs={"message": event, "state": state},  
)

После пятисекундного ожидания (чтобы сообщение с капчей не вываливалось сразу, как пользователь вступил в чат) идет ряд переменных:

• first_name — получаем имя пользователя.
• numbers — генерируем числа от 1 до 10 (можно указать любой диапазон).
• expression — создаем строку с выражением.
• image — получаем сгенерированное изображение из ранее написанной функции.

await sleep(5)  
first_name = event.new_chat_member.user.first_name  
numbers = range(1, 10)  
expression = f"{choice(numbers)} + {choice(numbers)}"  
image = await generate_image(expression=expression)

Далее обращаемся к переменной event, вызываем метод .answer_photo(), передав следующие аргументы:

• photo — экземпляр класса BufferedInputFile, в который передаем изображение из переменной image с вызванным методом .read() для получения массива байт и ключевой аргумент filename.
• caption — это подпись к фото. В него передаем сообщение пользователю.
• show_caption_above_media — интересный аргумент, принимающий True или False, помещающий изображение под текст, а не как оно по умолчанию — над ним.

Затем в данные состояния передаем словарь с выражением, количеством попыток (в самом начале 0) и идентификатором задачи в планировщике. Меняем состояние на ожидание ответа.

Полный код обработчика:

async def on_user_join(event: ChatMemberUpdated, state: FSMContext) -> None:  
    ban_job = scheduler.add_job(  
        func=ban_user,  
        trigger=DateTrigger(run_date=datetime.now() + timedelta(minutes=1)),  
        kwargs={"message": event, "state": state},  
    )  
    await sleep(5)  
    first_name = event.new_chat_member.user.first_name  
    numbers = range(1, 10)  
    expression = f"{choice(numbers)} + {choice(numbers)}"  
    image = await generate_image(expression=expression)  
    await event.answer_photo(  
        photo=BufferedInputFile(image.read(), filename="captcha"),  
        caption=views.pre_join_message(first_name=first_name),  
        show_caption_above_media=True,  
    )  
    await state.set_data(data={"expression": expression, "tries": 0, "ban_job": ban_job.id})  
    await state.set_state(state=CheckJoin.WAIT_ANSWER)

Обработчик ответов пользователя.

Ниже on_user_join создадим новую асинхронную функцию wait_join_answer, принимающую message и state.

В самом начале в переменную data получаем данные состояния, а в переменную tries — количество попыток из этих данных.

В блоке if проверяем, равно ли количество попыток трем. Если не равно, то пропускаем этот блок, а если равно, то удаляем последнее сообщение от пользователя (вдруг там вместо ответа спам?) и вызываем функцию бана пользователя. Для того чтобы код ниже не выполнялся, прописываем пустой return.

data = await state.get_data()  
tries = data.get("tries")  

if tries  3:  
    await message.delete()  
    await ban_user(message=message)  
    return

Далее еще один блок if, в котором проверяем, содержит ли сообщение текст и состоит ли текст сообщения только из чисел. Внутри блока еще один if, в котором сравниваем полученное от пользователя число с результатом выполнения нашего выражения. Если пользователь ответил верно, то мы отвечаем ему приветственным сообщением, вызываем метод удаления задачи из планировщика и очищаем состояние.

if message.text and message.text.isdigit():  
    if int(message.text)  eval(data.get("expression")):  
        await message.answer(text=views.join_message(first_name=message.from_user.first_name))  
        scheduler.remove_job(job_id=data.get("ban_job"))  
        await state.clear()  
        return

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

Полный код функции:

async def wait_join_answer(message: Message, state: FSMContext) -> None:  
    data = await state.get_data()  
    tries = data.get("tries")  

    if tries  3:  
        await message.delete()  
        await ban_user(message=message)  
        return  

    if message.text and message.text.isdigit():  
        if int(message.text)  eval(data.get("expression")):  
            await message.answer(text=views.join_message(first_name=message.from_user.first_name))  
            scheduler.remove_job(job_id=data.get("ban_job"))  
            await state.clear()  
            return  

    await message.delete()  
    await message.answer(text=views.wrong_answer_join_message(tries=tries + 1))  
    await state.update_data(tries=tries + 1)  
    await state.set_state(state=CheckJoin.WAIT_ANSWER)

Заключение.

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