👩‍🎤 🔇 🌞 電報ボット、webhook、および50行のコード 😤 🙄 🦂

どうして？ Telegramの公式ドキュメントを噛む別のチュートリアルはありますか？はい、いいえ！これは、 Python3.5 + 、 asyncio 、およびaiohttpを使用して機能的なボットサービスを構築する方法についての議論です。さらに興味深いことに、見出しは実際にはcです...

それでは、見出しのcは何ですか？第一に、コードは50行ではなく39行であり、第二に、ボットはそれほど複雑ではなく、単なるエコーボットです。しかし、私には、これはあなた自身のボットサービスを作るのが見た目ほど難しくないことを信じるのに十分なようです。

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+です。 asyncio [2]やシュガーasync 、 awaitなどのために、正確に3.5+である理由。
次に、 aiohttp 。サービスはwebhookに基づいているため、HTTPサーバーとHTTPクライアントの両方ですが、aiohttpでない場合にこれに使用するもの[3] ;
第三に、なぜwebhookで長いポーリングではないのですか？ボットスパマーが当初計画されていない場合、その主な機能は対話機能です。このタスクでは、HTTPサーバーの役割のボットはクライアントの役割よりも優れているとの意見を表明します。はい。作業の一部（メッセージ配信）を電報サービスに提供します。

それでも、制御されたドメイン名、有効な証明書または自己署名証明書が必要です。ドメイン名が指すサーバーへのアクセスは、サービスアドレスへのリバースプロキシを構成します。

コンテンツへ

2.使用方法

サーバー

現時点でのaiohttpライブラリの状態は、その使用により、本格的なWebサーバーをDjangoスタイルで構築できるようになっています[4] 。

スタンドアロンサービスの場合、すべての電力が役に立たないため、サーバーの作成は数行に制限されます。

Webアプリケーションを初期化します。

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

NBここで、ルーティングを定義し、着信メッセージハンドラのハンドラを設定することに注意してください。

Webサーバーを起動します。

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

お客様

メッセージを送信するには、Telegram APIのsendMessageメソッドを使用します。このため、適切に構成されたURLにJSONオブジェクトの形式のパラメーターを含むPOSTリクエストを送信する必要があります。そして、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)

注着信メッセージが正常に処理され、エコーが正常に送信されると、ハンドラーはHTTPステータス200の空の応答を返します。これが行われない場合、テレグラムサービスはしばらくの間、または応答として、またはメッセージに指定された時間が経過するまで、200を受け取ります。

コンテンツへ

3.改善できるもの

完璧に制限はありません。サービスをより機能的にする方法に関するいくつかのアイデアがあります。

ミドルウェアを使用します

着信メッセージをフィルタリングする必要があるとします。メッセージの前処理は、特別なWebハンドラーで実行できます。aiohtttpに関しては、 ミドルウェアです [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'])

会話から継承し、 _handlerを再定義すると、ボットの機能（天気、金融など）に応じて、カスタムハンドラーが取得されます。

そして、私たちのサービスは農場に変わります：

 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"

注：インストールするフックであるドメインは解決する必要があります。解決しないと、setWebhookメソッドが機能しません。

プロキシサーバーを使用します

ドキュメントが言うように： Webhookで現在サポートされているポート ： 443、80、88、8443。

必要なポートがすでにWebサーバーによって占有されている可能性が高く、サービスでHTTPS接続を構成していない場合、セルフホストの場合はどうすればよいですか？

答えは簡単です。利用可能なローカルインターフェースでサービスを開始し、リバースプロキシを使用します。nginxが他の困難なものを見つけて、HTTPS接続を整理し、サービスにリクエストを転送するタスクを引き継ぐことをお勧めします。

コンテンツへ

おわりに

ウェブフックを介してボットを操作することは、長いポーリングよりもはるかに複雑に見えないことを願っています。私にとっては、よりシンプルで、より柔軟で、より透明です。サーバーの組織の追加コストは、実際のボットボーダを怖がらせるべきではありません。

あなたのアイデアが実装にふさわしいツールを見つけてください。

電報ボット、webhook、および50行のコード

内容：