import base64 import os import hypercorn.asyncio from quart import Quart, render_template_string, request from telethon import TelegramClient, utils from telethon.errors import SessionPasswordNeededError def get_env(name, message): if name in os.environ: return os.environ[name] return input(message) BASE_TEMPLATE = ''' Telethon + Quart {{ content | safe }} ''' PHONE_FORM = '''
Phone (international format):
''' CODE_FORM = '''
Telegram code:
''' PASSWORD_FORM = '''
Telegram password:
''' # Session name, API ID and hash to use; loaded from environmental variables SESSION = os.environ.get('TG_SESSION', 'quart') API_ID = int(get_env('TG_API_ID', 'Enter your API ID: ')) API_HASH = get_env('TG_API_HASH', 'Enter your API hash: ') # Telethon client client = TelegramClient(SESSION, API_ID, API_HASH) client.parse_mode = 'html' # <- Render things nicely phone = None # Quart app app = Quart(__name__) app.secret_key = 'CHANGE THIS TO SOMETHING SECRET' # Helper method to format messages nicely async def format_message(message): if message.photo: content = '{}'.format( base64.b64encode(await message.download_media(bytes)).decode(), message.raw_text ) else: # client.parse_mode = 'html', so bold etc. will work! content = (message.text or '(action message)').replace('\n', '
') return '

{}: {}{}

'.format( utils.get_display_name(message.sender), content, message.date ) # Connect the client before we start serving with Quart @app.before_serving async def startup(): await client.connect() # After we're done serving (near shutdown), clean up the client @app.after_serving async def cleanup(): await client.disconnect() @app.route('/', methods=['GET', 'POST']) async def root(): # We want to update the global phone variable to remember it global phone # Check form parameters (phone/code) form = await request.form if 'phone' in form: phone = form['phone'] await client.send_code_request(phone) if 'code' in form: try: await client.sign_in(code=form['code']) except SessionPasswordNeededError: return await render_template_string(BASE_TEMPLATE, content=PASSWORD_FORM) if 'password' in form: await client.sign_in(password=form['password']) # If we're logged in, show them some messages from their first dialog if await client.is_user_authorized(): # They are logged in, show them some messages from their first dialog dialog = (await client.get_dialogs())[0] result = '

{}

'.format(dialog.title) async for m in client.iter_messages(dialog, 10): result += await(format_message(m)) return await render_template_string(BASE_TEMPLATE, content=result) # Ask for the phone if we don't know it yet if phone is None: return await render_template_string(BASE_TEMPLATE, content=PHONE_FORM) # We have the phone, but we're not logged in, so ask for the code return await render_template_string(BASE_TEMPLATE, content=CODE_FORM) async def main(): await hypercorn.asyncio.serve(app, hypercorn.Config()) # By default, `Quart.run` uses `asyncio.run()`, which creates a new asyncio # event loop. If we create the `TelegramClient` before, `telethon` will # use `asyncio.get_event_loop()`, which is the implicit loop in the main # thread. These two loops are different, and it won't work. # # So, we have to manually pass the same `loop` to both applications to # make 100% sure it works and to avoid headaches. # # To run Quart inside `async def`, we must use `hypercorn.asyncio.serve()` # directly. # # This example creates a global client outside of Quart handlers. # If you create the client inside the handlers (common case), you # won't have to worry about any of this, but it's still good to be # explicit about the event loop. if __name__ == '__main__': client.loop.run_until_complete(main())