Как, опять? Ещё один туториал, пережёвывающий официальную документацию от Telegram, подумали вы? Да, но нет! Это скорее рассуждения на тему того, как построить функциональный бот-сервис используя Python3.5+, asyncio и aiohttp. Тем интереснее, что заголовок на самом деле лукавит…
Так в чём же лукавство заголовка? Во-первых, кода не 50 строк, а всего 39, а во-вторых, и бот не такой сложный, просто эхо-бот. Но, как мне кажется, этого достаточно, чтобы поверить в то, что сделать свой собственный бот-сервис не столь сложно, как может показаться.
Далее, в нескольких словах, что для чего и как сделать лучше из того, что уже есть.
И ещё, у вас должно быть подконтрольное доменное имя, валидный или самоподписанный сертификат. Доступ к серверу на который указывает доменное имя для настройки реверс-прокси на адрес сервиса.
К содержанию
Состояние библиотеки aiohttp на текущий момент таково, что с её использованием можно построить полноценный web-сервер в Джанго-стиле [4].
Для standalone-сервиса вся мощь не пригодится, поэтому создание сервера ограничивается несколькими строками.
Инициализируем веб-приложение:
N.B. Обратите внимание, что здесь мы определяем роутинг и задаём обработчик входящих сообщений handler.
И стартуем веб-сервер:
Для отправки сообщения используем метод sendMessage из Telegram API, для этого необходимо отправить на оформленный должным образом URL POST-запрос с параметрами в виде JSON-объекта. И это мы делаем с помощью aiohttp:
N.B. Обратите внимание, что в случае успешной обработки входящего сообщения и удачной отправки «эха», обработчик возвращает пустой ответ со статусом HTTP 200. Если этого не сделать, сервисы Telegram продолжат в течение какого-то времени «дёргать» запросами хук, либо пока не получат в ответ 200, либо пока не истечёт определённое для сообщения время.
К содержанию
Совершенству нет предела, пара идей, как сделать сервис функциональней.
Допустим, возникла необходимость фильтровать входящие сообщения. Препроцессинг сообщений можно сделать на специальных веб-обработчиках, в терминах aiohtttp — это middlewares [5].
Пример, определяем мидлварь для игнора сообщений от пользователей из черного списка:
И добавляем обработчик при инициализации web-приложения:
Если бот будет сложнее, чем репитер-попугай, то можно предложить следующую иерархию объектов Api → Conversation → CustomConversation.
Псевдокод:
Наследуя от Conversation и переопределяя _handler получаем кастомные обработчики, в зависимости от функциональности бота — погодный, финансовый etc.
И наш сервис превращается в ферму:
К содержанию
Создаём data.json:
И вызываем соответствующий метод API любым доступным способом, например:
N.B. Ваш домен, хук на который вы устанавливаете, должен резолвится, иначе метод setWebhook не отработает.
Как говорит документация: ports currently supported for Webhooks: 443, 80, 88, 8443.
Как же быть в случае self-hosted, когда необходимые порты уже скорее всего заняты веб-сервером, да и соединение по HTTPS мы в нашем сервисе не настроили?
Ответ простой, запуск сервиса на любом доступном локальном интерфейсе и использование реверс-прокси, и лучше nginx здесь сложно найти что-то другое, пусть он возьмёт на себя задачу организации HTTPS-соединения и переадресацию запросов нашему сервису.
К содержанию
Надеюсь, что работа с ботом через вебхуки не показалась сильно сложнее long polling, как по мне так даже проще, гибче и прозрачнее. Дополнительные расходы на организацию сервера не должны пугать настоящего ботовода.
Пусть ваши идеи находят достойный инструмент для реализации.
Так в чём же лукавство заголовка? Во-первых, кода не 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. Что используем
- во-первых, 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, как по мне так даже проще, гибче и прозрачнее. Дополнительные расходы на организацию сервера не должны пугать настоящего ботовода.
Пусть ваши идеи находят достойный инструмент для реализации.
