Примеры разработки ботов для мессенджера Telegram на языке Python
Библиотеки Python для взаимодействия с Telegram Bot API
Какое решение является наиболее эффективным и современным?
python-telegram-bot (версия 20.x) на основе asyncio
Эта библиотека предоставляет полный асинхронный интерфейс, что позволяет эффективно обрабатывать множество запросов одновременно. Она поддерживает все методы Bot API, работу с вебхуками, планировщик задач (JobQueue), встроенные клавиатуры (InlineKeyboard), состояния (ConversationHandler) и многое другое. Установка:
pip install python-telegram-botTelegram python библиотека (библиотека python для telegram)
Пример простого эхо-бота:
import asyncio
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters
async def start(update: Update, context):
await update.message.reply_text("Привет! Я эхо-бот.")
async def echo(update: Update, context):
await update.message.reply_text(update.message.text)
async def main():
app = Application.builder().token("YOUR_TOKEN").build()
app.add_handler(CommandHandler("start", start))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
await app.run_polling()
if __name__ == "__main__":
asyncio.run(main())Python api программа (программа с использованием api на python)
Запуск бота происходит через app.run_polling(). Важно не забыть заменить "YOUR_TOKEN" на реальный токен от BotFather.
Типичные проблемы:
- Ошибка TelegramError: Conflict: terminated by other long poll or webhook - возникает, если одновременно запущено несколько экземпляров бота. Решение: требуется остановить другие процессы или можно использовать вебхук вместо поллинга.
- Проблема с asyncio в среде без главного цикла (например, в Jupyter Notebook) - необходима дополнительная настройка, например, использование nest_asyncio.
Как создать бота с использованием синхронного подхода?
PyTelegramBotAPI (telebot)
Это популярная синхронная библиотека, простая в освоении. Подходит для небольших проектов, где не требуется высокая производительность. Установка:
pip install pyTelegramBotAPI
Пример эхо-бота:
import telebot
bot = telebot.TeleBot("YOUR_TOKEN")
@bot.message_handler(commands=['start'])
def start(message):
bot.reply_to(message, "Привет! Я эхо-бот.")
@bot.message_handler(func=lambda m: True)
def echo(message):
bot.reply_to(message, message.text)
bot.polling()
Бот использует бесконечный цикл bot.polling(). Основное преимущество заключается в простоте. Недостаток - блокировка при длительных операциях (например, загрузка файлов).
Типичные проблемы:
- Ошибка ConnectionError: [Errno 111] Connection refused - часто связана с блокировкой исходящих подключений на сервере. Решение: следует проверить настройки сети или применить прокси.
- При использовании нескольких обработчиков может возникнуть конфликт commands и text. Рекомендуется указывать фильтры явно.
Какая асинхронная библиотека предоставляет встроенную поддержку машины состояний?
aiogram
Библиотека aiogram (версия 3.x) построена на asyncio и включает удобный механизм FSM (Finite State Machine). Широко используется для ботов со сложной логикой. Установка:
pip install aiogram
Пример эхо-бота:
import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
TOKEN = "YOUR_TOKEN"
bot = Bot(token=TOKEN)
dp = Dispatcher()
@dp.message(Command("start"))
async def start(message: types.Message):
await message.answer("Привет! Я эхо-бот.")
@dp.message()
async def echo(message: types.Message):
await message.answer(message.text)
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
aiogram использует декораторы с указанием фильтров. Для запуска вызывается dp.start_polling(bot). Библиотека активно развивается, поддерживает вебхуки, middleware.
Типичные проблемы:
- Ошибка aiogram.exceptions.TelegramRetryAfter - превышение лимитов запросов. Решение: рекомендуется добавить задержку или использовать aiogram.utils.rate_limiter.
- Проблемы с импортом в версии 3.x: некоторые классы перемещены в подмодули. Следует сверяться с документацией.
Как отправлять запросы к Telegram Bot API напрямую без библиотек?
Использование модуля requests
Для минимальных сценариев можно формировать HTTP-запросы вручную. Это даёт полный контроль, но требует обработки всех аспектов протокола. Пример отправки сообщения:
import requests
TOKEN = "YOUR_TOKEN"
URL = f"https://api.telegram.org/bot{TOKEN}/"
def send_message(chat_id, text):
url = URL + "sendMessage"
data = {"chat_id": chat_id, "text": text}
response = requests.post(url, data=data)
return response.json()
print(send_message(123456789, "Привет!"))
Такой подход подходит для одноразовых задач или тестирования. Однако для полноценного бота потребуется реализовать механизм получения обновлений (getUpdates или вебхук), что усложняет код.
Типичные проблемы:
- Ошибка {"ok":false, "error_code":401, "description":"Unauthorized"} - неверный токен. Необходимо проверить правильность.
- Проблема с кодировкой: текст должен быть в UTF-8. requests обрабатывает это автоматически, но при ручном формировании строк могут возникнуть ошибки.
Расширенные примеры работы с Telegram Bot API
Пример 1: Inline-клавиатура и обработка callback (python-telegram-bot)
Создание клавиатуры с кнопками и реакция на нажатие:
import asyncio
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, CallbackQueryHandler
async def start(update: Update, context):
keyboard = [
[InlineKeyboardButton("Кнопка 1", callback_data='btn1'),
InlineKeyboardButton("Кнопка 2", callback_data='btn2')]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text("Выберите опцию:", reply_markup=reply_markup)
async def button_callback(update: Update, context):
query = update.callback_query
await query.answer()
if query.data == 'btn1':
await query.edit_message_text("Вы выбрали Кнопку 1")
else:
await query.edit_message_text("Вы выбрали Кнопку 2")
async def main():
app = Application.builder().token("YOUR_TOKEN").build()
app.add_handler(CommandHandler("start", start))
app.add_handler(CallbackQueryHandler(button_callback))
await app.run_polling()
if __name__ == "__main__":
asyncio.run(main())
Результат: после нажатия /start пользователь видит две кнопки. Нажатие на любую изменяет текст сообщения.
(No output in console, bot running)
Пример 2: Отправка фотографии (aiogram)
Отправка изображения из файла:
import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
TOKEN = "YOUR_TOKEN"
bot = Bot(token=TOKEN)
dp = Dispatcher()
@dp.message(Command("photo"))
async def send_photo(message: types.Message):
with open("example.jpg", "rb") as photo:
await message.answer_photo(photo, caption="Это тестовое фото")
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
Вместо локального файла можно указать URL: types.FSInputFile.from_url("https://example.com/photo.jpg") (в aiogram 3.x). Результат: бот отправляет фото с подписью.
Пример 3: Машина состояний (FSM) на aiogram
Реализация опроса в два шага: имя и возраст:
import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
class Form(StatesGroup):
name = State()
age = State()
TOKEN = "YOUR_TOKEN"
bot = Bot(token=TOKEN)
dp = Dispatcher()
@dp.message(Command("start"))
async def cmd_start(message: types.Message, state: FSMContext):
await state.set_state(Form.name)
await message.answer("Как вас зовут?")
@dp.message(Form.name)
async def process_name(message: types.Message, state: FSMContext):
await state.update_data(name=message.text)
await state.set_state(Form.age)
await message.answer("Сколько вам лет?")
@dp.message(Form.age)
async def process_age(message: types.Message, state: FSMContext):
if not message.text.isdigit():
await message.answer("Пожалуйста, введите число.")
return
data = await state.get_data()
name = data['name']
age = message.text
await message.answer(f"Спасибо, {name}! Вам {age} лет.")
await state.clear()
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
Результат: бот последовательно запрашивает имя и возраст, сохраняя контекст.
Пример 4: Планировщик задач (JobQueue) в python-telegram-bot
Отправка сообщения каждые 10 секунд:
import asyncio
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
chat_id = update.effective_chat.id
context.job_queue.run_repeating(send_message, interval=10, first=0, data=chat_id)
await update.message.reply_text("Запущено! Приходите через 10 секунд.")
async def send_message(context: ContextTypes.DEFAULT_TYPE):
job = context.job
chat_id = job.data
await context.bot.send_message(chat_id, "Прошло 10 секунд!")
async def main():
app = Application.builder().token("YOUR_TOKEN").build()
app.add_handler(CommandHandler("start", start))
await app.run_polling()
if __name__ == "__main__":
asyncio.run(main())
Результат: после команды /start бот начинает отправлять сообщение каждые 10 секунд.
Пример 5: Использование вебхуков с python-telegram-bot (на Flask)
Настройка вебхука для продакшена:
import asyncio
from flask import Flask, request
import telegram
TOKEN = "YOUR_TOKEN"
WEBHOOK_URL = "https://your-domain.com/webhook"
bot = telegram.Bot(token=TOKEN)
app = Flask(__name__)
@app.route("/webhook", methods=["POST"])
def webhook():
update = telegram.Update.de_json(request.get_json(force=True), bot)
if update.message:
update.message.reply_text("Получено!")
return "OK"
if __name__ == "__main__":
bot.set_webhook(url=WEBHOOK_URL)
app.run()
Для production рекомендуется использовать асинхронные серверы, например aiohttp. Результат: бот получает обновления через вебхук.