Building a Telegram client in Python

Telegram is an application for encrypted chats that has gained enormous popularity. Fully open source, it has a complete API both for endpoints (TDLIB) and one for creating Bots.

C++, Java and .NET are supported, plus an API that they call JSON which supposedly can be used from multiple languages.

Since I wanted to investigate the framework itself, and not the language APIs, I opted to choose a Python-library called python-telegram. There is also a variant for Bots called (surprisingly) python-telegram-bot.

Installing the library is as always with Python a simple story:

# optionally create a virtual environment
python -m venv venv
venv/bin/activate
# install the library
pip install python-telegram

Next you must create yourself an api_id and and api_hash. You do that by going to https://my.telegram.org and registering an account. Enter an App Title and a Short Name. Copy the api_id and api_hash, you will use them in your app later.

The application

from telegram.client import Telegram
tg = Telegram(
    api_id='<the_api_id>',
    api_hash='<the_api_hash>',
    phone='+467012345678',
    database_encryption_key='mysecretencryptionkey',)
tg.login()

Define a helper method for sending messages

def send_message(chat_id, msg):
    result = tg.send_message(
        chat_id=chat_id,
        text=msg,
    )
    result.wait()

Message structure

This is the structure of the most common message, a simple text message. The most imoprtant fields are emphasized.

{
    "@type": "updateNewMessage",
    "message": {
        "@type": "message",
        "id": <the-message-id>,
        "sender": {
            "@type": "messageSenderUser",
            "user_id": <the-user-id>
        },
        "chat_id": <the-chat-id>,
        "is_outgoing": False,
        "is_pinned": False,
        "can_be_edited": False,
        "can_be_forwarded": True,
        "can_be_deleted_only_for_self": True,
        "can_be_deleted_for_all_users": True,
        "can_get_statistics": False,
        "can_get_message_thread": False,
        "can_get_viewers": False,
        "can_get_media_timestamp_links": False,
        "has_timestamped_media": True,
        "is_channel_post": False,
        "contains_unread_mention": False,
        "date": 1657206884,
        "edit_date": 0,
        "reply_in_chat_id": 0,
        "reply_to_message_id": 0,
        "message_thread_id": 0,
        "ttl": 0,
        "ttl_expires_in": 0.0,
        "via_bot_user_id": 0,
        "author_signature": "",
        "media_album_id": "0",
        "restriction_reason": "",
        "content": {
            "@type": "messageText",
            "text": {
                "@type": "formattedText",
                "text": "Hello, world!",
                "entities": []
            }
        }
    }
}

Handler for incoming messages

# this function will be called
# for each received message
def new_message_handler(msg):
    
    # You could inspect msg['message']['@type'] here!

    if msg['message']['is_outgoing'] == True:
        print("(outgoing message suppressed...)")
    else:
        # we want to process only text messages
        message_content = msg['message']['content'].get('text', {})
        message_text = message_content.get('text', '').lower()
        chat_id = msg['message']['chat_id']

        print(f'{message_text} has been received from {chat_id}')

        if message_text.startswith("//"):
            if message_text == "//help":
                send_message(chat_id, "type //help to get help.")
                send_message(chat_id, "type //bands to get a list of good rock bands.")
            elif message_text == "//bands":
                send_message(chat_id, "Status Quo, Kiss and Aerosmith is a good start!")
            else:
                send_message(chat_id, "Sorry dont understand that! Type //help to get help")

It is very important that you check whether the “is_outgoing” is set, because messages that you send yourself will also arrive in this callback. So if you auto-respond to messages you may enter an infinite loop unless you skip messages with the “is_outgoing” is not set.

You add the message handler like this:

tg.add_message_handler(new_message_handler)

if this is the first run, library needs to preload all chats otherwise the message will not be sent

result = tg.get_chats()
result.wait()

Last, setup a message loop

print("Waiting for incoming messages now...")
tg.idle()

Leave a Reply

Your email address will not be published. Required fields are marked *

This site uses Akismet to reduce spam. Learn how your comment data is processed.