Telegram-бот, webhook и 50 строк кода

в 23:15, , рубрики: aiohttp, asyncio, python, python 3, telegram, Программирование

Как, опять? Ещё один туториал, пережёвывающий официальную документацию от Telegram, подумали вы?
Да, но нет! Это скорее рассуждения на тему того, как построить функциональный бот-сервис используя Python3.5+, asyncio и aiohttp. Тем интереснее, что заголовок на самом деле лукавит…

Так в чём же лукавство заголовка?
Во-первых, кода не 50 строк, а всего 39, а во-вторых, и бот не такой сложный, просто эхо-бот.

Но, как мне кажется, этого достаточно, чтобы поверить в то, что сделать свой собственный бот-сервис не столь сложно, как может показаться.

Telegram-bot в 39 строк кода 
import asyncio
import aiohttp
from aiohttp import web
import json

TOKEN = '111111111:AAHKeYAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
API_URL = 'https://api.telegram.org/bot%s/sendMessage' % TOKEN

async def handler(request):
    data = await request.json()
    headers = {
        'Content-Type': 'application/json'
    }
    message = {
        'chat_id': data['message']['chat']['id'],
        'text': data['message']['text']
    }
    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.post(API_URL,
                                data=json.dumps(message),
                                headers=headers) as resp:
            try:
                assert resp.status == 200
            except:
                return web.Response(status=500)
    return web.Response(status=200)

async def init_app(loop):
    app = web.Application(loop=loop, middlewares=[])
    app.router.add_post('/api/v1', handler)
    return app

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        app = loop.run_until_complete(init_app(loop))
        web.run_app(app, host='0.0.0.0', port=23456)
    except Exception as e:
        print('Error create server: %r' % e)
    finally:
        pass
    loop.close()

Далее, в нескольких словах, что для чего и как сделать лучше из того, что уже есть.

Содержание:

  1. Что используем
  2. Как используем
  3. Что можно улучшить
  4. Реальный мир

1. Что используем

  • во-первых, Python 3.5+. Почему именно 3.5+, потому что asyncio [2] и потому что сахарные async, await etc;
  • во-вторых, aiohttp. Так как сервис на вебхуках, то он одновременно и HTTP-сервер и HTTP-клиент, а что для этого использовать, как не aiohttp [3];
  • в-третьих, почему webhook, а не long polling? Если не планируется изначально бот-рассыльщик, то интерактивность является его основной функцией. Выскажу своё мнение, что для этой задачи, бот в роли HTTP-сервера подходит лучше, чем в роли клиента. Да, и отдадим часть работы (доставку сообщений) сервисам Telegram.

И ещё, у вас должно быть подконтрольное доменное имя, валидный или самоподписанный сертификат. Доступ к серверу на который указывает доменное имя для настройки реверс-прокси на адрес сервиса.

к содержанию

2. Как используем

Сервер

Состояние библиотеки aiohttp на текущий момент таково, что с её использованием можно построить полноценный web-сервер в Джанго-стиле [4].
Для standalone-сервиса вся мощь не пригодится, поэтому создание сервера ограничивается несколькими строками.

Инициализируем веб приложение

async def init_app(loop):
    app = web.Application(loop=loop, middlewares=[])
    app.router.add_post('/api/v1', handler)
    return app

N.B. Обратите внимание, что здесь мы определяем роутинг и задаём обработчик входящих сообщений handler.

И стартуем веб-сервер:

app = loop.run_until_complete(init_app(loop))
web.run_app(app, host='0.0.0.0', port=23456)

Клиент

Для отправки сообщения используем метод sendMessage из Telegram API, для этого необходимо отправить на оформленный должным образом URL POST-запрос с параметрами в виде JSON-объекта. И это мы делаем с помощью aiohttp:

TOKEN = '111111111:AAHKeYAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
API_URL = 'https://api.telegram.org/bot%s/sendMessage' % TOKEN

...

async def handler(request):
    data = await request.json()
    headers = {
        'Content-Type': 'application/json'
    }
    message = {
        'chat_id': data['message']['chat']['id'],
        'text': data['message']['text']
    }
    async with aiohttp.ClientSession(loop=loop) as session:
        async with session.post(API_URL,
                                data=json.dumps(message),
                                headers=headers) as resp:
            try:
                assert resp.status == 200
            except:
                return web.Response(status=500)
    return web.Response(status=200)

N.B. Обратите внимание, что в случае успешной обработки входящего сообщения и удачной отправки «эха», обработчик возвращает пустой ответ со статусом HTTP 200. Если этого не сделать, сервисы Telegram продолжат в течение какого-то времени «дёргать» запросами хук, либо пока не получат в ответ 200, либо пока не истечёт определённое для сообщения время.

к содержанию

3. Что можно улучшить

Совершенству нет предела, пара идей, как сделать сервис функциональней.

Используем middleware

Допустим, возникла необходимость фильтровать входящие сообщения.
Препроцессинг сообщений можно сделать на специальных веб-обработчиках, в терминах aiohtttp — это middlewares [5].
Пример, определяем мидлварь для игнора сообщений от пользователей из черного списка:

async def middleware_factory(app, handler):
    async def middleware_handler(request):
        data = await request.json()
        if data['message']['from']['id'] in black_list:
            return web.Response(status=200)
        return await handler(request)
    return middleware_handler

и добавляем обработчик при инициализации web-приложения:

async def init_app(loop):
    app = web.Application(loop=loop, middlewares=[])
    app.router.add_post('/api/v1', handler)
    app.middlewares.append(middleware_factory)
    return app

Мысли по поводу обработки входящих сообщений

Если бот будет сложнее, чем репитер-попугай, то можно предложить следующую иерархию объектов Api -> Conversation -> CustomConversation.
Псевдокод:

class Api(object):
    URL = 'https://api.telegram.org/bot%s/%s'

    def __init__(self, token, loop):
        self._token = token
        self._loop = loop

    async def _request(self, method, message):
        headers = {
            'Content-Type': 'application/json'
        }
        async with aiohttp.ClientSession(loop=self._loop) as session:
            async with session.post(self.URL % (self._token, method),
                                    data=json.dumps(message),
                                    headers=headers) as resp:
                try:
                    assert resp.status == 200
                except:
                    pass

    async def sendMessage(self, chatId, text):
        message = {
            'chat_id': chatId,
            'text': text
        }
        await self._request('sendMessage', message)


class Conversation(Api):
    def __init__(self, token, loop):
        super().__init__(token, loop)

    async def _handler(self, message):
        pass

    async def handler(self, request):
        message = await request.json()
        asyncio.ensure_future(self._handler(message['message']))
        return aiohttp.web.Response(status=200)


class EchoConversation(Conversation):
    def __init__(self, token, loop):
        super().__init__(token, loop)

    async def _handler(self, message):
        await self.sendMessage(message['chat']['id'],
                               message['text'])

Наследуя от Conversation и переопределяя _handler получаем кастомные обработчики, в зависимости от функциональности бота — погодный, финансовый etc.

И наш сервис превращается в ферму:

echobot = EchoConversation(TOKEN1, loop)
weatherbot = WeatherConversation(TOKEN2, loop)
finbot = FinanceConversation(TOKEN3, loop)

...

app.router.add_post('/api/v1/echo', echobot.handler)
app.router.add_post('/api/v1/weather', weatherbot.handler)
app.router.add_post('/api/v1/finance', finbot.handler)

к содержанию

4. Реальный мир

Регистрация webhook

Создаём data.json:

{
  "url": "https://bots.domain.tld/api/v1/echo"
}

И вызываем соответствующий метод API любым доступным способом, например:

curl -X POST -d @data.json -H "Content-Type: application/json" "https://api.telegram.org/botYOURBOTTOKEN/setWebhook"

N.B. Ваш домен, хук на который вы устанавливаете, должен резолвится, иначе метод setWebhook не отработает.

Используем прокси-сервер

Как говорит документация: ports currently supported for Webhooks: 443, 80, 88, 8443.
Как же быть в случае self-hosted, когда необходимые порты уже скорее всего заняты веб-сервером, да и соединение по HTTPS мы в нашем сервисе не настроили?
Ответ простой, запуск сервиса на любом доступном локальном интерфейсе и использование реверс-прокси, и лучше nginx здесь сложно найти что-то другое, пусть он возьмёт на себя задачу организации HTTPS-соединения и переадресацию запросов нашему сервису.

к содержанию

Заключение

Надеюсь, что работа с ботом через вебхуки не показалась сильно сложнее long polling, как по мне так даже проще, гибче и прозрачнее. Дополнительные расходы на организацию сервера не должны пугать настоящего ботовода.

Пусть ваши идеи находят достойный инструмент для реализации.

Полезное:

  1. Telegram Bot API
  2. 18.5. asyncio — Asynchronous I/O, event loop, coroutines and tasks
  3. aiohttp: Asynchronous HTTP Client/Server
  4. aiohttp: Server Tutorial
  5. aiohttp: Server Usage — Middlewares

Автор: tmnhy

Источник

* - обязательные к заполнению поля

Главная   |  Архив новостей  |   Android  |   Google  |   Apple  |   Microsoft  |   Информационная безопасность  |   Веб – разработка
Публикации RSS  |  Комментарии RSS
https://ajax.googleapis.com/ajax/libs/jquery/3.4.1/jquery.min.js