Telegram bot, webhook and 50 lines of code
- Tutorial
- Recovery mode
How, again? Another tutorial chewing on the official documentation from Telegram, did you think? Yes but no! This is more of a discussion on how to build a functional bot service using Python3.5 + , asyncio and aiohttp . It’s all the more interesting that the headline is actually
cunning ... So what's the cunning of the heading? Firstly, the code is not 50 lines, but only 39, and secondly, the bot is not so complicated, just an echo bot. But, it seems to me, this is enough to believe that making your own bot service is not as difficult as it might seem.
Further, in a few words, what for what and how to make better of what already exists.
And yet, you must have a controlled domain name, a valid or self-signed certificate. Access to the server to which the domain name points to configure the reverse proxy to the service address.
To the content
The state of the aiohttp library at the moment is such that with its use it is possible to build a full-fledged web server in the Django style [4] .
For standalone- service, all the power is not useful, so the creation of the server is limited to several lines.
We initialize the web application:
NB Please note that here we define the routing and set the handler for the incoming message handler .
And start the web server:
To send a message, we use the sendMessage method from the Telegram API, for this you need to send a POST request with parameters in the form of a JSON object to a properly configured URL. And we do this with aiohttp:
NB Please note that if the incoming message is successfully processed and the echo is sent successfully, the handler returns an empty response with the HTTP status of 200. If this is not done, Telegram services will continue to “pull” hook requests for some time, or until they will receive 200 in response, or until the time specified for the message expires.
To the content
There is no limit to perfection, a couple of ideas on how to make the service more functional.
Suppose you need to filter incoming messages. Message preprocessing can be done on special web handlers; in terms of aiohtttp , these are middlewares [5] .
Example, we define the middleware for ignoring messages from users from the black list:
And add a handler when initializing the web application:
If the bot is more complicated than the parrot repeater, then the following hierarchy of objects Api → Conversation → CustomConversation can be proposed .
Pseudocode:
Inheriting from Conversation and redefining _handler we get custom handlers, depending on the functionality of the bot - weather, financial etc.
And our service turns into a farm:
To the content
Create data.json :
And we call the appropriate API method in any way possible, for example:
NB Your domain, the hook on which you are installing, must resolve, otherwise the setWebhook method will not work.
As the documentation says: ports currently supported for Webhooks: 443, 80, 88, 8443.
What about self-hosted when the necessary ports are most likely already occupied by the web server, and we didn’t configure the HTTPS connection in our service?
The answer is simple, start the service on any available local interface and use a reverse proxy, and it is better for nginx to find something else difficult, let it take on the task of organizing an HTTPS connection and forwarding requests to our service.
To the content
I hope that working with the bot via webhooks did not seem much more complicated than long polling, as for me it is even simpler, more flexible and more transparent. Additional costs for the organization of the server should not scare the real botvoda.
Let your ideas find a worthy tool for implementation.
cunning ... So what's the cunning of the heading? Firstly, the code is not 50 lines, but only 39, and secondly, the bot is not so complicated, just an echo bot. But, it seems to me, this is enough to believe that making your own bot service is not as difficult as it might seem.
Telegram-bot in 39 lines of code
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()
Further, in a few words, what for what and how to make better of what already exists.
Content:
1. What we use
- firstly, Python 3.5+ . Why exactly 3.5+, because asyncio [2] and because sugar async , await etc;
- secondly, aiohttp . Since the service is based on webhooks, it is both an HTTP server and an HTTP client, but what to use for this if not aiohttp [3] ;
- thirdly, why webhook and not long polling ? If you are not originally planning a bot messenger, then interactivity is its main function. I will express my opinion that for this task, a bot in the role of an HTTP server is better than in the role of a client. Yes, and we will give part of the work (message delivery) to Telegram services.
And yet, you must have a controlled domain name, a valid or self-signed certificate. Access to the server to which the domain name points to configure the reverse proxy to the service address.
To the content
2. How to use
Server
The state of the aiohttp library at the moment is such that with its use it is possible to build a full-fledged web server in the Django style [4] .
For standalone- service, all the power is not useful, so the creation of the server is limited to several lines.
We initialize the web application:
async def init_app(loop):
app = web.Application(loop=loop, middlewares=[])
app.router.add_post('/api/v1', handler)
return app
NB Please note that here we define the routing and set the handler for the incoming message handler .
And start the web server:
app = loop.run_until_complete(init_app(loop))
web.run_app(app, host='0.0.0.0', port=23456)
Client
To send a message, we use the sendMessage method from the Telegram API, for this you need to send a POST request with parameters in the form of a JSON object to a properly configured URL. And we do this with 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)
NB Please note that if the incoming message is successfully processed and the echo is sent successfully, the handler returns an empty response with the HTTP status of 200. If this is not done, Telegram services will continue to “pull” hook requests for some time, or until they will receive 200 in response, or until the time specified for the message expires.
To the content
3. What can be improved
There is no limit to perfection, a couple of ideas on how to make the service more functional.
We use middleware
Suppose you need to filter incoming messages. Message preprocessing can be done on special web handlers; in terms of aiohtttp , these are middlewares [5] .
Example, we define the middleware for ignoring messages from users from the black list:
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
And add a handler when initializing the web application:
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
Thoughts on handling incoming messages
If the bot is more complicated than the parrot repeater, then the following hierarchy of objects Api → Conversation → CustomConversation can be proposed .
Pseudocode:
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'])
Inheriting from Conversation and redefining _handler we get custom handlers, depending on the functionality of the bot - weather, financial etc.
And our service turns into a farm:
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)
To the content
4. The real world
Register webhook
Create data.json :
{
"url": "https://bots.domain.tld/api/v1/echo"
}
And we call the appropriate API method in any way possible, for example:
curl -X POST -d @data.json -H "Content-Type: application/json" "https://api.telegram.org/botYOURBOTTOKEN/setWebhook"
NB Your domain, the hook on which you are installing, must resolve, otherwise the setWebhook method will not work.
We use a proxy server
As the documentation says: ports currently supported for Webhooks: 443, 80, 88, 8443.
What about self-hosted when the necessary ports are most likely already occupied by the web server, and we didn’t configure the HTTPS connection in our service?
The answer is simple, start the service on any available local interface and use a reverse proxy, and it is better for nginx to find something else difficult, let it take on the task of organizing an HTTPS connection and forwarding requests to our service.
To the content
Conclusion
I hope that working with the bot via webhooks did not seem much more complicated than long polling, as for me it is even simpler, more flexible and more transparent. Additional costs for the organization of the server should not scare the real botvoda.
Let your ideas find a worthy tool for implementation.