Simple moderator bot on Aiogram 3.x

Hello, in this instruction I will touch upon the topic of creating a simple moderator bot, which we will teach to issue and remove blocking and permission to send messages to the user for a while, after which we will deploy it on a cloud service.

Before we begin, I would like to clarify that this article or series is not a full-fledged tutorial on creating telegram bots in Python, but is only a step-by-step instruction on how to develop exactly the functionality described above. Also, the article is not from the category “… from scratch” and requires basic knowledge of both the language and the library.

There are many different libraries in Python, but we will use the asynchronous library aiogram version 3.x.

Creating a bot

Before we start coding, we need to create the bot itself using the official bot @BotFather.

We enter the command /newbot to start the process of creating a bot. Creation takes less than a minute – you need to enter:

And now, if you did everything correctly, we get a bot token, which will be useful to us very soon 🙂

In BotFather you can also give the bot an avatar, description, commands, etc.

Bot code

Let's move on to the most interesting part – creating the bot functionality. The code will consist of several command handlers and two functions – checking administrator rights and parsing time.

Let's start with the basics – imports, variable declarations and running the bot:

import asyncio
import logging
import re
import os

from contextlib import suppress
from datetime import datetime, timedelta

from aiogram import Bot, Dispatcher, types, F, Router
from aiogram.filters import Command, CommandObject
from aiogram.client.default import DefaultBotProperties
from aiogram.exceptions import TelegramBadRequest
from aiogram.enums.parse_mode import ParseMode
from aiogram.enums.chat_member_status import ChatMemberStatus

TOKEN = os.getenv("TOKEN")

logging.basicConfig(level=logging.INFO)

bot = Bot(TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()

router = Router()
router.message.filter(F.chat.type != "private")

async def main():
    dp.include_router(router)

    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

For the security of the token during deployment, we will store it in environment variables, which we will later set on the cloud. For convenience, logging is connected using the logging library.

The bot is set to parse HTML by default, and the router is set to filter F.chat.type != “private”. This is necessary so that the router responds only to messages in groups, and the default Dispatcher is responsible for responding to personal messages. Thus, we got rid of unnecessary checks.

In the launch function, the code removes webhooks so that the bot does not react to events sent while the bot was not running during launch, and to prevent possible problems during deployment.

Moderation command handlers and required functions

Now we need to teach the bot to respond to the commands mute, ban, unmute, unban.

In the application under consideration, the user must send any of the above commands in response to any message, and the bot will check whether the user and the bot itself are administrators, and then issue or remove the ban. In this case, the administrator can specify the ban time in hours, days or weeks (h, d, w).

For example, the /mute command handler would look like this:

@router.message(Command("mute"))
async def func_mute(message: types.Message, command: CommandObject, bot: Bot):
    reply_message = message.reply_to_message

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b> Произошла ошибка!</b>")
        return
    
    date = parse_time(command.args)
    mention = reply_message.from_user.mention_html(reply_message.from_user.first_name)

    with suppress(TelegramBadRequest):
        await bot.restrict_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, until_date=date, permissions=types.ChatPermissions(can_send_messages=False))
        await message.answer(f" Пользователь <b>{mention}</b> был заглушен!") 

Of interest is the structure with suppress(TelegramBadRequest). This entry will ignore the error when the bot tries to restrict or block the current administrator.

You can mute a user using bot.restrict_chat_member with the permissions parameter (user permissions – in our case this is can_send_messages in meaning False). And the time is set using the until_date parameter. The code for parsing time and checking is given below:

Functions is_admin and parse_time:

async def is_admin(message, bot):
    member = await bot.get_chat_member(message.chat.id, message.from_user.id)
    bot = await bot.get_chat_member(message.chat.id, bot.id)
    if member.status not in [ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.CREATOR] or bot.status != ChatMemberStatus.ADMINISTRATOR:
        return False
    return True

def parse_time(time: str | None):
    if not time:
        return None
    
    re_match = re.match(r"(\d+)([a-z])", time.lower().strip())
    now_datetime = datetime.now()

    if re_match:
        value = int(re_match.group(1))
        unit = re_match.group(2)

        match unit:
            case "h": time_delta = timedelta(hours=value)
            case "d": time_delta = timedelta(days=value)
            case "w": time_delta = timedelta(weeks=value)
            case _: return None
    else:
        return None
    
    new_datetime = now_datetime + time_delta
    return new_datetime

With is_admin everything is simple. We take the user and bot status and check whether this status is an administrator or a chat creator. If the user and bot are administrators, True is returned. Otherwise, False.

But with parsing, everything is more complicated. Time is passed here – this is the argument of the mute or ban command specified by the user. If this argument is not present, None is returned, and the user is permanently restricted.

Next, we divide time into groups value – a number and unit – a designation of time (i.e. hours, days or weeks) using re.match and get the current time. If the required unit is not found in the structure match case or it was not possible to divide time into groups, then None is also returned.

The logic of getting until_date is that the current time previously received is added to the time specified by the user. It's that simple 🙂

Processing the removal of restrictions

How can we do without removing restrictions? Everything is even simpler here, the same checks and the only difference is in the transmitted parameter – we allow the user to write text messages and other types of messages using can_send_other_messages=True.

@router.message(Command("unmute"))
async def func_unmute(message: types.Message, command: CommandObject, bot: Bot):
    reply_message = message.reply_to_message 

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b>  Произошла ошибка!</b>")
        return
    
    mention = reply_message.from_user.mention_html(reply_message.from_user.first_name)
    
    await bot.restrict_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, permissions=types.ChatPermissions(can_send_messages=True, can_send_other_messages=True))
    await message.answer(f" Все ограничения с пользователя <b>{mention}</b> были сняты!")

Now it will not be difficult to write a handler for the ban and unban commands.

Handler for ban and unban commands

I think you can guess that the only differences between these handlers are that we will use a different method for banning – bot.ban_chat_member

The handlers themselves:

@router.message(Command("ban"))
async def func_ban(message: types.Message, command: CommandObject, bot: Bot):
    reply_message = message.reply_to_message

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b>  Произошла ошибка!</b>")
        return
    
    date = parse_time(command.args)
    mention = reply_message.from_user.mention_html(reply_message.from_user.first_name)

    with suppress(TelegramBadRequest):
        await bot.ban_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, until_date=date)
        await message.answer(f" Пользователь <b>{mention}</b> был заблокирован!")

@router.message(Command("unban"))
async def func_unban(message: types.Message, bot: Bot):
    reply_message = message.reply_to_message

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b>  Произошла ошибка!</b>")
        return
    
    await bot.unban_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, only_if_banned=True)
    await message.answer(" Блокировка была снята")

Naturally, we do not forget that we left the dispatcher for processing personal messages. Our bot is intended to work only in the chat, so the bot will respond to all messages in the PM with one message:

@dp.message(F.chat.type == "private")
async def private(message: types.Message):
    await message.reply(" <b>Бот работает только в группах</b>")

Final code

import asyncio
import logging
import re
import os

from contextlib import suppress
from datetime import datetime, timedelta

from aiogram import Bot, Dispatcher, types, F, Router
from aiogram.filters import Command, CommandObject
from aiogram.client.default import DefaultBotProperties
from aiogram.exceptions import TelegramBadRequest
from aiogram.enums.parse_mode import ParseMode
from aiogram.enums.chat_member_status import ChatMemberStatus

TOKEN = os.getenv("TOKEN")

logging.basicConfig(level=logging.INFO)

bot = Bot(TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()

router = Router()
router.message.filter(F.chat.type != "private")

@dp.message(F.chat.type == "private")
async def private(message: types.Message):
    await message.reply(" <b>Бот работает только в группах</b>")

async def is_admin(message, bot):
    member = await bot.get_chat_member(message.chat.id, message.from_user.id)
    bot = await bot.get_chat_member(message.chat.id, bot.id)
    if member.status not in [ChatMemberStatus.ADMINISTRATOR, ChatMemberStatus.CREATOR] or bot.status != ChatMemberStatus.ADMINISTRATOR:
        return False
    return True

def parse_time(time: str | None):
    if not time:
        return None
    
    re_match = re.match(r"(\d+)([a-z])", time.lower().strip())
    now_datetime = datetime.now()

    if re_match:
        value = int(re_match.group(1))
        unit = re_match.group(2)

        match unit:
            case "h": time_delta = timedelta(hours=value)
            case "d": time_delta = timedelta(days=value)
            case "w": time_delta = timedelta(weeks=value)
            case _: return None
    else:
        return None
    
    new_datetime = now_datetime + time_delta
    return new_datetime

@router.message(Command("ban"))
async def func_ban(message: types.Message, command: CommandObject, bot: Bot):
    reply_message = message.reply_to_message

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b>  Произошла ошибка!</b>")
        return
    
    date = parse_time(command.args)
    mention = reply_message.from_user.mention_html(reply_message.from_user.first_name)

    with suppress(TelegramBadRequest):
        await bot.ban_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, until_date=date)
        await message.answer(f" Пользователь <b>{mention}</b> был заблокирован!")

@router.message(Command("unban"))
async def func_unban(message: types.Message, bot: Bot):
    reply_message = message.reply_to_message

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b>  Произошла ошибка!</b>")
        return
    
    await bot.unban_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, only_if_banned=True)
    await message.answer(" Блокировка была снята")

@router.message(Command("mute"))
async def func_mute(message: types.Message, command: CommandObject, bot: Bot):
    reply_message = message.reply_to_message

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b>  Произошла ошибка!</b>")
        return
    
    date = parse_time(command.args)
    mention = reply_message.from_user.mention_html(reply_message.from_user.first_name)

    with suppress(TelegramBadRequest):
        await bot.restrict_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, until_date=date, permissions=types.ChatPermissions(can_send_messages=False))
        await message.answer(f" Пользователь <b>{mention}</b> был заглушен!")

@router.message(Command("unmute"))
async def func_unmute(message: types.Message, command: CommandObject, bot: Bot):
    reply_message = message.reply_to_message 

    if not reply_message or not await is_admin(message, bot):
        await message.reply("<b> Произошла ошибка!</b>")
        return
    
    mention = reply_message.from_user.mention_html(reply_message.from_user.first_name)
    
    await bot.restrict_chat_member(chat_id=message.chat.id, user_id=reply_message.from_user.id, permissions=types.ChatPermissions(can_send_messages=True, can_send_other_messages=True))
    await message.answer(f" Все ограничения с пользователя <b>{mention}</b> были сняты!")

async def main():
    dp.include_router(router)

    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

Deploying a bot to the cloud

Now that we have developed the full functionality of the bot, we can deploy it to the cloud. Amvera Cloud.

Why Amvera?

  • Quick start – for deployment we only need two additional files and some free time;

  • Delivery of updates via git – we will not need to re-upload files on the site – it will be enough to use git push;

  • Due to containerization, the application consumes fewer resources;

  • Upon registration you will receive a free one promo balance 111 rubles.

Preparing for deployment

As written above, to deploy a Python application on Amvera, we will need two additional files:

  • amvera.yml – a configuration file where we will set the parameters for the application to work

  • requirements.txt – dependency file – needed to build the application (install dependencies-libraries used in the project)

Now we will start filling these files.

Dependency file

requirements.txt can be built in different ways – both automatically and manually. I recommend doing it manually, as the pip freeze command (the command for the automatic method of listing all locally installed dependencies) when used without a virtual environment can list too many dependencies, which will cause build times to drop.

It is very easy to create this file – you just need to write out all the used pip libraries and their versions in this format:

библиотека1==версия1
библиотека2==версия2

In our case, the dependencies file will look like this:

 

The configuration file

As mentioned above, amvera.yml is a file with instructions for the correct operation of the application. This file will be more convenient for beginners to create in the site interface. You can also use generator or Amvera documentation

The pre-created amvera.yml file looks like this:

meta:
  environment: python
  toolchain:
    name: pip
    version: "3.12"
build:
  requirementsPath: requirements.txt
run:
  persistenceMount: /data
  containerPort: "80"
  scriptName: main.py

Pay attention to the scriptName parameter of the run section – the value should be the name of your main file!

Deploy

And finally, let's move on to deployment! We have everything ready: an account, configuration files, and dependencies. All that's left is to choose a method for sending files to the project repository and create this very project.

To create our first project, we need to click on the “Create” button in our personal account Amvera.

In the window that opens, select a name for our project (Latin/Cyrillic), select “Application” as the service type, and the “Initial” tariff is more than suitable for our bot. Click next.

Now we have the data upload window available. We can upload all the files right now through the interface, but in the future it will be more practical to use the upload method with git push. The methods are not mutually exclusive – you can use both git and the interface.

Loading code through the interface is intuitive – click “Load data” and simply transfer the necessary files/folders. But now I will show the sequence of commands for working with git.

Delivering code via Git

One of the main advantages of the Amvera cloud is the most convenient work with updates. In order to deliver the changed code, it is enough to enter a couple of commands in the terminal instead of dragging and dropping through the interface. In addition, after the push, the assembly of your project will begin automatically.

For convenience, I will provide an explicit sequence of commands:

  1. Initialize the local git repository with the command

git init

  1. Let's connect a remote repository amvera (this url can be copied from the project page in the “repository” tab)

git remote add amvera https://git.amvera.ru/имя_пользователя/название_проекта

  1. Team git add add all files to the local repository and commit

    git add .
git commit -m "Комментарий"

  2. Finally, push all the files to the remote amvera repository

git push amvera master

If you have previously added files through the interface, you may need to run the command
git pull amvera master

Completion of project creation and first launch

Let's move on to the window with creating a configuration file. It can also be created either through the interface or uploaded in any way in the .yml file format. Since we already have a configuration file ready, we skip creating it.

Once the project is created, we can open it and download all the necessary files using the “Download Data” button in the “Repository” tab.

And don’t forget about the TOKEN variable, which we need to set as a secret in the “Variables” tab, where we need to write the name of the “TOKEN” variable and its value.

Now everything is ready to run! Go to the “configuration” tab and click on the “Build” button. And if you used Git, the build will start automatically.

Once the build is complete, the app will start running. When the app is in the “Application running” status, it means that your bot is working.

Summary

In this article, we discussed how to write a moderator bot in Python and were able to deploy the bot to the server Amvera. If the article was useful for you.

Similar Posts

Leave a Reply

Your email address will not be published. Required fields are marked *