Какие знаешь основные шаги при написании бота?

Основные шаги при написании бота

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

1. Определение требований и архитектуры

Перед написанием кода нужно:

• Определить функционал бота
• Выбрать платформу (Telegram, Discord, Slack и т.д.)
• Спроектировать пользовательские сценарии
• Выбрать архитектуру: polling vs webhooks

# Пример: простой бот-помощник
BOT_NAME = "My Helper Bot"
BOT_VERSION = "1.0.0"

FEATURES = {
    "greeting": "Приветствие пользователя",
    "help": "Справка по командам",
    "echo": "Повтор сообщения"
}

2. Выбор фреймворка и установка зависимостей

Для Telegram наиболее популярны:

• aiogram — асинхронный, современный
• python-telegram-bot — синхронный, стабильный
• pyTelegramBotAPI — простой для начинающих

Основные зависимости:

pip install aiogram python-dotenv aiohttp

3. Получение API ключа и создание конфигурации

Всегда используй переменные окружения, не хардкодь токены:

import os
from dotenv import load_dotenv

load_dotenv()

BOT_TOKEN = os.getenv("BOT_TOKEN")
ADMIN_ID = int(os.getenv("ADMIN_ID", 0))
API_URL = os.getenv("API_URL", "https://api.example.com")

# Валидация
if not BOT_TOKEN:
    raise ValueError("BOT_TOKEN not set in .env")

4. Инициализация бота и диспетчера

Для aiogram 3.x:

from aiogram import Bot, Dispatcher, F
from aiogram.types import Message
from aiogram.filters.command import Command
import logging

# Логирование
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Инициализация
bot = Bot(token=BOT_TOKEN)
dispatcher = Dispatcher()

5. Написание обработчиков команд

Обработчик — это функция, которая реагирует на события:

@dispatcher.message(Command("start"))
async def cmd_start(message: Message):
    """Обработчик команды /start"""
    await message.answer(
        f"Привет, {message.from_user.first_name}!\n"
        "Я помощник бот. Используй /help для справки."
    )

@dispatcher.message(Command("help"))
async def cmd_help(message: Message):
    """Обработчик команды /help"""
    help_text = """
    Доступные команды:
    /start - Начать работу
    /help - Эта справка
    /about - Информация о боте
    """
    await message.answer(help_text)

@dispatcher.message()
async def echo_handler(message: Message):
    """Обработчик всех текстовых сообщений"""
    await message.answer(f"Вы написали: {message.text}")

6. Работа с состояниями (FSM)

Для многошагового взаимодействия используй машину состояний:

from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import ReplyKeyboardMarkup, KeyboardButton

class FormStates(StatesGroup):
    waiting_for_name = State()
    waiting_for_email = State()
    waiting_for_confirmation = State()

@dispatcher.message(Command("form"))
async def start_form(message: Message, state: FSMContext):
    await state.set_state(FormStates.waiting_for_name)
    await message.answer("Как вас зовут?")

@dispatcher.message(FormStates.waiting_for_name)
async def process_name(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(FormStates.waiting_for_email)
    await message.answer("Укажите вашу почту:")

@dispatcher.message(FormStates.waiting_for_email)
async def process_email(message: Message, state: FSMContext):
    await state.update_data(email=message.text)
    data = await state.get_data()
    await message.answer(
        f"Итого:\nИмя: {data['name']}\nПочта: {data['email']}"
    )
    await state.clear()

7. Работа с кнопками и меню

from aiogram.types import (
    ReplyKeyboardMarkup, KeyboardButton,
    InlineKeyboardMarkup, InlineKeyboardButton
)

# Обычные кнопки
reply_keyboard = ReplyKeyboardMarkup(
    keyboard=[
        [KeyboardButton(text="Опция 1")],
        [KeyboardButton(text="Опция 2")],
        [KeyboardButton(text="Опция 3")]
    ],
    resize_keyboard=True,
    one_time_keyboard=True
)

@dispatcher.message()
async def send_menu(message: Message):
    await message.answer("Выберите действие:", reply_markup=reply_keyboard)

# Встроенные кнопки
inline_keyboard = InlineKeyboardMarkup(
    inline_keyboard=[
        [
            InlineKeyboardButton(text="Ссылка", url="https://example.com"),
            InlineKeyboardButton(text="Callback", callback_data="action_1")
        ]
    ]
)

@dispatcher.callback_query(F.data == "action_1")
async def handle_callback(callback_query):
    await callback_query.answer("Нажата кнопка!")
    await callback_query.message.edit_text("Кнопка нажата")

8. Обработка ошибок

from aiogram import BaseMiddleware
from aiogram.types import Update

class ErrorHandlerMiddleware(BaseMiddleware):
    async def __call__(self, handler, event: Update, data):
        try:
            return await handler(event, data)
        except Exception as e:
            logger.error(f"Error processing update: {e}")
            # Отправить админу уведомление об ошибке
            return None

dispatcher.message.middleware(ErrorHandlerMiddleware())

9. Подключение к базе данных

Если нужно хранить данные:

import asyncpg

class Database:
    def __init__(self, dsn: str):
        self.dsn = dsn
        self.pool = None
    
    async def connect(self):
        self.pool = await asyncpg.create_pool(self.dsn)
    
    async def disconnect(self):
        await self.pool.close()
    
    async def add_user(self, user_id: int, username: str):
        query = "INSERT INTO users (user_id, username) VALUES ($1, $2)"
        async with self.pool.acquire() as conn:
            await conn.execute(query, user_id, username)

db = Database(os.getenv("DATABASE_URL"))

10. Запуск бота

async def main():
    # Инициализация
    await bot.session.close()
    
    # Выбор способа получения обновлений
    # Вариант 1: Polling (опрос)
    await dispatcher.start_polling(bot)
    
    # Вариант 2: Webhooks (требует SSL и публичного URL)
    # from aiogram.types import Update
    # webhook_url = "https://example.com/webhook"
    # await bot.set_webhook(webhook_url)

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

11. Логирование и мониторинг

import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler(f"logs/bot_{datetime.now().date()}.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

@dispatcher.message()
async def log_message(message: Message):
    logger.info(f"User {message.from_user.id}: {message.text}")
    await echo_handler(message)

12. Развертывание

Для продакшена:

# Используй supervisor или systemd для фонового запуска
# Пример supervisor конфига:
[program:mybot]
command=python /path/to/bot.py
autostart=true
autorestart=true
user=bot
stderr_logfile=/var/log/mybot.err.log
stdout_logfile=/var/log/mybot.out.log

Полный пример минимального бота

import asyncio
from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.filters.command import Command
import os
from dotenv import load_dotenv

load_dotenv()

BOT_TOKEN = os.getenv("BOT_TOKEN")
bot = Bot(token=BOT_TOKEN)
dispatcher = Dispatcher()

@dispatcher.message(Command("start"))
async def start_handler(message: Message):
    await message.answer("Привет!")

@dispatcher.message()
async def echo(message: Message):
    await message.answer(message.text)

async def main():
    await dispatcher.start_polling(bot)

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

Резюме основных шагов

1. Требования — функционал и архитектура
2. Зависимости — установка фреймворка
3. Конфигурация — токены в .env
4. Инициализация — создание Bot и Dispatcher
5. Обработчики — логика команд и сообщений
6. Состояния — FSM для многошагового взаимодействия
7. UI — кнопки и клавиатуры
8. Ошибки — обработка исключений
9. Данные — база данных при необходимости
10. Запуск — polling или webhooks
11. Логирование — отслеживание ошибок
12. Развертывание — продакшн-окружение

В своей практике я также использую тесты (pytest с mocks), контейнеризацию (Docker) и CI/CD для автоматизации развертывания.