Webhook example old

webhook_example_2.py
  1"""
  2Example outdated
  3"""
  4
  5import asyncio
  6import ssl
  7import sys
  8
  9from aiohttp import web
 10
 11import aiogram
 12from aiogram import Bot, types
 13from aiogram.contrib.fsm_storage.memory import MemoryStorage
 14from aiogram.dispatcher import Dispatcher
 15from aiogram.dispatcher.webhook import get_new_configured_app, SendMessage
 16from aiogram.types import ChatType, ParseMode, ContentTypes
 17from aiogram.utils.markdown import hbold, bold, text, link
 18
 19TOKEN = 'BOT TOKEN HERE'
 20
 21WEBHOOK_HOST = 'example.com'  # Domain name or IP addres which your bot is located.
 22WEBHOOK_PORT = 443  # Telegram Bot API allows only for usage next ports: 443, 80, 88 or 8443
 23WEBHOOK_URL_PATH = '/webhook'  # Part of URL
 24
 25# This options needed if you use self-signed SSL certificate
 26# Instructions: https://core.telegram.org/bots/self-signed
 27WEBHOOK_SSL_CERT = './webhook_cert.pem'  # Path to the ssl certificate
 28WEBHOOK_SSL_PRIV = './webhook_pkey.pem'  # Path to the ssl private key
 29
 30WEBHOOK_URL = f"https://{WEBHOOK_HOST}:{WEBHOOK_PORT}{WEBHOOK_URL_PATH}"
 31
 32# Web app settings:
 33#   Use LAN address to listen webhooks
 34#   User any available port in range from 1024 to 49151 if you're using proxy, or WEBHOOK_PORT if you're using direct webhook handling
 35WEBAPP_HOST = 'localhost'
 36WEBAPP_PORT = 3001
 37
 38BAD_CONTENT = ContentTypes.PHOTO & ContentTypes.DOCUMENT & ContentTypes.STICKER & ContentTypes.AUDIO
 39
 40bot = Bot(TOKEN)
 41storage = MemoryStorage()
 42dp = Dispatcher(bot, storage=storage)
 43
 44
 45async def cmd_start(message: types.Message):
 46    # Yep. aiogram allows to respond into webhook.
 47    # https://core.telegram.org/bots/api#making-requests-when-getting-updates
 48    return SendMessage(chat_id=message.chat.id, text='Hi from webhook!',
 49                       reply_to_message_id=message.message_id)
 50
 51
 52async def cmd_about(message: types.Message):
 53    # In this function markdown utils are userd for formatting message text
 54    return SendMessage(message.chat.id, text(
 55        bold('Hi! I\'m just a simple telegram bot.'),
 56        '',
 57        text('I\'m powered by', bold('Python', Version(*sys.version_info[:]))),
 58        text('With', link(text('aiogram', aiogram.VERSION), 'https://github.com/aiogram/aiogram')),
 59        sep='\n'
 60    ), parse_mode=ParseMode.MARKDOWN)
 61
 62
 63async def cancel(message: types.Message):
 64    # Get current state context
 65    state = dp.current_state(chat=message.chat.id, user=message.from_user.id)
 66
 67    # If current user in any state - cancel it.
 68    if await state.get_state() is not None:
 69        await state.set_state(state=None)
 70        return SendMessage(message.chat.id, 'Current action is canceled.')
 71        # Otherwise do nothing
 72
 73
 74async def unknown(message: types.Message):
 75    """
 76    Handler for unknown messages.
 77    """
 78    return SendMessage(message.chat.id,
 79                       f"I don\'t know what to do with content type `{message.content_type()}`. Sorry :c")
 80
 81
 82async def cmd_id(message: types.Message):
 83    """
 84    Return info about user.
 85    """
 86    if message.reply_to_message:
 87        target = message.reply_to_message.from_user
 88        chat = message.chat
 89    elif message.forward_from and message.chat.type == ChatType.PRIVATE:
 90        target = message.forward_from
 91        chat = message.forward_from or message.chat
 92    else:
 93        target = message.from_user
 94        chat = message.chat
 95
 96    result_msg = [hbold('Info about user:'),
 97                  f"First name: {target.first_name}"]
 98    if target.last_name:
 99        result_msg.append(f"Last name: {target.last_name}")
100    if target.username:
101        result_msg.append(f"Username: {target.mention}")
102    result_msg.append(f"User ID: {target.id}")
103
104    result_msg.extend([hbold('Chat:'),
105                       f"Type: {chat.type}",
106                       f"Chat ID: {chat.id}"])
107    if chat.type != ChatType.PRIVATE:
108        result_msg.append(f"Title: {chat.title}")
109    else:
110        result_msg.append(f"Title: {chat.full_name}")
111    return SendMessage(message.chat.id, '\n'.join(result_msg), reply_to_message_id=message.message_id,
112                       parse_mode=ParseMode.HTML)
113
114
115async def on_startup(app):
116    # Demonstrate one of the available methods for registering handlers
117    # This command available only in main state (state=None)
118    dp.register_message_handler(cmd_start, commands=['start'])
119
120    # This handler is available in all states at any time.
121    dp.register_message_handler(cmd_about, commands=['help', 'about'], state='*')
122    dp.register_message_handler(unknown, content_types=BAD_CONTENT,
123                                func=lambda message: message.chat.type == ChatType.PRIVATE)
124
125    # You are able to register one function handler for multiple conditions
126    dp.register_message_handler(cancel, commands=['cancel'], state='*')
127    dp.register_message_handler(cancel, func=lambda message: message.text.lower().strip() in ['cancel'], state='*')
128
129    dp.register_message_handler(cmd_id, commands=['id'], state='*')
130    dp.register_message_handler(cmd_id, func=lambda message: message.forward_from or
131                                                             message.reply_to_message and
132                                                             message.chat.type == ChatType.PRIVATE, state='*')
133
134    # Get current webhook status
135    webhook = await bot.get_webhook_info()
136
137    # If URL is bad
138    if webhook.url != WEBHOOK_URL:
139        # If URL doesnt match current - remove webhook
140        if not webhook.url:
141            await bot.delete_webhook()
142
143        # Set new URL for webhook
144        await bot.set_webhook(WEBHOOK_URL, certificate=open(WEBHOOK_SSL_CERT, 'rb'))
145        # If you want to use free certificate signed by LetsEncrypt you need to set only URL without sending certificate.
146
147
148async def on_shutdown(app):
149    """
150    Graceful shutdown. This method is recommended by aiohttp docs.
151    """
152    # Remove webhook.
153    await bot.delete_webhook()
154
155    # Close Redis connection.
156    await dp.storage.close()
157    await dp.storage.wait_closed()
158
159
160if __name__ == '__main__':
161    # Get instance of :class:`aiohttp.web.Application` with configured router.
162    app = get_new_configured_app(dispatcher=dp, path=WEBHOOK_URL_PATH)
163
164    # Setup event handlers.
165    app.on_startup.append(on_startup)
166    app.on_shutdown.append(on_shutdown)
167
168    # Generate SSL context
169    context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
170    context.load_cert_chain(WEBHOOK_SSL_CERT, WEBHOOK_SSL_PRIV)
171
172    # Start web-application.
173    web.run_app(app, host=WEBAPP_HOST, port=WEBAPP_PORT, ssl_context=context)
174    # Note:
175    #   If you start your bot using nginx or Apache web server, SSL context is not required.
176    #   Otherwise you need to set `ssl_context` parameter.