← Назад к курсу
Полное пособие по разработке Telegram ботов с aiogram 3.x
🎯 Часть 1: Основы работы с aiogram 3.x
1.1 Базовый бот с обработкой сообщений
Минимальный рабочий бот:
import asyncio
import logging
from aiogram import Bot, Dispatcher, types, F
from aiogram.filters import CommandStart, Command
from aiogram.types import Message
from aiogram.enums import ParseMode
# Настройка логирования
logging.basicConfig(level=logging.INFO)
# Инициализация бота
bot = Bot(token="ВАШ_ТОКЕН_БОТА")
dp = Dispatcher()
# Обработка команды /start
@dp.message(CommandStart())
async def cmd_start(message: Message):
# Ответ с HTML-разметкой
await message.answer(
f"<b>Привет, {message.from_user.full_name}!</b>\n\n"
f"ID: <code>{message.from_user.id}</code>",
parse_mode=ParseMode.HTML
)
# Обработка текстовых сообщений
@dp.message(F.text == "привет")
async def hello_handler(message: Message):
await message.reply("И тебе привет! 👋")
# Обработка любых текстовых сообщений
@dp.message(F.text)
async def echo_handler(message: Message):
# Эхо-ответ с пользовательским текстом
await message.answer(f"Вы написали: {message.text}")
# Обработка изображений
@dp.message(F.photo)
async def photo_handler(message: Message):
await message.answer("Классное фото! 📸")
# Запуск бота
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
1.2 Работа с клавиатурами
Reply-клавиатура (простая клавиатура под полем ввода)
from aiogram.types import ReplyKeyboardMarkup, KeyboardButton, ReplyKeyboardRemove
from aiogram.utils.keyboard import ReplyKeyboardBuilder
# Создание Reply-клавиатуры (старый способ)
def get_main_keyboard():
keyboard = ReplyKeyboardMarkup(
keyboard=[
[KeyboardButton(text="📱 Профиль"), KeyboardButton(text="🛒 Магазин")],
[KeyboardButton(text="🆘 Помощь"), KeyboardButton(text="📞 Контакты")],
[KeyboardButton(text="📝 Оставить отзыв")]
],
resize_keyboard=True, # Подгоняет размер кнопок
one_time_keyboard=False, # Не скрывать после нажатия
selective=True # Показывать только нужным пользователям
)
return keyboard
# Создание Reply-клавиатуры через Builder (рекомендуется)
def get_main_kb():
builder = ReplyKeyboardBuilder()
builder.add(
KeyboardButton(text="📱 Профиль"),
KeyboardButton(text="🛒 Магазин"),
KeyboardButton(text="🆘 Помощь"),
KeyboardButton(text="📞 Контакты"),
KeyboardButton(text="📍 Отправить локацию", request_location=True),
KeyboardButton(text="📱 Отправить контакт", request_contact=True)
)
builder.adjust(2, 2, 1, 1) # Распределение по строкам (2, 2, 1, 1)
return builder.as_markup(resize_keyboard=True)
@dp.message(Command("menu"))
async def cmd_menu(message: Message):
await message.answer(
"Выберите действие:",
reply_markup=get_main_kb()
)
# Удаление клавиатуры
@dp.message(Command("hide"))
async def cmd_hide(message: Message):
await message.answer(
"Клавиатура скрыта",
reply_markup=ReplyKeyboardRemove()
)
Inline-клавиатура (встроенные кнопки в сообщении)
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
# Создание Inline-клавиатуры
def get_inline_keyboard():
builder = InlineKeyboardBuilder()
builder.row(
InlineKeyboardButton(text="👍 Нравится", callback_data="like"),
InlineKeyboardButton(text="👎 Не нравится", callback_data="dislike")
)
builder.row(
InlineKeyboardButton(text="🌐 Открыть сайт", url="https://example.com"),
InlineKeyboardButton(text="🔍 Поиск", switch_inline_query_current_chat="поиск"),
InlineKeyboardButton(text="📱 Поделиться", switch_inline_query="поделиться")
)
builder.row(
InlineKeyboardButton(text="ℹ️ Информация", callback_data="info"),
InlineKeyboardButton(text="⚙️ Настройки", callback_data="settings")
)
return builder.as_markup()
@dp.message(Command("inline"))
async def cmd_inline(message: Message):
await message.answer(
"Выберите действие:",
reply_markup=get_inline_keyboard()
)
# Обработка нажатий на inline-кнопки
@dp.callback_query(F.data == "like")
async def like_handler(callback: types.CallbackQuery):
# Ответ на callback_query (убираем "часики")
await callback.answer("Спасибо за лайк! ❤️", show_alert=False)
# Редактирование сообщения
await callback.message.edit_text(
"Вы поставили лайк!",
reply_markup=get_inline_keyboard()
)
@dp.callback_query(F.data.in_(["like", "dislike"]))
async def vote_handler(callback: types.CallbackQuery):
await callback.answer(f"Вы выбрали: {callback.data}")
Специальные inline-кнопки для WebApp
def get_webapp_keyboard():
builder = InlineKeyboardBuilder()
# Кнопка для открытия WebApp
builder.row(
InlineKeyboardButton(
text="🎮 Открыть приложение",
web_app=types.WebAppInfo(url="https://ваш-сайт.com/webapp")
)
)
return builder.as_markup()
@dp.message(Command("app"))
async def cmd_app(message: Message):
await message.answer(
"Нажмите кнопку ниже, чтобы открыть Web-приложение:",
reply_markup=get_webapp_keyboard()
)
1.3 Машина состояний (FSM) для диалогов
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.storage.memory import MemoryStorage
# Хранилище состояний (в памяти - для разработки)
storage = MemoryStorage()
dp = Dispatcher(storage=storage)
# Определение состояний
class RegistrationStates(StatesGroup):
waiting_for_name = State()
waiting_for_age = State()
waiting_for_email = State()
# Начало регистрации
@dp.message(Command("register"))
async def cmd_register(message: Message, state: FSMContext):
await state.set_state(RegistrationStates.waiting_for_name)
await message.answer("Введите ваше имя:")
# Обработка имени
@dp.message(RegistrationStates.waiting_for_name)
async def process_name(message: Message, state: FSMContext):
# Сохраняем имя в контекст состояния
await state.update_data(name=message.text)
await state.set_state(RegistrationStates.waiting_for_age)
await message.answer("Отлично! Теперь введите ваш возраст:")
# Обработка возраста
@dp.message(RegistrationStates.waiting_for_age)
async def process_age(message: Message, state: FSMContext):
if not message.text.isdigit():
await message.answer("Пожалуйста, введите число!")
return
age = int(message.text)
await state.update_data(age=age)
await state.set_state(RegistrationStates.waiting_for_email)
await message.answer("И последнее - введите ваш email:")
# Завершение регистрации
@dp.message(RegistrationStates.waiting_for_email)
async def process_email(message: Message, state: FSMContext):
email = message.text
# Получаем все сохраненные данные
data = await state.get_data()
# Сохраняем в базу данных (здесь просто выводим)
await message.answer(
f"✅ Регистрация завершена!\n\n"
f"Имя: {data['name']}\n"
f"Возраст: {data['age']}\n"
f"Email: {email}"
)
# Очищаем состояние
await state.clear()
# Отмена регистрации в любой момент
@dp.message(Command("cancel"))
@dp.message(F.text.lower() == "отмена")
async def cmd_cancel(message: Message, state: FSMContext):
current_state = await state.get_state()
if current_state is None:
return
await state.clear()
await message.answer("❌ Действие отменено")
1.4 Middleware и фильтры
# Кастомный фильтр
class AdminFilter(Filter):
def __init__(self, admin_ids: list[int]):
self.admin_ids = admin_ids
async def __call__(self, message: Message) -> bool:
return message.from_user.id in self.admin_ids
# Пример middleware для логирования
class LoggingMiddleware(BaseMiddleware):
async def __call__(self, handler, event, data):
user = data.get("event_from_user")
if user:
logging.info(f"User {user.id} ({user.username}) sent: {event.text}")
return await handler(event, data)
# Регистрация middleware
dp.update.outer_middleware(LoggingMiddleware())
# Использование кастомного фильтра
@dp.message(Command("admin"), AdminFilter(admin_ids=[123456789]))
async def admin_command(message: Message):
await message.answer("Вы администратор!")
🌐 Часть 2: Полная реализация WebApp
2.1 Frontend часть WebApp (HTML/JS)
index.html:
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Telegram WebApp</title>
<script src="https://telegram.org/js/telegram-web-app.js"></script>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: 100vh;
padding: 20px;
}
.container {
max-width: 800px;
margin: 0 auto;
background: white;
border-radius: 20px;
padding: 30px;
box-shadow: 0 20px 60px rgba(0,0,0,0.3);
}
.header {
text-align: center;
margin-bottom: 30px;
}
.user-info {
background: #f8f9fa;
padding: 20px;
border-radius: 10px;
margin-bottom: 30px;
}
.btn {
display: block;
width: 100%;
padding: 15px;
margin: 10px 0;
border: none;
border-radius: 10px;
background: #007bff;
color: white;
font-size: 16px;
cursor: pointer;
transition: background 0.3s;
}
.btn:hover {
background: #0056b3;
}
.btn-danger {
background: #dc3545;
}
.btn-danger:hover {
background: #c82333;
}
.product-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 20px;
margin: 30px 0;
}
.product-card {
border: 1px solid #ddd;
border-radius: 10px;
padding: 20px;
text-align: center;
}
.product-price {
color: #28a745;
font-weight: bold;
font-size: 18px;
margin: 10px 0;
}
.hidden {
display: none;
}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1 id="app-title">Telegram WebApp</h1>
<p id="app-description">Интерактивное приложение в Telegram</p>
</div>
<div class="user-info">
<h3>👤 Информация о пользователе</h3>
<p id="user-name">Загрузка...</p>
<p id="user-id">ID: ...</p>
</div>
<div id="main-content">
<h2>🎮 Демонстрация функций</h2>
<button class="btn" onclick="showAlert()">Показать алерт</button>
<button class="btn" onclick="changeTheme()">Сменить тему</button>
<button class="btn" onclick="vibrate()">Вибрация</button>
<button class="btn" onclick="closeApp()">Закрыть приложение</button>
<button class="btn btn-danger" onclick="sendDataToBot()">Отправить данные боту</button>
<div class="product-grid">
<div class="product-card">
<h3>🎁 Продукт 1</h3>
<p class="product-price">10 звёзд</p>
<button class="btn" onclick="buyProduct(1)">Купить</button>
</div>
<div class="product-card">
<h3>🎮 Продукт 2</h3>
<p class="product-price">25 звёзд</p>
<button class="btn" onclick="buyProduct(2)">Купить</button>
</div>
</div>
<div id="response-area" class="hidden">
<h3>Ответ от бота:</h3>
<pre id="bot-response"></pre>
</div>
</div>
</div>
<script>
// Инициализация Telegram WebApp
const tg = window.Telegram.WebApp;
// Расширяем WebApp на весь экран
tg.expand();
// Устанавливаем тему
if (tg.colorScheme === 'dark') {
document.body.style.background = '#1a1a1a';
document.querySelector('.container').style.background = '#2d2d2d';
document.querySelector('.container').style.color = 'white';
}
// Получаем данные пользователя из Telegram
const user = tg.initDataUnsafe.user;
// Отображаем информацию о пользователе
if (user) {
document.getElementById('user-name').textContent =
`${user.first_name} ${user.last_name || ''} (@${user.username || 'нет'})`;
document.getElementById('user-id').textContent = `ID: ${user.id}`;
}
// Функции WebApp
function showAlert() {
tg.showAlert("Привет из WebApp! 🚀");
}
function changeTheme() {
if (tg.colorScheme === 'dark') {
tg.setHeaderColor('#667eea');
tg.setBackgroundColor('#667eea');
} else {
tg.setHeaderColor('#764ba2');
tg.setBackgroundColor('#764ba2');
}
}
function vibrate() {
tg.HapticFeedback.impactOccurred('medium');
}
function closeApp() {
tg.close();
}
// Отправка данных боту
function sendDataToBot() {
const data = {
action: 'webapp_data',
timestamp: new Date().toISOString(),
user_id: user?.id,
random_value: Math.random()
};
// Отправляем данные через Telegram WebApp
tg.sendData(JSON.stringify(data));
// Показываем ответ
document.getElementById('bot-response').textContent =
'Данные отправлены! Ждите ответа от бота...';
document.getElementById('response-area').classList.remove('hidden');
}
// Покупка продукта
function buyProduct(productId) {
const products = {
1: { name: 'Продукт 1', price: 10 },
2: { name: 'Продукт 2', price: 25 }
};
const product = products[productId];
// Отправляем запрос на покупку
tg.sendData(JSON.stringify({
action: 'buy_product',
product_id: productId,
product_name: product.name,
price: product.price
}));
tg.showAlert(`Покупка "${product.name}" за ${product.price} звёзд!`);
}
// Обработка сообщений от бота
tg.onEvent('webAppDataReceived', (event) => {
console.log('Получены данные:', event);
});
// Инициализация завершена
tg.ready();
</script>
</body>
</html>
2.2 Backend обработка WebApp (Python)
from aiogram.types import WebAppData, InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.utils.web_app import safe_parse_webapp_init_data
import json
# Обработка данных из WebApp
@dp.message(F.web_app_data)
async def web_app_data_handler(message: Message, bot: Bot):
web_app_data = message.web_app_data
try:
# Парсим JSON данные из WebApp
data = json.loads(web_app_data.data)
action = data.get('action')
if action == 'buy_product':
# Создаем инвойс для оплаты
product_id = data.get('product_id')
price = data.get('price')
await message.answer_invoice(
title=f"Покупка продукта {product_id}",
description=f"Цена: {price} звёзд",
provider_token="", # Для Telegram Stars
currency="XTR",
prices=[LabeledPrice(label="Звёзды", amount=price)],
payload=f"product_{product_id}_{price}"
)
elif action == 'webapp_data':
# Отправляем ответ обратно в WebApp через сообщение
await message.answer(
f"✅ Данные получены!\n\n"
f"Действие: {action}\n"
f"Время: {data.get('timestamp')}\n"
f"Случайное число: {data.get('random_value')}"
)
except json.JSONDecodeError:
await message.answer("❌ Ошибка разбора данных из WebApp")
# Обработчик для кнопки WebApp с проверкой данных
@dp.callback_query(F.data == "open_webapp")
async def open_webapp_handler(callback: types.CallbackQuery, bot: Bot):
# Создаем безопасную ссылку с временным токеном
web_app_url = "https://ваш-сайт.com/webapp"
# Можно добавить параметры для конкретного пользователя
user_id = callback.from_user.id
web_app_url += f"?user_id={user_id}"
keyboard = InlineKeyboardMarkup(
inline_keyboard=[[
InlineKeyboardButton(
text="Открыть приложение 🚀",
web_app=WebAppInfo(url=web_app_url)
)
]]
)
await callback.message.answer(
"Нажмите кнопку ниже, чтобы открыть Web-приложение:",
reply_markup=keyboard
)
await callback.answer()
# Валидация данных из WebApp на сервере
async def validate_webapp_data(init_data: str, bot_token: str) -> bool:
"""
Валидация данных из WebApp на сервере.
Используйте эту функцию при получении данных на ваш backend.
"""
try:
web_app_init_data = safe_parse_webapp_init_data(
token=bot_token,
init_data=init_data
)
return True
except ValueError:
return False
2.3 Настройка сервера для WebApp
from aiohttp import web
import ssl
# Настройки Webhook для продакшена
async def on_startup(bot: Bot):
# Для работы WebApp нужен HTTPS
await bot.set_webhook(
url="https://ваш-домен.com/webhook",
secret_token="ваш-секретный-токен",
drop_pending_updates=True
)
print("Webhook установлен")
async def on_shutdown(bot: Bot):
await bot.delete_webhook()
print("Webhook удален")
# Создание HTTPS сервера с SSL
def setup_ssl():
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(
certfile='path/to/cert.pem',
keyfile='path/to/key.pem'
)
return ssl_context
# Запуск через webhook (рекомендуется для WebApp)
async def start_webhook():
bot = Bot(token=TOKEN)
await on_startup(bot)
app = web.Application()
# Webhook handler
webhook_handler = SimpleRequestHandler(
dispatcher=dp,
bot=bot,
secret_token="ваш-секретный-токен"
)
webhook_handler.register(app, path="/webhook")
# Статика для WebApp
app.router.add_static('/webapp/', path='static/webapp/', name='webapp')
# Ручка для проверки WebApp данных
@app.post("/validate-webapp")
async def validate_webapp(request):
data = await request.json()
init_data = data.get('initData')
is_valid = await validate_webapp_data(init_data, TOKEN)
return web.json_response({
'valid': is_valid,
'timestamp': datetime.now().isoformat()
})
# Запуск сервера
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, '0.0.0.0', 8443)
await site.start()
print("Сервер запущен на https://0.0.0.0:8443")
try:
await asyncio.Future() # Бесконечный цикл
finally:
await on_shutdown(bot)
2.4 Пример: Мини-приложение магазина в WebApp
# Полная интеграция магазина с WebApp и платежами
class Product:
def __init__(self, id, name, description, price_stars):
self.id = id
self.name = name
self.description = description
self.price_stars = price_stars
# Каталог продуктов
products = [
Product(1, "🎮 Игровой контент", "Эксклюзивные предметы", 10),
Product(2, "💰 Премиум подписка", "Месяц премиум-доступа", 25),
Product(3, "🎁 Подарочный набор", "Коллекция бонусов", 50),
Product(4, "👑 VIP статус", "Особые привилегии", 100),
]
# Генерация HTML для WebApp
def generate_products_html():
html_products = []
for product in products:
html_products.append(f"""
<div class="product" data-id="{product.id}">
<h3>{product.name}</h3>
<p>{product.description}</p>
<div class="price">{product.price_stars} ⭐</div>
<button onclick="buyProduct({product.id})">Купить</button>
</div>
""")
return "\n".join(html_products)
# Backend обработчик покупок из WebApp
@dp.message(F.web_app_data)
async def handle_webapp_purchase(message: Message):
try:
data = json.loads(message.web_app_data.data)
if data.get('action') == 'purchase':
product_id = data['product_id']
product = next(p for p in products if p.id == product_id)
# Создание инвойса
await message.answer_invoice(
title=product.name,
description=product.description,
provider_token="",
currency="XTR",
prices=[LabeledPrice(label="Звёзды", amount=product.price_stars)],
payload=f"purchase_{product_id}_{message.from_user.id}"
)
except Exception as e:
await message.answer(f"Ошибка обработки покупки: {str(e)}")
# Команда для открытия магазина
@dp.message(Command("shop"))
async def cmd_shop(message: Message):
keyboard = InlineKeyboardBuilder()
for product in products:
keyboard.button(
text=f"{product.name} - {product.price_stars}⭐",
callback_data=f"view_product_{product.id}"
)
keyboard.button(
text="🎮 Открыть магазин в WebApp",
web_app=WebAppInfo(url="https://ваш-сайт.com/shop")
)
keyboard.adjust(1)
await message.answer(
"🛒 Магазин бота\n\n"
"Выберите товар или откройте полный каталог в WebApp:",
reply_markup=keyboard.as_markup()
)
💰 Часть 3: Платежи через Telegram Stars (полная реализация)
3.1 Настройка платежей в @BotFather
- Создайте бота: /newbot в @BotFather
- Включите платежи: /mybots → Ваш бот → Bot Settings → Payments
- Выберите провайдера: Telegram Stars
- Настройте валюту: XTR (Telegram Stars)
- Укажите информацию о продавце
3.2 Полная система платежей
from datetime import datetime
from typing import Dict, Optional
class PaymentSystem:
def __init__(self):
self.transactions: Dict[str, dict] = {}
self.user_balances: Dict[int, int] = {}
async def create_invoice(self, user_id: int, amount: int, description: str) -> dict:
"""Создание счета на оплату"""
invoice_id = f"inv_{user_id}_{datetime.now().timestamp()}"
transaction = {
'invoice_id': invoice_id,
'user_id': user_id,
'amount': amount,
'description': description,
'status': 'pending',
'created_at': datetime.now(),
'paid_at': None
}
self.transactions[invoice_id] = transaction
return transaction
async def process_payment(self, payment_data: dict) -> bool:
"""Обработка успешного платежа"""
telegram_charge_id = payment_data.get('telegram_payment_charge_id')
user_id = payment_data.get('user_id')
amount = payment_data.get('amount')
# Обновляем баланс пользователя
self.user_balances[user_id] = self.user_balances.get(user_id, 0) + amount
# Логируем транзакцию
print(f"Платеж обработан: {telegram_charge_id}, сумма: {amount} звёзд")
return True
async def get_user_balance(self, user_id: int) -> int:
"""Получение баланса пользователя"""
return self.user_balances.get(user_id, 0)
payment_system = PaymentSystem()
# Команда для проверки баланса
@dp.message(Command("balance"))
async def cmd_balance(message: Message):
balance = await payment_system.get_user_balance(message.from_user.id)
keyboard = InlineKeyboardBuilder()
keyboard.button(text="💳 Пополнить баланс", callback_data="add_balance")
keyboard.button(text="🛒 Магазин", callback_data="shop_main")
await message.answer(
f"💰 Ваш баланс: {balance} звезд\n\n"
"1 звезда = 1 цент (примерно)\n"
"100 звезд = 1 USD",
reply_markup=keyboard.as_markup()
)
# Пополнение баланса
@dp.callback_query(F.data == "add_balance")
async def add_balance_handler(callback: types.CallbackQuery):
keyboard = InlineKeyboardBuilder()
# Кнопки с разными суммами
amounts = [10, 25, 50, 100, 250, 500]
for amount in amounts:
keyboard.button(
text=f"{amount} ⭐",
callback_data=f"add_{amount}_stars"
)
keyboard.button(text="🎮 WebApp магазин", web_app=WebAppInfo(
url="https://ваш-сайт.com/add-balance"
))
keyboard.adjust(2)
await callback.message.edit_text(
"Выберите сумму для пополнения:",
reply_markup=keyboard.as_markup()
)
await callback.answer()
# Создание инвойса для пополнения
@dp.callback_query(F.data.startswith("add_"))
async def create_payment_invoice(callback: types.CallbackQuery):
amount = int(callback.data.split("_")[1])
# Создаем транзакцию
transaction = await payment_system.create_invoice(
user_id=callback.from_user.id,
amount=amount,
description=f"Пополнение баланса на {amount} звезд"
)
# Создаем инвойс в Telegram
await callback.message.answer_invoice(
title=f"Пополнение баланса на {amount} звезд",
description="После оплаты звезды будут зачислены на ваш баланс",
provider_token="", # Для Telegram Stars
currency="XTR",
prices=[LabeledPrice(label="Звезды", amount=amount)],
payload=f"balance_topup_{callback.from_user.id}_{amount}"
)
await callback.answer()
# Обработка успешного платежа
@dp.message(F.successful_payment)
async def successful_payment_handler(message: Message):
payment = message.successful_payment
# Извлекаем данные из payload
payload_parts = payment.invoice_payload.split("_")
action = payload_parts[0]
if action == "balance":
# Обработка пополнения баланса
user_id = int(payload_parts[2])
amount = int(payload_parts[3])
await payment_system.process_payment({
'telegram_payment_charge_id': payment.telegram_payment_charge_id,
'user_id': user_id,
'amount': amount
})
balance = await payment_system.get_user_balance(user_id)
await message.answer(
f"✅ Баланс успешно пополнен!\n\n"
f"Зачислено: {amount} звезд\n"
f"Новый баланс: {balance} звезд\n\n"
f"ID транзакции: {payment.telegram_payment_charge_id}"
)
elif action == "purchase":
# Обработка покупки товара
product_id = payload_parts[1]
user_id = int(payload_parts[2])
await message.answer(
f"✅ Покупка успешно завершена!\n\n"
f"Товар: {product_id}\n"
f"Сумма: {payment.total_amount} звезд\n\n"
"Спасибо за покупку!"
)
# Возврат средств
@dp.message(Command("refund"))
async def cmd_refund(message: Message, bot: Bot):
# В реальном приложении здесь должна быть форма для запроса возврата
# с указанием ID транзакции
await message.answer(
"🔄 Возврат средств\n\n"
"Для возврата платежа свяжитесь с поддержкой:\n"
"1. Укажите ID транзакции из чека\n"
"2. Причину возврата\n"
"3. Ваш ID пользователя\n\n"
"Возврат обрабатывается в течение 24 часов."
)
3.3 Пример: Цифровой товар с мгновенной доставкой
class DigitalProduct:
def __init__(self, id, name, type, content):
self.id = id
self.name = name
self.type = type # 'code', 'link', 'text', 'file'
self.content = content
digital_products = [
DigitalProduct(1, "🎮 Ключ Steam", "code", "XXXXX-XXXXX-XXXXX"),
DigitalProduct(2, "📚 Электронная книга", "link", "https://example.com/book.pdf"),
DigitalProduct(3, "🎵 Премиум доступ", "text", "Ваш доступ активирован до 01.01.2025"),
]
@dp.callback_query(F.data.startswith("buy_digital_"))
async def buy_digital_product(callback: types.CallbackQuery):
product_id = int(callback.data.split("_")[2])
product = next(p for p in digital_products if p.id == product_id)
# Создаем инвойс
await callback.message.answer_invoice(
title=product.name,
description="Мгновенная доставка после оплаты",
provider_token="",
currency="XTR",
prices=[LabeledPrice(label="Цифровой товар", amount=25)],
payload=f"digital_{product_id}_{callback.from_user.id}"
)
await callback.answer()
# Доставка цифрового товара после оплаты
async def deliver_digital_product(user_id: int, product: DigitalProduct):
"""Доставка цифрового товара пользователю"""
if product.type == "code":
message = f"🎁 Ваш ключ: {product.content}"
elif product.type == "link":
message = f"📥 Скачать: {product.content}"
elif product.type == "text":
message = product.content
try:
await bot.send_message(
chat_id=user_id,
text=message,
parse_mode=ParseMode.HTML
)
return True
except Exception as e:
print(f"Ошибка доставки товара: {e}")
return False
🚀 Часть 4: Запуск и развертывание
4.1 Файл конфигурации
# config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv
load_dotenv()
@dataclass
class Config:
BOT_TOKEN: str = os.getenv("BOT_TOKEN")
ADMINS: list = list(map(int, os.getenv("ADMINS", "").split(",")))
WEBHOOK_URL: str = os.getenv("WEBHOOK_URL")
WEBHOOK_PATH: str = "/webhook"
WEBAPP_HOST: str = "0.0.0.0"
WEBAPP_PORT: int = int(os.getenv("PORT", 8000))
REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")
DATABASE_URL: str = os.getenv("DATABASE_URL")
config = Config()
4.2 Docker-контейнеризация
# Dockerfile FROM python:3.11-slim WORKDIR /app # Установка зависимостей COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Копирование кода COPY . . # Создание пользователя для безопасности RUN useradd -m -u 1000 botuser && chown -R botuser:botuser /app USER botuser # Запуск бота CMD ["python", "main.py"]
4.3 requirements.txt
aiogram==3.8.1 aiohttp==3.9.1 python-dotenv==1.0.0 redis==5.0.1 sqlalchemy==2.0.23 alembic==1.12.1 python-multipart==0.0.6
4.4 Главный файл запуска
# main.py
import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.fsm.storage.redis import RedisStorage
from aiogram.webhook.aiohttp_server import SimpleRequestHandler
from aiohttp import web
from config import config
from handlers import routers
async def main():
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
# Инициализация бота и диспетчера
bot = Bot(token=config.BOT_TOKEN, parse_mode="HTML")
storage = RedisStorage.from_url(config.REDIS_URL)
dp = Dispatcher(storage=storage)
# Регистрация роутеров
for router in routers:
dp.include_router(router)
try:
# Режим разработки - polling
if not config.WEBHOOK_URL:
logging.info("Запуск в режиме polling...")
await dp.start_polling(bot)
# Продакшен - webhook
else:
logging.info("Запуск в режиме webhook...")
await bot.set_webhook(
url=config.WEBHOOK_URL + config.WEBHOOK_PATH,
secret_token=config.WEBHOOK_SECRET,
drop_pending_updates=True
)
app = web.Application()
webhook_handler = SimpleRequestHandler(
dispatcher=dp,
bot=bot,
secret_token=config.WEBHOOK_SECRET
)
webhook_handler.register(app, path=config.WEBHOOK_PATH)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, config.WEBAPP_HOST, config.WEBAPP_PORT)
await site.start()
logging.info(f"Сервер запущен на {config.WEBAPP_HOST}:{config.WEBAPP_PORT}")
await asyncio.Event().wait()
finally:
await bot.session.close()
if __name__ == "__main__":
asyncio.run(main())
📊 Часть 5: Мониторинг и логирование
import logging
from datetime import datetime
class PaymentLogger:
def __init__(self):
self.payment_logger = logging.getLogger('payments')
self.webapp_logger = logging.getLogger('webapp')
async def log_payment(self, user_id: int, amount: int, status: str, details: dict = None):
log_entry = {
'timestamp': datetime.now().isoformat(),
'user_id': user_id,
'amount': amount,
'status': status,
'details': details or {}
}
self.payment_logger.info(f"Платеж: {log_entry}")
# Здесь можно сохранить в базу данных
# await self.save_to_database(log_entry)
async def log_webapp_action(self, user_id: int, action: str, data: dict):
self.webapp_logger.info(
f"WebApp действие - Пользователь: {user_id}, "
f"Действие: {action}, Данные: {data}"
)
# Инициализация логгера
payment_logger = PaymentLogger()
# Использование в хэндлерах
@dp.message(F.successful_payment)
async def log_successful_payment(message: Message):
payment = message.successful_payment
await payment_logger.log_payment(
user_id=message.from_user.id,
amount=payment.total_amount,
status='success',
details={
'charge_id': payment.telegram_payment_charge_id,
'payload': payment.invoice_payload
}
)
# Дальнейшая обработка платежа...
🎯 Заключение
Это полное руководство охватывает все этапы разработки Telegram бота с aiogram 3.x:
- Основы: обработка сообщений, клавиатуры, FSM
- WebApp: создание интерактивных приложений внутри Telegram
- Платежи: полная интеграция с Telegram Stars
- Развертывание: конфигурация, Docker, мониторинг
Ключевые моменты для запоминания:
- Telegram Stars — единственный способ принимать платежи за цифровые товары
- WebApp требует HTTPS и правильной настройки CORS
- Всегда проверяйте данные из WebApp на сервере
- Реализуйте систему возврата средств
- Храните чувствительные данные в переменных окружения
- Используйте базу данных для хранения транзакций и балансов
Для дальнейшего развития проекта рассмотрите:
- Интеграцию с базами данных (PostgreSQL, Redis)
- Систему кеширования для быстрого доступа к данным
- Аналитику платежей и пользовательской активности
- Систему уведомлений для администраторов
- Поддержку нескольких языков
Удачи в разработке! 🚀