Вот такой я коварный: пообещал продолжение туториала и ничего не публиковал несколько месяцев. Исправляюсь.
Первая часть тут.
Полный код второй части здесь.
Итак, мы хотим реализовать команды в боте и прочие прелести, которые будут работать только для админов. Поэтому бот должен уметь проверять, является ли участник чата админом.
Нет, мы не получаем эту информацию с каждым сообщением от участника. Так что мы не можем использовать что-нибудь вроде 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):
tablename = 'chats'
id = fields.BigIntField(pk=True)
last_admins_update = fields.DatetimeField(null=True)
class ChatMember(Model):
tablename = 'chat_members'
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) будет делать следующее:
"Обнулять" всех админов группы
Получать всех админов группы с помощью функции
bot.get_participants(...)Сохранять новое время
last_admins_update
Позже мы будем вызывать эту функцию тогда, когда кто-то пишет в чат команду /reload, а также каждый час при получении новых сообщений.
В utils.py:
from tortoise import timezone
from telethon.tl.types import ChannelParticipantsAdmins
...
async def reload_admins(chat_id):
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)И мы сделали это. Теперь наш бот "знает", кто в чате администратор, и мы можем это использовать!
Что дальше
В следующей части туториала мы сделаем полноценные команды для админов. Админы смогут банить и разбанивать, ограничивать участников и выдавать им предупреждения (которые бот будет считать).
Получившийся код вы можете посмотреть на GitHub.
