import asyncio import ssl import sys from aiohttp import web import aiogram from aiogram import Bot, types, Version from aiogram.contrib.fsm_storage.memory import MemoryStorage from aiogram.dispatcher import Dispatcher from aiogram.dispatcher.webhook import get_new_configured_app, SendMessage from aiogram.types import ChatType, ParseMode, ContentType from aiogram.utils.markdown import hbold, bold, text, link TOKEN = 'BOT TOKEN HERE' WEBHOOK_HOST = 'example.com' # Domain name or IP addres which your bot is located. WEBHOOK_PORT = 443 # Telegram Bot API allows only for usage next ports: 443, 80, 88 or 8443 WEBHOOK_URL_PATH = '/webhook' # Part of URL # This options needed if you use self-signed SSL certificate # Instructions: https://core.telegram.org/bots/self-signed WEBHOOK_SSL_CERT = './webhook_cert.pem' # Path to the ssl certificate WEBHOOK_SSL_PRIV = './webhook_pkey.pem' # Path to the ssl private key WEBHOOK_URL = f"https://{WEBHOOK_HOST}:{WEBHOOK_PORT}{WEBHOOK_URL_PATH}" BAD_CONTENT = ContentType.PHOTO & ContentType.DOCUMENT & ContentType.STICKER & ContentType.AUDIO loop = asyncio.get_event_loop() bot = Bot(TOKEN, loop=loop) storage = MemoryStorage() dp = Dispatcher(bot, storage=storage) async def cmd_start(message: types.Message): # Yep. aiogram allow to send response over webhook. # https://core.telegram.org/bots/api#making-requests-when-getting-updates return SendMessage(chat_id=message.chat.id, text='Hi from webhook!', reply_to_message_id=message.message_id) async def cmd_about(message: types.Message): # In this function used markdown utils for formatting message text return SendMessage(message.chat.id, text( bold('Hi! I\'m simple telegram bot.'), '', text('I\'m worked on', bold('Python', Version(*sys.version_info[:]))), text('With', link(text('aiogram', aiogram.VERSION), 'https://bitbucket.org/illemius/aiogram')), sep='\n' ), parse_mode=ParseMode.MARKDOWN) async def cancel(message: types.Message): # Get current state context state = dp.current_state(chat=message.chat.id, user=message.from_user.id) # If now user in any state - cancel it. if await state.get_state() is not None: await state.set_state(state=None) return SendMessage(message.chat.id, 'Current action is canceled.') # Otherwise do nothing async def unknown(message: types.Message): """ Handler for unknown messages. """ return SendMessage(message.chat.id, 'I don\'t know what to do with that content type. Sorry :c') async def cmd_id(message: types.Message): """ Return info about user. """ if message.reply_to_message: target = message.reply_to_message.from_user chat = message.chat elif message.forward_from and message.chat.type == ChatType.PRIVATE: target = message.forward_from chat = message.forward_from or message.chat else: target = message.from_user chat = message.chat result_msg = [hbold('Info about user:'), f'First name: {target.first_name}'] if target.last_name: result_msg.append(f"Last name: {target.last_name}") if target.username: result_msg.append(f"Username: {target.mention}") result_msg.append(f'User ID: {target.id}') result_msg.extend([hbold('Chat:'), f"Type: {chat.type}", f"Chat ID: {chat.id}"]) if chat.type != ChatType.PRIVATE: result_msg.append(f"Title: {chat.title}") else: result_msg.append(f"Title: {chat.full_name}") return SendMessage(message.chat.id, '\n'.join(result_msg), reply_to_message_id=message.message_id, parse_mode=ParseMode.HTML) async def on_startup(app): # Demonstrate one of available method of registering handlers # This command available only in main state (state=None) dp.register_message_handler(cmd_start, commands=['start']) # This handler available in all states in any time. dp.register_message_handler(cmd_about, commands=['help', 'about'], state='*') dp.register_message_handler(unknown, content_types=BAD_CONTENT, func=lambda message: message.chat.type == ChatType.PRIVATE) # You can register one handler with multiple filters set dp.register_message_handler(cancel, commands=['cancel'], state='*') dp.register_message_handler(cancel, func=lambda message: message.text.lower().strip() in ['cancel'], state='*') dp.register_message_handler(cmd_id, commands=['id'], state='*') dp.register_message_handler(cmd_id, func=lambda message: message.forward_from or message.reply_to_message and message.chat.type == ChatType.PRIVATE, state='*') # Get current webhook status webhook = await bot.get_webhook_info() # If URL is bad if webhook.url != WEBHOOK_URL: # If URL doesnt match with by current remove webhook if not webhook.url: await bot.delete_webhook() # Set new URL for webhook await bot.set_webhook(WEBHOOK_URL, certificate=open(WEBHOOK_SSL_CERT, 'rb')) # If you want to use free certificate signed by LetsEncrypt need to set only URL without sending certificate. async def on_shutdown(app): """ Graceful shutdown. This method is recommended by aiohttp doc's. """ # Remove webhook. await bot.delete_webhook() # Close Redis connection. dp.storage.close() await dp.storage.wait_closed() if __name__ == '__main__': # Get instance of :class:`aiohttp.web.Application` with configured router. app = get_new_configured_app(dispatcher=dp, path=WEBHOOK_URL_PATH) # Setup event handlers. app.on_startup.append(on_startup) app.on_shutdown.append(on_shutdown) # Generate SSL context context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) context.load_cert_chain(WEBHOOK_SSL_CERT, WEBHOOK_SSL_PRIV) # Start web-application. web.run_app(app, host=WEBHOOK_HOST, port=WEBHOOK_PORT, ssl_context=context) # Note: # If you should start bot over nginx or Apache web server SSL context is not required here. # Otherwise you need to set that parameter.