16.07.2024 16:39
NikitinIlya 2 1800 Источник

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

Перед началом хотелось бы уточнить, что данная статья или цикл не является полноценным учебником по созданию telegram ботов на Python, а лишь является пошаговой инструкцией, как разработать именно тот функционал, который был описан выше. Также статья не из разряда «… с нуля» и требует базовых знаний как языка, так и библиотеки.

На Python есть множество различных библиотек, но мы будем использовать именно асинхронную библиотеку aiogram 3.x версии.

Создание бота

Прежде чем начать написание кода, нам нужно создать самого бота с помощью официального бота @BotFather.

Прописываем команду /newbot для начала процесса создания бота. Создание занимает меньше минуты - нужно ввести:

  • Имя бота

  • Любой не занятый username с окончанием на bot

 

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

Ещё в BotFather можно задать боту аватарку, описание, команды и прочее.

Код бота

Перейдём к самому интересному — созданию функциональности бота. Код будет состоять из нескольких обработчиков команд и двух функций — проверка права администратора и парсинг времени.

Начнем с базового — импортов, объявления переменных и запуск бота:

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())

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

Боту задан парсинг HTML по умолчанию, а для роутера router задан фильтр F.chat.type != “private”. Нужно это для того, чтоб роутер реагировал только на сообщения в группах, а за ответ в личных сообщениях отвечает дефолтный Dispatcher. Так, мы избавились от лишних проверок.

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

Обработчики команд модерирования и требуемые функции

Теперь нам нужно научить бота реагировать на команды mute, ban, unmute, unban.

В рассматриваемом приложении пользовать должен отправлять в ответ на какое-либо сообщение любую из вышеуказанных команд, а бот проверит, является ли пользователь и сам бот администраторами, а уже после выдаст или снимет бан. При этом администратор может указать время бана в часах, днях или неделях (h, d, w).

К примеру, обработчик команды /mute будет выглядеть так:

@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> был заглушен!") 

Из интересного можно отметить структуру with suppress(TelegramBadRequest). С помощью этой записи будет игнорироваться ошибка, когда бот пытается ограничить или заблокировать действующего администратора.

Ограничивать (mute) пользователя можно с помощью bot.restrict_chat_member с параметром permissions (разрешения пользователя — в нашем случае это can_send_messages в значении False). А время задаётся с помощью параметра until_date. Код парсинга времени и проверки приведём ниже:

Функции is_admin и 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

С is_admin всё просто. Мы берём статус пользователя и бота и проверяем, является ли этот статус администратором или создателем чата. В случае если пользователь и бот является администраторами, возвращается True. В ином случае, False.

А вот с парсингом всё сложнее. Сюда передаётся time — это аргумент команды mute или ban, прописываемой пользователем. Если этого аргумента нет — возвращается None, и пользователю принимаются ограничения навсегда.

Далее мы делим time на группы value — число и unit — обозначение времени (т.е. часы, дни или недели) с помощью re.match и получаем нынешнее время. Если не найден нужный юнит в структуре match case или не удалось поделить time на группы, то также возвращается None.

Логика получения until_date заключается в том, что к ранее полученному нынешнему времени прибавляется время, указанное пользователем. Вот так всё просто :)

Обработка снятия ограничения

Как же без снятия ограничений? Тут всё ещё проще, те же проверки и различие только в передаваемом параметре — мы разрешаем пользователю писать текстовые сообщения и остальные типы сообщений с помощью 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> были сняты!")

Теперь не составит труда написать обработчик команд ban и unban.

Обработчик команд ban и unban

Я думаю, что вы догадываетесь, что единственные отличия этих обработчиков в том, что для бана мы будем использовать другой метод — bot.ban_chat_member

Сами обработчики:

@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(" Блокировка была снята")

Естественно, не забываем про то, что мы оставили диспетчер для обработки личных сообщений. Наш бот подразумевает работу только в чате, так что бот будет отвечать на все сообщения в ЛС одним сообщением:

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

Итоговый код

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())

Деплой бота на облако

Теперь, после того как мы разработали полностью функциональность бота, мы можем развернуть его в облаке Amvera Cloud.

Почему Amvera?

  • Быстрый запуск — для деплоя нам понадобится только два дополнительных файла и немного свободного времени;

  • Доставка обновлений через git — нам не нужно будет перезаливать файлы на сайте — достаточно будет воспользоваться git push;

  • За счёт контейнеризации, приложение потребляет меньше ресурсов;

  • При регистрации выдаётся бесплатный промо баланс 111 рублей.

Подготовка к деплою

Как писалось выше, для развёртывания приложения на Python на Amvera, нам понадобится два дополнительных файла:

  • amvera.yml — файл конфигурации, где мы зададим параметры для работы приложения

  • requirements.txt — файл зависимостей — нужен для сборки приложения (установки зависимостей‑библиотек, используемых в проекте)

Сейчас мы займёмся заполнением этих файлов.

Файл зависимостей

requirements.txt можно собрать разными путями — как автоматически, так и вручную. Я рекомендую делать это вручную, так как команда pip freeze (команда для автоматического метода вывода всех локально установленных зависимостей) при использовании без виртуального окружения может вывести слишком много зависимостей, из‑за чего время сборки упадёт.

Составить этот файл очень легко — нужно просто выписать все используемые библиотеки pip и их версии в таком формате:

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

В нашем случае файл зависимостей будет выглядеть так:

DateTime==5.5
aiogram==3.9
asyncio==3.4.3

Файл конфигурации

Как говорилось выше amvera.yml — файл с инструкцией для правильной работы приложения. Этот файл для новичков будет более удобно создать в интерфейсе сайта. Также можно воспользоваться генератором или документацией Amvera

Заранее созданный файл amvera.yml выглядит так:

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

Обратите внимание на параметр scriptName секции run — значение должно быть названием вашего основного файла!

Деплой

И наконец-то перейдём к деплою! У нас готово всё: аккаунт, файлы конфигурации и зависимостей. Осталось лишь выбрать способ отправки файлов в репозиторий проекта и создать этот самый проект.

Для создания нашего первого проекта нам нужно нажать на кнопку “Создать” в личном кабинете Amvera.

В открывшемся окне выбираем название для нашего проекта (латиница/кириллица), тип сервиса выбираем “Приложение”, а как тариф для нашего бота более чем подойдёт тариф “Начальный”. Нажимаем далее.

Теперь нам доступно окно загрузки данных. Мы можем загрузить все файлы прямо сейчас через интерфейс, но в будущем будет практичнее использовать метод загрузки с git push. Способы не взаимоисключающие — можно пользоваться и git, и интерфейсом.

С загрузкой кода через интерфейс всё интуитивно ясно — нажимаем «Загрузить данные» и просто перекидываем нужные файлы/папки. Но сейчас я покажу последовательность команд для работы с git.

Доставка кода через Git

Одним из главных плюсов облака Amvera является наиболее удобная работа с обновлениями. Для того, чтобы доставить измененный код, достаточно ввести пару команд в терминал вместо перетаскивания через интерфейс. К тому же после пуша сборка вашего проекта начнётся автоматически.

Для удобства предоставлю явную последовательность команд:

  1. Инициализируем локальный репозиторий git командой

git init

  1. Подключим удалённый репозиторий amvera (этот url можно скопировать на странице проекта во вкладке “репозиторий”)

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

  1. Командой git add добавим все файлы в локальном репозитории и сделаем commit 

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

  2. Наконец, запушим все файлы в удалённый репозиторий amvera

git push amvera master

Если вы ранее добавляли файлы через интерфейс, возможно, потребуется выполнить команду git pull amvera master

Окончание создания проекта и первый запуск

Переходим к окну с созданием конфигурационного файла. Его также можно создать либо через интерфейс, либо загрузить любым способом в формате .yml файла. Так как у нас уже готов конфигурационный файл, мы пропускаем создание.

Когда создался проект, мы можем открыть его и загрузить все нужные файлы с помощью кнопки «Загрузить данные» во вкладке «Репозиторий».

И не забываем про переменную TOKEN, которую нам нужно задать секретом во вкладке «Переменные», где нам нужно прописать название переменной «TOKEN» и значение.

Теперь всё готово к запуску! Переходим во вкладку «конфигурация» и жмём на кнопку «Собрать». А если вы использовали Git, сборка начнется автоматически.

После завершения сборки приложение начнёт запускаться. Когда приложение будет находиться в статусе «Приложение запущено» значит, что ваш бот работает.

Итог

В этой статье мы разобрали, как написать бота модератора на Python и смогли задеплоить бота на сервер Amvera. Если статья была полезна для Вас.

Комментарии (2)


  1. Mcublog
    17.07.2024 02:06
    #27050608

    Здравствуйте, спасибо за статью.

    Значение токена бота наверное лучше убрать из статьи, ну или дополнительно написать, что токен нужно держать в секрете.


    1. NikitinIlya Автор
      17.07.2024 02:06
      #27051304

      Да, вы правы. Токен в коде лучше не хранить. Для этого есть переменные и секреты. Для токенов и другой чувствительной информации именно секреты подойдут