-
-
Notifications
You must be signed in to change notification settings - Fork 802
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
ce0bbed
commit f42f4d0
Showing
1 changed file
with
166 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,166 @@ | ||
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. |