Telegram bot, webhook and 50 lines of code

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.

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
  2. How to use
  3. What can be improved
  4. Real world

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 ApiConversationCustomConversation 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.

Useful:


  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

Also popular now: