Telegram bot that sends reCAPTCHA styled image when someone joins a chat
import asyncio
from pathlib import Path
import os
import logging

from aiogram import Bot, Dispatcher

from captcha_bot.config import parse_config
from captcha_bot.handlers import member, admin
from captcha_bot.image import ImageMaker


async def main():
    logging.basicConfig(format='%(levelname)s | %(asctime)s | %(message)s', level=logging.INFO)

    config = parse_config(os.path.join(Path(__file__).parents[1], 'settings.json'))

    image_maker = ImageMaker('images', config['include_directories'], config['exclude_directories'],
                             config['no_caption_directories'])

    newcomers = {}

    bot = Bot(config['token'])
    dp = Dispatcher(image_maker=image_maker, owner_id=config['owner_id'], chat_ids=config['chat_ids'],
                    newcomers=newcomers, kick_delay=config['kick_delay'], messages_text=config['messages_text'])
    dp.include_routers(member.router, admin.router)

    await dp.start_polling(bot)

if __name__ == '__main__':
    asyncio.run(main())
captcha_bot/__main__.py
Dofoerix/recaptcha-tg-bot
import json
import os


def parse_config(path: str) -> dict[str, str | int | list]:
    """Return a dict with specified config file keys and values. If file isn't found create a new one."""
    try:
        with open(path, 'r', encoding='UTF-8') as config_file:
            return json.load(config_file)
    except FileNotFoundError:
        config = {
            'token': '',
            'owner_id': 0,
            'chat_ids': [],
            'include_directories': [],
            'exclude_directories': [],
            'no_caption_directories': [],
            'kick_delay': 0,
            'messages_text': {
                'joined': ('Привет, {username}! Реши капчу, пожалуйста\n'
                           'В своём следующем сообщении напиши цифры от 1 до 9, соответствующие картинкам '
                           '(отсчёт начинается слева сверху)'),
                'answer': ('Твой ответ содержит {correct} правильных ответов. {congrats}\n'
                           'Правильные ответы: {answers}'),
                'no_nums': 'Ни одной цифры не написал...',
                'no_text': 'Это даже не текст...',
                '0_correct': 'Что-то грустно...',
                '1_correct': 'Ну хоть что-то...',
                '2_correct': 'Лучше, чем хоть что-то...',
                '3_correct': 'Неплохо!',
                '4_correct': 'Поздравляю!',
                'incorrect': 'Кажется ты решал как то не так...'
            }
        }

        with open(path, 'w', encoding='UTF-8') as config_file:
            json.dump(config, config_file, indent=4, ensure_ascii=False)

        raise FileNotFoundError(f'{os.path.basename(path)} file wasn\'t found. Fill the fields in the created file.')
captcha_bot/config.py
Dofoerix/recaptcha-tg-bot
from typing import Optional

from aiogram.filters import BaseFilter
from aiogram.types import Message


class ChatFilter(BaseFilter):
    async def __call__(self, message: Message, chat_ids: list[Optional[int]]) -> bool:
        if chat_ids:
            return message.chat.id in chat_ids
        else:
            return True
captcha_bot/filters/chat.py
Dofoerix/recaptcha-tg-bot
import asyncio

from aiogram.filters import BaseFilter
from aiogram.types import Message


class NewcomerFiter(BaseFilter):
    async def __call__(self, message: Message, newcomers: dict[int, tuple[list[int], asyncio.Task]]) -> bool:
        return message.from_user.id in newcomers
captcha_bot/filters/newcomer.py
Dofoerix/recaptcha-tg-bot
from aiogram.filters import BaseFilter
from aiogram.types import Message


class OwnerFilter(BaseFilter):
    async def __call__(self, message: Message, owner_id: int) -> bool:
        return message.from_user.id == owner_id
captcha_bot/filters/owner.py
Dofoerix/recaptcha-tg-bot
from aiogram import Router, F
from aiogram.types import Message, BufferedInputFile
from aiogram.filters import CommandObject
from aiogram.filters import Command, CommandStart

from captcha_bot.filters.owner import OwnerFilter
from captcha_bot.image import ImageMaker


router = Router()
router.message.filter(F.chat.type == 'private', OwnerFilter())

@router.message(CommandStart())
async def start(message: Message):
    await message.reply('Привет, вот мои команды:\n'
                        '/start - вывести этот текст\n'
                        '/create <подпись> - вывести капчу с заданной подписью\n'
                        '/create_random - вывести капчу со случайной подписью')

@router.message(Command('create'))
async def create(message: Message, command: CommandObject, image_maker: ImageMaker):
    if command.args:
        try:
            image, answers = image_maker.create(command.args)
        except KeyError:
            await message.reply('К сожалению, директории для этой подписи нет')
            return
        await message.reply_photo(BufferedInputFile(image, 'captcha.jpg'),
                                  caption=f'Ответы: {str(answers).strip("[]")}')
    else:
        await message.reply('Напишите подпись после команды')

@router.message(Command('create_random'))
async def create_random(message: Message, image_maker: ImageMaker):
    image, answers = image_maker.create_random()
    await message.reply_photo(BufferedInputFile(image, 'captcha.jpg'), caption=f'Ответы: {str(answers).strip("[]")}')
captcha_bot/handlers/admin.py
Dofoerix/recaptcha-tg-bot
import asyncio
import logging

from aiogram import Router, F
from aiogram.filters.chat_member_updated import ChatMemberUpdatedFilter, IS_MEMBER, IS_NOT_MEMBER
from aiogram.types import ChatMemberUpdated, BufferedInputFile, Message

from captcha_bot.image import ImageMaker
from captcha_bot.filters.newcomer import NewcomerFiter
from captcha_bot.filters.chat import ChatFilter


router = Router()
router.chat_member.filter(F.chat.type.in_({'group', 'supergroup'}), ChatFilter())
router.message.filter(F.chat.type.in_({'group', 'supergroup'}), NewcomerFiter(), ChatFilter())

async def _delay_kick(event: ChatMemberUpdated, newcomers: dict[int, tuple[list[int], asyncio.Task]], delay: int) -> None:
    user = event.new_chat_member.user
    logging.info(f'{user.username} ({user.full_name}) will be kicked in {delay} minute(s)')
    try:
        await asyncio.sleep(delay * 60)
        await event.chat.unban(user.id)
        logging.info(f'{user.username} ({user.full_name}) was kicked')
    except asyncio.CancelledError:
        logging.info(f'{user.username} ({user.full_name}) won\'t be kicked')
    finally:
        del newcomers[user.id]


@router.chat_member(ChatMemberUpdatedFilter(IS_NOT_MEMBER >> IS_MEMBER))
async def member_joined(
    event: ChatMemberUpdated,
    image_maker: ImageMaker,
    newcomers: dict[int, tuple[list[int], asyncio.Task]],
    kick_delay: int,
    messages_text: dict[str, str]
):
    user = event.new_chat_member.user
    image, answers = image_maker.create_random()
    text = messages_text['joined'].format(username=f'@{user.username}', first_name=user.first_name,
                                          full_name=user.full_name, answers=str(answers).strip('[]'))
    await event.answer_photo(
        BufferedInputFile(image, 'captcha.jpg'),
        caption=text
    )
    if kick_delay:
        kick_task = asyncio.create_task(_delay_kick(event, newcomers, kick_delay))
    else:
        kick_task = None
    newcomers[user.id] = (answers, kick_task)

@router.message()
async def solution(message: Message, newcomers: dict[int, tuple[list[int], asyncio.Task]],
                   messages_text: dict[str, str]):
    user = message.from_user

    if message.text:
        text = message.text
    elif message.caption:
        text = message.caption
    else:
        text = None

    answers = newcomers[user.id][0]
    correct = 0
    incorrect = 0
    user_answers = []

    if text:
        for symbol in text:
            if symbol.isnumeric():
                if int(symbol) in answers and symbol not in user_answers:
                    user_answers.append(symbol)
                    correct += 1
                elif int(symbol) not in answers:
                    incorrect += 1

        if not any(symbol.isdigit() for symbol in text):
            congrats = messages_text['no_nums']
        elif correct == 0:
            congrats = messages_text['0_correct']
        elif correct == 1:
            congrats = messages_text['1_correct']
        elif correct == 2:
            congrats = messages_text['2_correct']
        elif correct == 3:
            congrats = messages_text['3_correct']
        elif correct > 3 and incorrect > 2:
            congrats = messages_text['incorrect']
        elif correct > 3:
            congrats = messages_text['4_correct']
    else:
        congrats = messages_text['no_text']

    text = messages_text['answer'].format(username=f'@{user.username}', first_name=user.first_name,
                                          full_name=user.full_name, correct=correct, incorrect=incorrect,
                                          congrats=congrats, answers=str(answers).strip('[]'))

    await message.reply(text)

    kick_task = newcomers[user.id][1]
    if isinstance(kick_task, asyncio.Task):
        kick_task.cancel()
        logging.info(f'{user.username} ({user.full_name}) kick task was cancelled')
    else:
        del newcomers[user]
captcha_bot/handlers/member.py
Dofoerix/recaptcha-tg-bot
import random
import io
import os
from pathlib import Path
from typing import Optional

from PIL import Image, ImageFont
from PIL.ImageDraw import ImageDraw


class ImageMaker:
    def __init__(self, images_directory: str, include_directories: Optional[list[str]] = None,
                 exclude_directories: Optional[list[str]] = None,
                 no_caption_directories: Optional[list[str]] = None) -> None:
        self.images_path = Path(images_directory)

        self.images_dict: dict[str, list[str]] = {}

        for item in self.images_path.iterdir():
            if item.is_dir():
                if include_directories:
                    if item.name not in include_directories:
                        continue
                if exclude_directories:
                    if item.name in exclude_directories:
                        continue
                self.images_dict[item.name] = []
                for inner_item in Path(os.path.join(self.images_path, item.name)).iterdir():
                    if inner_item.is_file():
                        self.images_dict[item.name].append(inner_item.name)

        for directory, files in self.images_dict.items():
            if len(files) < 4:
                raise ValueError(f'There must be at least four images in \'{directory}\' directory')

        self.images_list = [(directory, file) for directory, files in self.images_dict.items() for file in files]

        # These images can't be used as caption, but still appear in captcha
        for directory in list(self.images_dict.keys()):
            if directory in no_caption_directories:
                del self.images_dict[directory]

        self.coordinates = [(8, 128), (138, 128), (268, 128),
                            (8, 258), (138, 258), (268, 258),
                            (8, 388), (138, 388), (268, 388)]

    def create(self, text: str) -> tuple[bytes, list[int]]:
        """Create reCAPTCHA styled image with defined caption."""
        with Image.open(os.path.join(Path(__file__).parents[1], 'sample.jpg')) as sample:
            draw = ImageDraw(sample)
            draw.text((32, 50), text, font=ImageFont.truetype('Roboto-Black.ttf', size=27))

            correct = random.sample([(text, image) for image in self.images_dict[text]], 4)

            all_incorrect = [image for image in self.images_list if image[0] != text]
            incorrect = random.sample(all_incorrect, 5)

            all = correct + incorrect
            random.shuffle(all)

            correct_nums: list[int] = []

            for num, image, coord in zip(range(9), all, self.coordinates):
                if image in correct:
                    correct_nums.append(num + 1)
                with Image.open(os.path.join(self.images_path, image[0], image[1])) as image:
                    image = image.resize((126, 126), Image.Resampling.LANCZOS)
                    sample.paste(image, coord)

            buf = io.BytesIO()
            sample.save(buf, 'JPEG')

            return buf.getvalue(), correct_nums

    def create_random(self) -> tuple[bytes, list[int]]:
        """Create reCAPTCHA styled image with random caption."""
        return self.create(random.choice(list(self.images_dict.keys())))
captcha_bot/image.py
Dofoerix/recaptcha-tg-bot