Вот такой я коварный: пообещал продолжение туториала и ничего не публиковал несколько месяцев. Исправляюсь.

Продолжаем писать своего крутого бота-модератора чатов на Python. Все части туториала:

• Часть 1. Создание бота

• Часть 2. Проверка админов

• Часть 3. Команды для модерации

Полный код для этой части на GitHub

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

Нет, мы не получаем эту информацию с каждым сообщением от участника. Так что мы не можем использовать что-нибудь вроде if event.is_sender_admin.

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

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

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

Все боты-модераторы в Телеграме делают примерно то же самое.

Подключаем базу данных

Для хранения данных, нам, конечно, понадобится БД. В этом туториале я хочу взять несложную для использования асинхронную ORM. Будем использовать Tortoise ORM и её инструмент для миграций Aerich.

Устанавливаем библиотеки:

$ pip install tortoise-orm aerich

Драйвер для SQLite устанавливается по умолчанию. Можете использовать другие СУБД (вот так).

Модели

Для моделей создадим файл models.py. Сделаем две модели:

• Чаты. У каждого чата будем хранить время последнего обновления админов

• Участники чатов. У них будут такие поля: айди пользователя, айди самого чата и is_admin — является ли этот пользователь админом в этом чате.

from tortoise.models import Model
from tortoise import fields

class Chat(Model):
    id = fields.BigIntField(pk=True)
    last_admins_update = fields.DatetimeField(null=True)

class ChatMember(Model):
    id = fields.IntField(pk=True)
    user_id = fields.BigIntField()
    chat_id = fields.BigIntField()
    is_admin = fields.BooleanField(default=False)

Замечу, что для айди пользователей и чатов следует использовать BigInt. С недавних пор они перестали влезать в обычный Int :)

Конфигурация

В config.py прописываем путь к базе данных (файл создастся автоматически). Например:

DATABASE_URI = 'sqlite:///admin_bot.db'

В __init__.py указываем конфиг Tortoise:

TORTOISE_ORM = {
    'connections': {'default': config.DATABASE_URI},
    'apps': {
        'app': {
            'models': ['app.models', 'aerich.models'],
            'default_connection': 'default',
        },
    },
}

Кроме пути к моделям app.models мы указали тут aerich.models для миграций.

Подключаем:

async def start():
    await Tortoise.init(config=TORTOISE_ORM)
    ...

Миграции

Настроим автоматические миграции. Используем конфиг в переменной TORTOISE_ORM:

$ aerich init -t app.TORTOISE_ORM

Должны сгенерироваться файл aerich.ini и папка migrations.

Инициализируем базу данных:

$ aerich init-db

Ура. Мы сделали всё, что нужно, чтобы работать с базой данных. Теперь начинаем эти самые данные хранить.

Храним все чаты

Мы будем вносить группу в базу данных только тогда, когда кто-то добавляет бота в эту группу. Просто дополним нашу функцию on_join:

@bot.on(events.ChatAction(func=lambda event: event.is_group and event.user_added and event.user_id == bot.me.id))
async def on_join(event: events.ChatAction.Event):
    await event.respond('Приветствую, господа!')
    chat = await Chat.get_or_none(id=event.chat.id)
    if chat is None:
        chat = Chat(id=event.chat.id)
        await chat.save()

Здесь мы просто получаем чат в БД по его id, а если такого чата нет, то добавляем его в БД.

Важно отметить одну особенность Telethon. Есть event.chat_id (свойство события), а есть event.chat.id (свойство чата из события). Как ни странно, они различаются.

event.chat_id — это id чата, начинающееся с -100 — как в Bot API. event.chat.id — это обычный id чата, без -100.

В БД, конечно, нужно хранить все id в одном формате, чтобы не было несостыковок. Поэтому будем в таких случаях всегда использовать event.chat.id.

Вспомогательные функции

Нам понадобятся функции для получения и изменения об участниках групп. Для всех вспомогательных функций создадим отдельный файл utils.py. Сейчас напишем пару функций:

• update_chat_member(chat_id, user_id, **kwargs) будет обновлять объект участника с нужными значениями (или создавать его, если его ещё нет)

• is_admin(chat_id, user_id) будет возвращать True, если если в базе данных этот пользователь записан как админ этой группы

from app import bot
from app.models import Chat, ChatMember


async def update_chat_member(chat_id: int, user_id: int, **kwargs):
    await ChatMember.update_or_create(chat_id=chat_id, user_id=user_id, defaults=kwargs)


async def is_admin(chat_id: int, user_id: int):
    member = await ChatMember.get_or_none(chat_id=chat_id, user_id=user_id)
    return member and member.is_admin

Обновление списка админов

Теперь мы, наконец, можем писать код для сохранения администраторов чата. Функция reload_admins(chat_id) будет делать следующее:

1. "Обнулять" всех админов группы

2. Получать всех админов группы с помощью функции bot.get_participants(...)

3. Сохранять новое время last_admins_update

Позже мы будем вызывать эту функцию тогда, когда кто-то пишет в чат команду /reload, а также каждый час при получении новых сообщений.

В utils.py:

from tortoise import timezone

from telethon.tl.types import ChannelParticipantsAdmins
...

async def reload_admins(chat_id: int):
    await ChatMember.filter(chat_id=chat_id, is_admin=True).update(is_admin=False)

    participants = await bot.get_participants(chat_id, filter=ChannelParticipantsAdmins())
		for participant in participants:
        await update_chat_member(chat_id, participant.id, is_admin=True)

    chat = await Chat.get(id=chat_id)
    chat.last_admins_update = timezone.now()
    await chat.save()

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

Чтобы получить текущее время и не запутаться в таймзонах, мы используем специальную функцию tortoise.timezone.now().

Обновление каждый час и по команде

Ну что же, теперь нам нужно эту самую функцию reload_admins использовать.

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

@bot.on(events.ChatAction(func=lambda event: event.is_group and event.user_added and event.user_id == bot.me.id))
async def on_join(event: events.ChatAction.Event):
    await event.respond('Приветствую, господа!')
    chat = await Chat.get_or_none(id=event.chat.id)
    if chat is None:
        chat = Chat(id=event.chat.id)
        await chat.save()
        await reload_admins(event.chat.id)

Во-вторых, обновлять админов нужно по команде /reload в группе. Напишем её хендлер в handlers.py:

from app.utils import reload_admins

...
@bot.on(events.NewMessage(func=lambda event: event.text.lower() == '/reload' and event.is_group))
async def reload_command(event: Message):
    await reload_admins(event.chat.id)
    await event.respond('Список админов группы обновлен')

Теперь вы можете проверить, что всё работает.

И в-третьих, при каждом получении сообщения из группы, мы будем проверять: как давно мы обновляли список администраторов в этой группе? Если больше часа назад, то обновляем ещё раз.

(Если будете тестировать бота, сделайте его админом в группе, чтобы он видел все сообщения.)

from datetime import timedelta
from tortoise import timezone
...

@bot.on(events.NewMessage(func=lambda event: event.is_group))
async def new_message(event: Message):
    query = chats.select(chats.c.id == event.chat.id)
    chat = await db.fetch_one(query)
    if datetime.now() - chat['last_admins_update'] > timedelta(hours=1):
        await reload_admins(event.chat.id)

И мы сделали это. Теперь наш бот "знает", кто в чате администратор, и мы можем это использовать!

Что дальше

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

Следующая часть