Adwanced executor example¶
!/usr/bin/env python3 This example is outdated In this example used ArgumentParser for configuring Your bot. Provided to start bot with webhook: python adwanced_executor_example.py –token TOKEN_HERE –host 0.0.0.0 –port 8084 –host-name example.com –webhook-port 443 Or long polling: python adwanced_executor_example.py –token TOKEN_HERE So… In this example found small trouble: can’t get bot instance in handlers. If you want to automatic change getting updates method use executor utils (from aiogram.utils.executor)
TODO: Move token to environment variables.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | import argparse
import logging
import ssl
import sys
from aiogram import Bot
from aiogram.dispatcher import Dispatcher
from aiogram.dispatcher.webhook import *
from aiogram.utils.executor import start_polling, start_webhook
logging.basicConfig(level=logging.INFO)
# Configure arguments parser.
parser = argparse.ArgumentParser(description='Python telegram bot')
parser.add_argument('--token', '-t', nargs='?', type=str, default=None, help='Set working directory')
parser.add_argument('--sock', help='UNIX Socket path')
parser.add_argument('--host', help='Webserver host')
parser.add_argument('--port', type=int, help='Webserver port')
parser.add_argument('--cert', help='Path to SSL certificate')
parser.add_argument('--pkey', help='Path to SSL private key')
parser.add_argument('--host-name', help='Set webhook host name')
parser.add_argument('--webhook-port', type=int, help='Port for webhook (default=port)')
parser.add_argument('--webhook-path', default='/webhook', help='Port for webhook (default=port)')
async def cmd_start(message: types.Message):
return SendMessage(message.chat.id, f"Hello, {message.from_user.full_name}!")
def setup_handlers(dispatcher: Dispatcher):
# This example has only one messages handler
dispatcher.register_message_handler(cmd_start, commands=['start', 'welcome'])
async def on_startup(dispatcher, url=None, cert=None):
setup_handlers(dispatcher)
bot = dispatcher.bot
# Get current webhook status
webhook = await bot.get_webhook_info()
if url:
# If URL is bad
if webhook.url != url:
# If URL doesnt match with by current remove webhook
if not webhook.url:
await bot.delete_webhook()
# Set new URL for webhook
if cert:
with open(cert, 'rb') as cert_file:
await bot.set_webhook(url, certificate=cert_file)
else:
await bot.set_webhook(url)
elif webhook.url:
# Otherwise remove webhook.
await bot.delete_webhook()
async def on_shutdown(dispatcher):
print('Shutdown.')
def main(arguments):
args = parser.parse_args(arguments)
token = args.token
sock = args.sock
host = args.host
port = args.port
cert = args.cert
pkey = args.pkey
host_name = args.host_name or host
webhook_port = args.webhook_port or port
webhook_path = args.webhook_path
# Fi webhook path
if not webhook_path.startswith('/'):
webhook_path = '/' + webhook_path
# Generate webhook URL
webhook_url = f"https://{host_name}:{webhook_port}{webhook_path}"
# Create bot & dispatcher instances.
bot = Bot(token)
dispatcher = Dispatcher(bot)
if (sock or host) and host_name:
if cert and pkey:
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
ssl_context.load_cert_chain(cert, pkey)
else:
ssl_context = None
start_webhook(dispatcher, webhook_path,
on_startup=functools.partial(on_startup, url=webhook_url, cert=cert),
on_shutdown=on_shutdown,
host=host, port=port, path=sock, ssl_context=ssl_context)
else:
start_polling(dispatcher, on_startup=on_startup, on_shutdown=on_shutdown)
if __name__ == '__main__':
argv = sys.argv[1:]
if not len(argv):
parser.print_help()
sys.exit(1)
main(argv)
|