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

Друзья, приветствую! В предыдущих статьях я подробно рассказал о разработке телеграм-ботов на aiogram версии 3.7. Мы уже рассмотрели следующие темы:

• Работа с текстовыми и медиа сообщениями

• Использование клавиатур

• Командное меню и команды, и как с ними работать

• Основы использования FSM

• Интеграция в телеграм-бота PostgreSQL

• Магические фильтры, имитация действий бота и многое другое.

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

Теперь, опираясь на этот фундамент, я предлагаю сделать мои статьи о телеграм-ботах на aiogram 3 более практическими и насыщенными примерами, с минимальным количеством теории. Это, как мне кажется, будет более увлекательно и полезно, давая вам больше практического опыта.

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

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

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

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

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

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

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

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. Далее будет добавлять пользователя в базу данных (обновлять счетчик рефералов у того кто пригласил, если там был его айдишник).

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 мы вытянули телеграмм айди. Передали его в созданную функцию и получили данные по себе.

Сама реферальная ссылка имеет 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)))

Тем самым мы проверили, является ли айди пользователя, который вызвал команду «Админ панель» в списке администраторов (подробнее про магические фильтры ТУТ).

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

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

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

Заключение

Друзья, поздравляю вас с успешным завершением очередного этапа! Сегодня мы разобрали, как создать телеграм-бота с личным профилем, админ-панелью и простой реферальной системой, используя библиотеку aiogram 3.7 и PostgreSQL через asyncpg-lite.

Это только начало нашего увлекательного пути в мире разработки телеграм-ботов. В следующих статьях вас ждет еще больше практических примеров и интересных задач. Мы будем углубляться в более сложные темы, такие как интеграция с внешними API, скрипты по расписанию, оплата в боте, деплой, мидлвари и многое другое.

Если пост был полезен — подписывайтесь, ставьте лайки и оставляйте комментарии.

До новых встреч и успешного кодинга!