Примеры разработки ботов для мессенджера Telegram на языке Python

Раздел: Веб-разработка -> Взаимодействие с внешними сервисами

Библиотеки Python для взаимодействия с Telegram Bot API

Какое решение является наиболее эффективным и современным?

python-telegram-bot (версия 20.x) на основе asyncio

Эта библиотека предоставляет полный асинхронный интерфейс, что позволяет эффективно обрабатывать множество запросов одновременно. Она поддерживает все методы Bot API, работу с вебхуками, планировщик задач (JobQueue), встроенные клавиатуры (InlineKeyboard), состояния (ConversationHandler) и многое другое. Установка:

pip install python-telegram-bot

Telegram 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. Результат: бот получает обновления через вебхук.

Библиотека Python для Telegram - comments

En
Telegram python библиотека (python)