Aiogram 3: Профиль, админ-панель и реферальная система

https://habr.com/ru/articles/822809/

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

Сегодняшний бот будет включать следующие функции:

1. Возможность регистрации пользователей (создание таблицы в PostgreSQL, написание функции для добавления пользователей).
2. Функция для получения информации о пользователях (информации о себе для личного профиля и всех пользователей для админ-панели).
3. Простой профиль пользователя с информацией о количестве приглашенных людей и возможностью скопировать реферальную ссылку.
4. Простая админ-панель с проверкой на права администратора и отображением информации о всех пользователях с базы данных.
5. Простая реферальная система, фиксирующая, кто пригласил пользователя и сколько пользователей было приглашено с подгрузкой этих данных в личный профиль.

Для удобства подготовил готовый проект, о котором пойдет речь, на своем GitHub. Устанавливайте Git, если он у вас еще не установлен, и выполняйте команду в директории, в которой хотите:

git clone https://github.com/Yakvenalex/easy_refer_bot

Структура проекта:

- db_handler
    - __init__.py: Инициализация модуля.
    - db_funk.py: Функции для взаимодействия с PostgreSQL.

- handlers
    - __init__.py: Инициализация модуля.
    - admin_panel.py: Роутер для админ-панели.
    - user_router.py: Роутер для пользовательской части.

- keyboards
    - __init__.py: Инициализация модуля.
    - kbs.py: Файл со всеми клавиатурами.

- utils
    - __init__.py: Инициализация модуля.
    - utils.py: Файл с утилитами.

После клонирования репозитория необходимо выполнить простые настройки:

1. Установите библиотеки при помощи команды pip install -r requirements.txt

2. Настройте файл .env следующим образом:

TOKEN=your_bot_token
ADMINS=admin1,admin2
ROOT_PASS=dasfg531KKK331xklaS
PG_LINK=postgresql://username:password@host:port/dbname

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

Обязательно передайте список telegram_id – это список администраторов, которые будут иметь доступ к админ-панели.

ROOT_PASS это дополнительный пароль для базы данных (не для подключения, а для дополнительной проверки при дефолтных операциях – подробнее про библиотеку asyncpg-lite).

Рассмотрим файл с настройками (create_bot.py):

import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from asyncpg_lite import DatabaseManager
from decouple import config

# получаем список администраторов из .env
admins = [int(admin_id) for admin_id in config('ADMINS').split(',')]

# настраиваем логирование и выводим в переменную для отдельного использования в нужных местах
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# инициируем объект, который будет отвечать за взаимодействие с базой данных
db_manager = DatabaseManager(dsn=config('PG_LINK'), deletion_password=config('ROOT_PASS'))

# инициируем объект бота, передавая ему parse_mode=ParseMode.HTML по умолчанию
bot = Bot(token=config('TOKEN'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))

# инициируем объект бота
dp = Dispatcher()

Единственное на что обратите внимание:

bot = Bot(token=config('TOKEN'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))

Благодаря такой инициации наш бот по умолчанию будет считывать HTML теги с сообщений.

Инициируем объект, который будет отвечать за взаимодействие с базой данных:

db_manager = DatabaseManager(dsn=config('PG_LINK'), deletion_password=config('ROOT_PASS'))

Тут как раз и пригодится наш ROOT_PASS. Его необходимо будет отдельно передавать в дефолтных операциях, таких как удаление всех данных из таблицы или удаление самой таблицы (безопасность).

Рассмотрим главный файл бота (aiogram_run.py):

import asyncio
from create_bot import bot, dp, admins
from db_handler.db_funk import get_all_users
from handlers.admin_panel import admin_router
from handlers.user_router import user_router
from aiogram.types import BotCommand, BotCommandScopeDefault


# Функция, которая настроит командное меню (дефолтное для всех пользователей)
async def set_commands():
    commands = [BotCommand(command='start', description='Старт'),
                BotCommand(command='profile', description='Мой профиль')]
    await bot.set_my_commands(commands, BotCommandScopeDefault())


# Функция, которая выполнится когда бот запустится
async def start_bot():
    await set_commands()
    count_users = await get_all_users(count=True)
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, f'Я запущен🥳. Сейчас в базе данных <b>{count_users}</b> пользователей.')
    except:
        pass


# Функция, которая выполнится когда бот завершит свою работу
async def stop_bot():
    try:
        for admin_id in admins:
            await bot.send_message(admin_id, 'Бот остановлен. За что?😔')
    except:
        pass


async def main():
    # регистрация маршрутов
    dp.include_router(user_router)
    dp.include_router(admin_router)

    # регистрация функций при старте и завершении работы бота
    dp.startup.register(start_bot)
    dp.shutdown.register(stop_bot)

    # запуск бота в режиме long polling при запуске бот очищает все обновления, которые были за его моменты бездействия
    try:
        await bot.delete_webhook(drop_pending_updates=True)
        await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
    finally:
        await bot.session.close()


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

Обратите внимание на то что их необходимо регистрировать такой конструкцией:

dp.startup.register(start_bot)
dp.shutdown.register(stop_bot)

Функция start_bot подключает командую клавиатуру и вытягивает с базы данных общее количество пользователей.

В боте используются 2 роутера:

• user_router – отвечает за пользовательскую часть (регистрация и личный профиль)
• admin_router – отвечает за админ-панель (проверка на админа и подгрузка данных о всех пользователях)

Их мы разберем немного позже, а сейчас давайте поработаем с настройкой функцию в базе данных db_hanler/db_funk.py:

Импорты:

from create_bot import db_manager
import asyncio

После этого напишем функцию для создания таблицы:

async def create_table_users(table_name='users_reg'):
    async with db_manager as client:
        await client.create_table(table_name=table_name,
                                  columns=['user_id INT8 PRIMARY KEY',
                                           'full_name VARCHAR(255)',
                                           'user_login VARCHAR(255)',
                                           'refer_id INT8',
                                           'count_refer INT4 DEFAULT 0',
                                           'date_reg TIMESTAMP DEFAULT CURRENT_TIMESTAMP'])

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

• user_id – телеграм ID пользователя (берем с объекта message)
• full_name – полное имя пользователя (берем с объекта message)
• user_login – логин в телеграмм (берем с объекта message)
• refer_id – телеграмм id пользователя, который пригласил
• count_refer – количество приглашенных пользователей
• date_reg – дата и время регистрации, будет заполняться автоматически

Выполняем код:

asyncio.run(create_table_users())

В результате должна получиться такая таблица:

Таблица users_reg. Вы можете дать любое другое название.

Получаем информацию о себе (личный профиль):

async def get_user_data(user_id: int, table_name='users_reg'): async with db_manager as client: return await client.select_data(table_name=table_name, where_dict={'user_id': user_id}, one_dict=True) Достаточно передать только telegram_id (напоминаю, что с полным синтаксисом asyncpg-lite можно ознакомиться ТУТ).

Получаем всех пользователей:

async def get_all_users(table_name='users_reg', count=False):
    async with db_manager as client:
        all_users = await client.select_data(table_name=table_name)
        if count:
            return len(all_users)
        else:
            return all_users

Функция возвращает или количество всех пользователей (флаг count) или просто всех пользователей со всеми данными.

Функция для добавления пользователей (она обновляет и количество рефералов):

async def insert_user(user_data: dict, table_name='users_reg'):
    async with db_manager as client:
        await client.insert_data(table_name=table_name, records_data=user_data)
        if user_data.get('refer_id'):
            refer_info = await client.select_data(table_name=table_name,
                                                  where_dict={'user_id': user_data.get('refer_id')},
                                                  one_dict=True, columns=['user_id', 'count_refer'])
            await client.update_data(table_name=table_name,
                                     where_dict={'user_id': refer_info.get('user_id')},
                                     update_dict={'count_refer': refer_info.get('count_refer') + 1})

Изначально идет базовый синтаксис добавления пользователя, но после запускается проверка был ли передан refer_id. Если это так – то мы запускаем процесс обновления количества пользователей.

Далее нам останется просто импортировать отдельные функции в нужные места бота (в файлы aiogram_run.py, admin_panel.py, user_router.py).

Пользовательская часть handlers/user_router.py. Импортируем:

from aiogram import Router, F
from aiogram.filters import CommandStart, CommandObject, Command
from aiogram.types import Message
from create_bot import bot
from db_handler.db_funk import get_user_data, insert_user
from keyboards.kbs import main_kb, home_page_kb
from utils.utils import get_refer_id
from aiogram.utils.chat_action import ChatActionSender

Создаем роутер:

user_router = Router()

Далее напишем функцию, которая будет:

1. Проверять есть ли пользователь в базе данных
2. Если есть, то отправлять приветственное сообщение с клавиатурой входа в профиль
3. Если нет, то будет пытаться достать аргументы из command.args (для этого вынес функцию в utils, чтоб вы понимали зачем я вообще этот пакет использую).
4. Далее будет добавлять пользователя в базу данных (обновлять счетчик рефералов у того кто пригласил, если там был его id).

universe_text = 'Чтоб получить информацию о своем профиле воспользуйся кнопкой "Мой профиль" или специальной командой из командного меню.'

Хендлер команды старт

@user_router.message(CommandStart())
async def cmd_start(message: Message, command: CommandObject):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        user_info = await get_user_data(user_id=message.from_user.id)

    if user_info:
        response_text = f'{user_info.get("full_name")}, вижу что вы уже в моей базе данных. {universe_text}'
    else:
        refer_id = get_refer_id(command.args)
        await insert_user(user_data={
            'user_id': message.from_user.id,
            'full_name': message.from_user.full_name,
            'user_login': message.from_user.username,
            'refer_id': refer_id
        })
        if refer_id:
            response_text = (f'{message.from_user.full_name}, вы зарегистрированы в боте и закреплены за '
                             f'пользователем с ID <b>{refer_id}</b>. {universe_text}')
        else:
            response_text = (f'{message.from_user.full_name}, вы зарегистрированы в боте и ни за кем не закреплены. '
                             f'{universe_text}')

    await message.answer(text=response_text, reply_markup=main_kb(message.from_user.id))

universe_text отдельно вынес в глобальную переменную, чтоб код чище был.

В утилитах лежит эта простая функция:

def get_refer_id(command_args):
    try:
        return int(command_args)
    except (TypeError, ValueError):
        return None

Давайте коротко разберем функцию cmd_start.

В ней подгрузка данных с базы данных под имитацию набора текста. Берите на заметку, иногда сильно помогает.

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

Иначе запустится логика описанная выше.

Главная клавиатура выглядит так:

def main_kb(user_telegram_id: int):
    kb_list = [[KeyboardButton(text="👤 Мой профиль")]]
    if user_telegram_id in admins:
        kb_list.append([KeyboardButton(text="⚙️ Админ панель")])
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Воспользуйтесь меню:"
    )

Внутри реализовал простую проверку на админа. Для этого нужно передавать телеграм айди пользователя в функцию main_kb.

В боте это выглядит так:

Если зайдем в созданную таблицу, то увидим:

Это значит, что все работает. Переходим к профилю.

@user_router.message(Command('profile'))
@user_router.message(F.text.contains('Мой профиль'))
async def get_profile(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        user_info = await get_user_data(user_id=message.from_user.id)
        text = (f'👉 Ваш телеграм ID: <code><b>{message.from_user.id}</b></code>\n'
                f'👥 Количество приглашенных тобой пользователей: <b>{user_info.get("count_refer")}</b>\n\n'
                f'🚀 Вот твоя персональная ссылка на приглашение: '
                f'<code>https://t.me/easy_refer_bot?start={message.from_user.id}</code>')
    await message.answer(text, reply_markup=home_page_kb(message.from_user.id))

Тут сразу показываю новую фишку – использование двух декораторов на функции. Тут я обработал команду /profile из командного меню и текст «Мой профиль», так как и команда и текст – это все части message – ошибок никаких не будет.

Когда используете текстовый вход и вход с call_data – делите на разные функции и выносите общие куски.

С объекта message вытянули телеграмм id. Передали его в созданную функцию и получили данные по себе.

Сама реферальная ссылка имеет 2 части:

• Ссылка на бота (в моем случае: https://t.me/easy_refer_bot)
• Командный аргумент, который идет после ?start= (как раз для обработки этого случая предавали command: CommandObject)

После отправляется клавиатура, с возможностью выйти назад:

def home_page_kb(user_telegram_id: int):
    kb_list = [[KeyboardButton(text="🔙 Назад")]]
    if user_telegram_id in admins:
        kb_list.append([KeyboardButton(text="⚙️ Админ панель")])
    return ReplyKeyboardMarkup(
        keyboard=kb_list,
        resize_keyboard=True,
        one_time_keyboard=True,
        input_field_placeholder="Воспользуйтесь меню:"
    )

Для обработки этой команды использовал:

@user_router.message(F.text.contains('Назад'))
async def cmd_start(message: Message):
    await message.answer(f'{message.from_user.first_name}, Вижу что вы уже в моей базе данных. {universe_text}',
                         reply_markup=main_kb(message.from_user.id))

Чтобы избежать излишних обращений к базе данных и возможных ошибок, связанных с использованием CommandObject, лучше не смешивать хендлеры, обрабатывающие команды, с хендлерами, обрабатывающими текстовые сообщения.

На этом с кодом пользовательской части все. Давайте смотреть в бота:

Обратите внимание – при клике на ссылку она сразу добавится в буфер обмена. Это происходит благодаря тегу <code></code>.

Давайте теперь попробуем кого-то пригласить. Для этого мы поделимся своей реферальной ссылкой, пользователь по ней перейдет и увидит следующее:

Видим, что боту удалось забрать аргументом командного объекта телеграмм айди рефера. Проверим:

Видим, что пользователь не просто зарегистрирован, но за ним прикреплен его рефер. Кроме того, у рефера (того кто пригласил) счетчик приглашенных увеличился на 1.

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

Давайте посмотрим в моем личном профиле:

Видим, что данные с PostgreSQL подтянулись, а это говорит о том что можно переходить к админ-панели (файл admin_panel.py):

from aiogram import F, Router
from aiogram.types import Message
from aiogram.utils.chat_action import ChatActionSender
from create_bot import admins, bot
from db_handler.db_funk import get_all_users
from keyboards.kbs import home_page_kb


admin_router = Router()


@admin_router.message((F.text.endswith('Админ панель')) & (F.from_user.id.in_(admins)))
async def get_profile(message: Message):
    async with ChatActionSender.typing(bot=bot, chat_id=message.from_user.id):
        all_users_data = await get_all_users()

        admin_text = (
            f'👥 В базе данных <b>{len(all_users_data)}</b> человек. Вот короткая информация по каждому:\n\n'
        )

        for user in all_users_data:
            admin_text += (
                f'👤 Телеграм ID: {user.get("user_id")}\n'
                f'📝 Полное имя: {user.get("full_name")}\n'
            )

            if user.get("user_login") is not None:
                admin_text += f'🔑 Логин: {user.get("user_login")}\n'

            if user.get("refer_id") is not None:
                admin_text += f'👨‍💼 Его пригласил: {user.get("refer_id")}\n'

            admin_text += (
                f'👥 Он пригласил: {user.get("count_refer")} человек\n'
                f'📅 Зарегистрирован: {user.get("date_reg")}\n'
                f'\n〰️〰️〰️〰️〰️〰️〰️〰️〰️\n\n'
            )

    await message.answer(admin_text, reply_markup=home_page_kb(message.from_user.id))

Тут заслуживает внимания – это конструкция магического фильтра:

((F.text.endswith('Админ панель')) & (F.from_user.id.in_(admins)))

Проверили, является ли id пользователя, который вызвал команду «Админ панель» в списке администраторов.

Сам список администраторов сначала добавили в .env файл, после вытянули через python-decouple в create_bot.py файл, а затем его импортировали в файл админ панели.

Далее идет получение всех данных о пользователе и простое форматирование текста. Смотрим:

Админ-панель готова!