Installation
The Basics
Dealing with Inline Query
Class-based Message Handling
Maintain Threads of Conversation
Follow User's Every Action
Inline-only Handler
Async Version (Python 3.4.2 or newer)
Webhook Interface
Deep Linking
Reference
Mailing List
Examples
- Dicey Clock
- Skeletons
- Indoor climate monitor
- IP Cam using Telegram as DDNS
- Emodi - an Emoji Unicode Decoder
- Message Counter
- Guess a number
- Chatbox - a Mailbox for Chats
- User Tracker
- Inline-only Handler
- Pairing Patterns
- Webhooks
- Deep Linking
6.7 (2016-04-03)
- Added a few
TelegramErrorsubclasses to indicate specific errors - Override
BadHTTPResponse's__unicode__()and__str__()to shorten traceback
6.6 (2016-03-20)
- Changed
Answererinterface. Compute function is now passed to methodanswer(), not to the constructor. - Added parameter
disable_notificationto methodssendZZZ(). - Added function
telepot.delegate.per_application()andper_message(). - Used
datato pass POST parameters to prevent too-long query strings on URL - Async version support pushed back to Python 3.4.2
6.5 (2016-02-21)
- Supports file-like object and filename when sending files
- Moved all exceptions to module
telepot.exception - Expanded testing to Python 3.5
6.4 (2016-02-16)
- Introduced automatic message routing to
Bot.handle()andZZZHandler.on_message(). Messages are routed to sub-handlers according to flavor, by default. - As an alternative to implementing
Bot.handle(msg), you may implementBot.on_chat_message(msg),Bot.on_inline_query(msg), andBot.on_chosen_inline_result(msg)as needed. - As an alternative to implementing
ZZZHandler.on_message(), you may implementZZZHandler.on_chat_message(msg),ZZZHandler.on_inline_query(msg), andZZZHandler.on_chosen_inline_result(msg)as needed. notifyOnMessage()andmessageLoop()accept a dict as callback, routing messages according to flavor.- Added function
telepot.flavor_router(), classestelepot.helper.Routerandtelepot.helper.DefaultRouterMixin, and their async counterparts to facilitate message routing. - Many functions in
telepot.delegateandtelepot.helpernow have aliases in their respective async modules, making imports more symmetric.
Telepot has been tested on Raspbian and CentOS, using Python 2.7 - 3.5. Below discussions are based on Raspbian and Python 2.7, except noted otherwise.
pip:
$ sudo pip install telepot
$ sudo pip install telepot --upgrade # UPGRADE
easy_install:
$ easy_install telepot
$ easy_install --upgrade telepot # UPGRADE
Download manually:
$ wget https://pypi.python.org/packages/source/t/telepot/telepot-6.7.zip
$ unzip telepot-6.7.zip
$ cd telepot-6.7
$ python setup.py install
To use the Telegram Bot API, you first have to get a bot account by chatting with the BotFather.
BotFather will give you a token, something like 123456789:ABCdefGhIJKlmNoPQRsTUVwxyZ. With the token in hand, you can start using telepot to access the bot account.
>>> import telepot
>>> bot = telepot.Bot('***** PUT YOUR TOKEN HERE *****')
>>> bot.getMe()
{u'username': u'YourBot', u'first_name': u'Your Bot', u'id': 123456789}Bots cannot initiate conversations with users. You have to send it a message first. It gets the message by calling getUpdates().
>>> from pprint import pprint
>>> response = bot.getUpdates()
>>> pprint(response)
[{u'message': {u'chat': {u'first_name': u'Nick',
u'id': 999999999,
u'last_name': u'Lee',
u'type': u'private'},
u'date': 1444723969,
u'from': {u'first_name': u'Nick',
u'id': 999999999,
u'last_name': u'Lee'},
u'message_id': 4015,
u'text': u'Hello'},
u'update_id': 100000000}]999999999 is obviously a fake ID. Nick Lee is my real name, though.
The chat field indicates the source of the message. Its type can be private, group, or channel (whose meanings should be obvious, I hope). In the above example, Nick Lee just sent a private message to the bot.
I encourage you to experiment sending various types of messages (e.g. voice, photo, sticker, etc) to the bot, via different sources (e.g. private chat, group chat, channel), to see the varying structures of messages. Consult the Bot API documentations to learn what fields may be present under what circumstances. Bot API's object translates directly to Python dict. In the above example, getUpdates() just returns an array of Update objects represented as Python dicts.
Note the update_id. It is an ever-increasing number. Next time you should use getUpdates(offset=100000001) to avoid getting the same old messages over and over. Giving an offset essentially acknowledges to the server that you have received all update_ids lower than offset.
>>> bot.getUpdates(offset=100000001)
[]It is troublesome to keep checking messages and managing offset. Fortunately, telepot can take care of that for you, and notify you whenever new messages arrive.
>>> from pprint import pprint
>>> def handle_message(msg):
... pprint(msg)
...
>>> bot.notifyOnMessage(handle_message)After setting up this callback, sit back and monitor the arriving messages.
Below can be a skeleton for simple telepot programs:
import sys
import time
import pprint
import telepot
def handle(msg):
pprint.pprint(msg)
# Do your stuff here ...
# Getting the token from command-line is better than embedding it in code,
# because tokens are supposed to be kept secret.
TOKEN = sys.argv[1]
bot = telepot.Bot(TOKEN)
bot.notifyOnMessage(handle)
print 'Listening ...'
# Keep the program running.
while 1:
time.sleep(10)When processing a message, a few pieces of information are so central that you almost always have to extract them. Use glance() to extract a tuple of (content_type, chat_type, chat_id) from a message.
content_type can be: text, voice, sticker, photo, audio, document, video, contact, location, new_chat_participant, left_chat_participant, new_chat_title, new_chat_photo, delete_chat_photo, or group_chat_created.
chat_type can be: private, group, or channel.
It is a good habit to always check the content_type before further processing. Do not assume every message is a text.
A better skeleton would look like:
import sys
import time
import telepot
def handle(msg):
content_type, chat_type, chat_id = telepot.glance(msg)
print content_type, chat_type, chat_id
# Do your stuff according to `content_type` ...
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.Bot(TOKEN)
bot.notifyOnMessage(handle)
print 'Listening ...'
# Keep the program running.
while 1:
time.sleep(10)For a voice, sticker, photo, audio, document, or video message, look for the file_id to download the file. For example, a photo may look like:
{u'chat': {u'first_name': u'Nick', u'id': 999999999, u'type': u'private'},
u'date': 1444727788,
u'from': {u'first_name': u'Nick', u'id': 999999999},
u'message_id': 4017,
u'photo': [{u'file_id': u'JiLOABNODdbdP_q2vwXLtLxHFnUxNq2zszIABM4s1rQm6sTYG3QAAgI',
u'file_size': 734,
u'height': 67,
u'width': 90},
{u'file_id': u'JiLOABNODdbdP_q2vwXLtLxHFnUxNq2zszIABJDUyXzQs-kJGnQAAgI',
u'file_size': 4568,
u'height': 240,
u'width': 320},
{u'file_id': u'JiLOABNODdbdP_q2vwXLtLxHFnUxNq2zszIABHWm5IQnJk-EGXQAAgI',
u'file_size': 20480,
u'height': 600,
u'width': 800},
{u'file_id': u'JiLOABNODdbdP_q2vwXLtLxHFnUxNq2zszIABEn8PaFUzRhBGHQAAgI',
u'file_size': 39463,
u'height': 960,
u'width': 1280}]}It has a number of file_ids, with various file sizes. These are thumbnails of the same image. Download one of them by:
>>> bot.downloadFile(u'JiLOABNODdbdP_q2vwXLtLxHFnUxNq2zszIABEn8PaFUzRhBGHQAAgI', 'save/to/path')Enough about receiving messages. Sooner or later, your bot will want to send you messages. You should have discovered your own user ID from above interactions. I will keeping using my fake ID of 999999999. Remember to substitute your own (real) user ID.
>>> bot.sendMessage(999999999, 'Good morning!')After being added as an administrator to a channel, the bot can send messages to the channel:
>>> bot.sendMessage('@channelusername', 'Hi, everybody!')A custom keyboard presents custom buttons for users to tap. Check it out.
>>> show_keyboard = {'keyboard': [['Yes','No'], ['Maybe','Maybe not']]}
>>> bot.sendMessage(999999999, 'This is a custom keyboard', reply_markup=show_keyboard)>>> hide_keyboard = {'hide_keyboard': True}
>>> bot.sendMessage(999999999, 'I am hiding it', reply_markup=hide_keyboard)>>> f = open('zzzzzzzz.jpg', 'rb') # some file on local disk
>>> response = bot.sendPhoto(999999999, f)
>>> pprint(response)
{u'chat': {u'first_name': u'Nick', u'id': 999999999, u'type': u'private'},
u'date': 1444728667,
u'from': {u'first_name': u'Your Bot',
u'id': 123456789,
u'username': u'YourBot'},
u'message_id': 4022,
u'photo': [{u'file_id': u'JiLOABNODdbdPyNZjwa-sKYQW6TBqrWfsztABO2NukbYlhLYlREBAAEC',
u'file_size': 887,
u'height': 29,
u'width': 90},
{u'file_id': u'JiLOABNODdbdPyNZjwa-sKYQW6TBqrWfsztABHKq66Hh0YDrlBEBAAEC',
u'file_size': 9971,
u'height': 102,
u'width': 320},
{u'file_id': u'JiLOABNODdbdPyNZjwa-sKYQW6TBqrWfsztABNBLmxkkiqKikxEBAAEC',
u'file_size': 14218,
u'height': 128,
u'width': 400}]}The server returns a number of file_ids, with various file sizes. These are thumbnails of the uploaded image. If you want to resend the same file, just give one of the file_ids.
>>> bot.sendPhoto(999999999, u'JiLOABNODdbdPyNZjwa-sKYQW6TBqrWfsztABO2NukbYlhLYlREBAAEC')Besides sending photos, you may also sendAudio(), sendDocument(), sendSticker(), sendVideo(), and sendVoice().
By default, a bot only receives messages through a private chat, a group, or a channel. These are what I call normal messages or chat messages.
By sending a /setinline command to BotFather, you enable the bot to receive inline queries as well. Inline query is a way for users to ask your bot questions, even if they have not opened a chat with your bot, nor in the same group with your bot.
The important thing to note is that your bot will now receive two flavors of messages: normal messages and inline queries. A normal message has the flavor normal; an inline query has the flavor inline_query.
flavor = telepot.flavor(msg)
if flavor == 'normal':
print 'Normal message'
elif flavor == 'inline_query':
print 'Inline query'An inline query has this structure (refer to Bot API for the fields' meanings):
{u'from': {u'first_name': u'Nick', u'id': 999999999},
u'id': u'414251975480905552',
u'offset': u'',
u'query': u'abc'}Supply the correct flavor, and glance() extracts some "headline" info about the inline query:
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')The only way to respond to an inline query is to answerInlineQuery(). There are many types of answers you may give back:
- InlineQueryResultArticle
- InlineQueryResultPhoto
- InlineQueryResultGif
- InlineQueryResultMpeg4Gif
- InlineQueryResultVideo
These objects include a variety of fields with various meanings, most of them optional. It is beyond the scope of this document to discuss the effects of those fields. Refer to the links above for details.
As is the custom in telepot, you may construct these results using dictionaries. An alternative is to use the namedtuple classes provided by telepot.namedtuple module.
from telepot.namedtuple import InlineQueryResultArticle
articles = [{'type': 'article',
'id': 'abc', 'title': 'ABC', 'message_text': 'Good morning'},
InlineQueryResultArticle(
id='xyz', title='ZZZZZ', message_text='Good night')]
bot.answerInlineQuery(query_id, articles)from telepot.namedtuple import InlineQueryResultPhoto
photos = [{'type': 'photo',
'id': '123', 'photo_url': '...', 'thumb_url': '...'},
InlineQueryResultPhoto(
id='999', photo_url='...', thumb_url='...')]
bot.answerInlineQuery(query_id, photos)By sending /setinlinefeedback to BotFather, you enable the bot to know which of the provided results your users have chosen. After /setinlinefeedback, your bot will receive one more flavor of messages: chosen_inline_result.
flavor = telepot.flavor(msg)
if flavor == 'normal':
...
elif flavor == 'inline_query':
...
elif flavor == 'chosen_inline_result':
...A chosen inline result has this structure (refer to Bot API for the fields' meanings):
{u'from': {u'first_name': u'Nick', u'id': 999999999},
u'query': u'qqqqq',
u'result_id': u'abc'}The result_id refers to the id you have assigned to a particular answer.
Again, use glance() to extract "headline" info:
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')import sys
import time
import telepot
def handle(msg):
flavor = telepot.flavor(msg)
# normal message
if flavor == 'normal':
content_type, chat_type, chat_id = telepot.glance(msg)
print 'Normal Message:', content_type, chat_type, chat_id
# Do your stuff according to `content_type` ...
# inline query - need `/setinline`
elif flavor == 'inline_query':
query_id, from_id, query_string = telepot.glance(msg, flavor=flavor)
print 'Inline Query:', query_id, from_id, query_string
# Compose your own answers
articles = [{'type': 'article',
'id': 'abc', 'title': 'ABC', 'message_text': 'Good morning'}]
bot.answerInlineQuery(query_id, articles)
# chosen inline result - need `/setinlinefeedback`
elif flavor == 'chosen_inline_result':
result_id, from_id, query_string = telepot.glance(msg, flavor=flavor)
print 'Chosen Inline Result:', result_id, from_id, query_string
# Remember the chosen answer to do better next time
else:
raise telepot.BadFlavor(msg)
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.Bot(TOKEN)
bot.notifyOnMessage(handle)
print 'Listening ...'
# Keep the program running.
while 1:
time.sleep(10)Having always to check the flavor is troublesome. You may supply a routing table to bot.notifyOnMessage() to enable message routing:
bot.notifyOnMessage({'normal': on_chat_message,
'inline_query': on_inline_query,
'chosen_inline_result': on_chosen_inline_result})That results in a more succinct skeleton:
import sys
import time
import telepot
def on_chat_message(msg):
content_type, chat_type, chat_id = telepot.glance(msg)
print 'Chat Message:', content_type, chat_type, chat_id
def on_inline_query(msg):
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')
print 'Inline Query:', query_id, from_id, query_string
# Compose your own answers
articles = [{'type': 'article',
'id': 'abc', 'title': 'ABC', 'message_text': 'Good morning'}]
bot.answerInlineQuery(query_id, articles)
def on_chosen_inline_result(msg):
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')
print 'Chosen Inline Result:', result_id, from_id, query_string
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.Bot(TOKEN)
bot.notifyOnMessage({'normal': on_chat_message,
'inline_query': on_inline_query,
'chosen_inline_result': on_chosen_inline_result})
print 'Listening ...'
# Keep the program running.
while 1:
time.sleep(10)There is one more problem: dealing with inline queries this way is not ideal. As you types and pauses, types and pauses, types and pauses ... closely bunched inline queries arrive. In fact, a new inline query often arrives before we finish processing a preceding one. With only a single thread of execution, we can only process the (closely bunched) inline queries sequentially. Ideally, whenever we see a new inline query from the same user, it should override and cancel any preceding inline queries being processed (that belong to the same user).
An Answerer takes an inline query, inspects its from id (the originating user id), and checks to see whether that user has an unfinished thread processing a preceding inline query. If there is, the unfinished thread will be cancelled before a new thread is spawned to process the latest inline query. In other words, an Answerer ensures at most one active inline-query-processing thread per user.
Answerer also frees you from having to call bot.answerInlineQuery() every time. You supply it with an answer-computing function. It takes that function's returned value and calls bot.answerInlineQuery() to send the results. Being accessible by multiple threads, the answer-computing function must be thread-safe.
answerer = telepot.helper.Answerer(bot)
def on_inline_query(msg):
def compute_answer():
articles = [{'type': 'article',
'id': 'abc', 'title': 'ABC', 'message_text': 'XYZ'}]
return articles
answerer.answer(msg, compute_answer)The skeleton becomes:
import sys
import time
import telepot
def on_chat_message(msg):
content_type, chat_type, chat_id = telepot.glance(msg)
print 'Chat Message:', content_type, chat_type, chat_id
def on_inline_query(msg):
def compute_answer():
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')
print 'Computing for: %s' % query_string
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
answerer.answer(msg, compute_answer)
def on_chosen_inline_result(msg):
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')
print 'Chosen Inline Result:', result_id, from_id, query_string
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.Bot(TOKEN)
answerer = telepot.helper.Answerer(bot)
bot.notifyOnMessage({'normal': on_chat_message,
'inline_query': on_inline_query,
'chosen_inline_result': on_chosen_inline_result})
print 'Listening ...'
# Keep the program running.
while 1:
time.sleep(10)If you use telepot's async version (Python 3.4.2 or newer), you should also use the async version of Answerer. In that case, it will create tasks instead of spawning threads, and you don't have to worry about thread safety.
The proper way to deal with inline query is always through an Answerer's answer() method. If you don't like to use Answerer for some reason, then you should devise your own mechanism to deal with closely-bunched inline queries, always remembering to let a latter one supercede earlier ones. If you decide to go that path, Answerer may be a good starting reference point.
Defining a global message handler may lead to proliferation of global variables quickly. Encapsulation may be achieved by extending the Bot class, defining a handle method, then calling notifyOnMessage() with no callback function. This way, the object's handle method will be used as the callback.
Here is a Python 3 skeleton using this strategy. You may not need Answerer and the blocks dealing with inline_query and chosen_inline_result if you have not /setinline or /setinlinefeedback on the bot.
import sys
import time
import telepot
class YourBot(telepot.Bot):
def __init__(self, *args, **kwargs):
super(YourBot, self).__init__(*args, **kwargs)
self._answerer = telepot.helper.Answerer(self)
def handle(self, msg):
flavor = telepot.flavor(msg)
# normal message
if flavor == 'normal':
content_type, chat_type, chat_id = telepot.glance(msg)
print('Normal Message:', content_type, chat_type, chat_id)
# Do your stuff according to `content_type` ...
# inline query - need `/setinline`
elif flavor == 'inline_query':
query_id, from_id, query_string = telepot.glance(msg, flavor=flavor)
print('Inline Query:', query_id, from_id, query_string)
def compute_answer():
# Compose your own answers
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
self._answerer.answer(msg, compute_answer)
# chosen inline result - need `/setinlinefeedback`
elif flavor == 'chosen_inline_result':
result_id, from_id, query_string = telepot.glance(msg, flavor=flavor)
print('Chosen Inline Result:', result_id, from_id, query_string)
# Remember the chosen answer to do better next time
else:
raise telepot.BadFlavor(msg)
TOKEN = sys.argv[1] # get token from command-line
bot = YourBot(TOKEN)
bot.notifyOnMessage()
print('Listening ...')
# Keep the program running.
while 1:
time.sleep(10)Having always to check the flavor is troublesome. Alternatively, you may implement the method on_chat_message, on_inline_query, and on_chosen_inline_result. The bot will route messages to the correct handler according to flavor.
import sys
import time
import telepot
class YourBot(telepot.Bot):
def __init__(self, *args, **kwargs):
super(YourBot, self).__init__(*args, **kwargs)
self._answerer = telepot.helper.Answerer(self)
def on_chat_message(self, msg):
content_type, chat_type, chat_id = telepot.glance(msg)
print('Normal Message:', content_type, chat_type, chat_id)
def on_inline_query(self, msg):
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')
print('Inline Query:', query_id, from_id, query_string)
def compute_answer():
# Compose your own answers
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
self._answerer.answer(msg, compute_answer)
def on_chosen_inline_result(self, msg):
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')
print('Chosen Inline Result:', result_id, from_id, query_string)
TOKEN = sys.argv[1] # get token from command-line
bot = YourBot(TOKEN)
bot.notifyOnMessage()
print('Listening ...')
# Keep the program running.
while 1:
time.sleep(10)So far, we have been using a single line of execution to handle messages. That is adequate for simple programs. For more sophisticated programs where states need to be maintained across messages, a better approach is needed.
Consider this scenario. A bot wants to have an intelligent conversation with a lot of users, and if we could only use a single line of execution to handle messages (like what we have done so far), we would have to maintain some state variables about each conversation outside the message-handling function(s). On receiving each message, we first have to check whether the user already has a conversation started, and if so, what we have been talking about. To avoid such mundaneness, we need a structured way to maintain "threads" of conversation.
Let's look at my solution. Here, I implemented a bot that counts how many messages have been sent by an individual user. If no message is received after 10 seconds, it starts over (timeout). The counting is done per chat - that's the important point.
import sys
import telepot
from telepot.delegate import per_chat_id, create_open
class MessageCounter(telepot.helper.ChatHandler):
def __init__(self, seed_tuple, timeout):
super(MessageCounter, self).__init__(seed_tuple, timeout)
self._count = 0
def on_chat_message(self, msg):
self._count += 1
self.sender.sendMessage(self._count)
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.DelegatorBot(TOKEN, [
(per_chat_id(), create_open(MessageCounter, timeout=10)),
])
bot.notifyOnMessage(run_forever=True)A DelegatorBot is a Bot with the newfound ability to spawn delegates. Its constructor takes a list of tuples telling it when and how to spawn delegates. In the example above, it is spawning one MessageCounter per chat id.
For every received message, the function per_chat_id() digests it down to a seed - in this case, the chat id. At first, when there is no MessageCounter associated with a seed (chat id), a new MessageCounter is created. Next time, when there is already a MessageCounter associated with the seed (chat id), no new one is needed.
A MessageCounter is only an object encapsulating states; it says nothing about how to spawn a delegate. The function create_open() causes the spawning of a thread. Thread is the default delegation mechanism (that is why I use the verb "spawn"). There is a way to provide your own implementation of threads or other delegation mechanisms. The Chatbox example demonstrates this possibility.
The function create_open() requires the object MessageCounter to meet certain criteria. Being a subclass of ChatHandler, MessageCounter fulfills most of them. The only thing it has to do is implement the method on_chat_message(), which is called whenever a normal (chat) message arrives. How messages are distributed to the correct object is done by telepot. You don't have to worry about that.
There are two styles of extending ChatHandler in terms of which methods to implement/override:
-
You may override
on_message(), which is the first point of contact for every received message, regardless of flavor. If your bot can receive more than one flavor of messages, remember to check the flavor before further processing. If you don't overrideon_message(), it is still the the first point of contact and the default behaviour is to route a message to the appropriate handler according to flavor. Which leads us to the next style ... -
You may implement one or more of
on_chat_message(),on_inline_query(), andon_chosen_inline_result():on_chat_message()is called for anormalmessageon_inline_query()is called for aninline_queryon_chosen_inline_result()is called for achosen_inline_result
You have just seen the second style. And you are going to see the first style in a moment.
The Message Counter example only deals with normal messages. What if you want to maintain states across different flavors of messages? Here is a tracker that follows all messages originating from a user, regardless of flavor.
import sys
import telepot
from telepot.delegate import per_from_id, create_open
class UserTracker(telepot.helper.UserHandler):
def __init__(self, seed_tuple, timeout):
super(UserTracker, self).__init__(seed_tuple, timeout)
# keep track of how many messages of each flavor
self._counts = {'normal': 0,
'inline_query': 0,
'chosen_inline_result': 0}
self._answerer = telepot.helper.Answerer(self.bot)
def on_message(self, msg):
flavor = telepot.flavor(msg)
self._counts[flavor] += 1
print(self.id, ':', self._counts)
# Have to answer inline query to receive chosen result
if flavor == 'inline_query':
def compute_answer():
query_id, from_id, query_string = telepot.glance(msg, flavor=flavor)
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
self._answerer.answer(msg, compute_answer)
TOKEN = sys.argv[1]
bot = telepot.DelegatorBot(TOKEN, [
(per_from_id(), create_open(UserTracker, timeout=20)),
])
bot.notifyOnMessage(run_forever=True)All messages, regardless of flavor, as long as it is originating from a user, would have a from field containing an id. The function per_from_id() digests a message down to its originating user id, thus ensuring there is one and only one UserTracker per user id.
UserTracker, being a subclass of UserHandler, is automatically set up to capture messages originating from a certain user, regardless of flavor. Because the handling logic is similar for all flavors, it overrides on_message() instead of implementing on_ZZZ() separately. Note the use of an Answerer to properly deal with inline queries.
per_from_id() and UserHandler combined, we can track a user's every step.
What if you only care about inline query (and chosen inline result)? Well, here you go ...
import sys
import telepot
from telepot.delegate import per_inline_from_id, create_open
class InlineHandler(telepot.helper.UserHandler):
def __init__(self, seed_tuple, timeout):
super(InlineHandler, self).__init__(seed_tuple, timeout, flavors=['inline_query', 'chosen_inline_result'])
self._answerer = telepot.helper.Answerer(self.bot)
def on_inline_query(self, msg):
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')
print(self.id, ':', 'Inline Query:', query_id, from_id, query_string)
def compute_answer():
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
self._answerer.answer(msg, compute_answer)
def on_chosen_inline_result(self, msg):
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')
print(self.id, ':', 'Chosen Inline Result:', result_id, from_id, query_string)
TOKEN = sys.argv[1]
bot = telepot.DelegatorBot(TOKEN, [
(per_inline_from_id(), create_open(InlineHandler, timeout=10)),
])
bot.notifyOnMessage(run_forever=True)The function per_inline_from_id() digests a message down to its originating user id, but only for inline query and chosen inline result. It ignores normal (chat) messages.
InlineHandler, again, is a subclass of UserHandler. But it specifies which message flavors to capture (in the constructor). In this case, it only cares about inline query and chosen inline result. Then, it implements on_inline_query() and on_chosen_inline_result() to handle incoming messages.
Everything discussed so far assumes traditional Python. That is, network operations are blocking; if you want to serve many users at the same time, some kind of threads are usually needed. Another option is to use an asynchronous or event-driven framework, such as Twisted.
Python 3.4 introduces its own asynchronous architecture, the asyncio module. Telepot supports that, too. If your bot is to serve many people, I strongly recommend doing it asynchronously.
The latest Raspbian (Jessie) comes with Python 3.4.2. If you are using older Raspbian, or if you want to use the latest Python 3, you have to compile it yourself. For Python 3.5.1, follow these steps:
$ sudo apt-get update
$ sudo apt-get upgrade
$ sudo apt-get install libssl-dev openssl libreadline-dev
$ cd ~
$ wget https://www.python.org/ftp/python/3.5.1/Python-3.5.1.tgz
$ tar zxf Python-3.5.1.tgz
$ cd Python-3.5.1
$ ./configure
$ make
$ sudo make install
Finally:
$ sudo pip3.5 install telepot
In case you are not familiar with asynchronous programming, let's start by learning about generators and coroutines:
... why we want asynchronous programming:
... how generators and coroutines are applied to asynchronous programming:
- Understanding Asynchronous IO
- What are the main uses of
yield from? - A Curious Course on Coroutines and Concurrency
... and how an asyncio program is generally structured:
The async version of Bot, SpeakerBot, and DelegatorBot basically mirror the traditional version's. Main differences are:
- blocking methods (e.g.
sendMessage()) are now coroutines, and should be called withyield from - delegation is achieved by coroutine and task
Because of that (and this is true of asynchronous Python in general), a lot of methods will not work in the interactive Python interpreter like regular functions would. They will have to be driven by an event loop.
import sys
import asyncio
import telepot
import telepot.async
def on_chat_message(msg):
content_type, chat_type, chat_id = telepot.glance(msg)
print('Normal Message:', content_type, chat_type, chat_id)
def on_inline_query(msg):
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')
print('Inline Query:', query_id, from_id, query_string)
def compute_answer():
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
answerer.answer(msg, compute_answer)
def on_chosen_inline_result(msg):
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')
print('Chosen Inline Result:', result_id, from_id, query_string)
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.async.Bot(TOKEN)
answerer = telepot.async.helper.Answerer(bot)
loop = asyncio.get_event_loop()
loop.create_task(bot.messageLoop({'normal': on_chat_message,
'inline_query': on_inline_query,
'chosen_inline_result': on_chosen_inline_result}))
print('Listening ...')
loop.run_forever()import sys
import asyncio
import telepot
import telepot.async
class YourBot(telepot.async.Bot):
def __init__(self, *args, **kwargs):
super(YourBot, self).__init__(*args, **kwargs)
self._answerer = telepot.async.helper.Answerer(self)
def on_chat_message(self, msg):
content_type, chat_type, chat_id = telepot.glance(msg)
print('Normal Message:', content_type, chat_type, chat_id)
def on_inline_query(self, msg):
query_id, from_id, query_string = telepot.glance(msg, flavor='inline_query')
print('Inline Query:', query_id, from_id, query_string)
def compute_answer():
articles = [{'type': 'article',
'id': 'abc', 'title': query_string, 'message_text': query_string}]
return articles
self._answerer.answer(msg, compute_answer)
def on_chosen_inline_result(self, msg):
result_id, from_id, query_string = telepot.glance(msg, flavor='chosen_inline_result')
print('Chosen Inline Result:', result_id, from_id, query_string)
TOKEN = sys.argv[1] # get token from command-line
bot = YourBot(TOKEN)
loop = asyncio.get_event_loop()
loop.create_task(bot.messageLoop())
print('Listening ...')
loop.run_forever()I have re-done the MessageCounter example here. Again, it is very similar to the traditional version.
Note: If you are a "long-time" user of telepot and want to change from the old style of implementing on_message() to the new style of implementing on_chat_message() and company, be aware that your handler's superclass is now in the telepot.async.helper module. (before, it was in telepot.helper)
import sys
import asyncio
import telepot
from telepot.async.delegate import per_chat_id, create_open
class MessageCounter(telepot.async.helper.ChatHandler):
def __init__(self, seed_tuple, timeout):
super(MessageCounter, self).__init__(seed_tuple, timeout)
self._count = 0
@asyncio.coroutine
def on_chat_message(self, msg):
self._count += 1
yield from self.sender.sendMessage(self._count)
TOKEN = sys.argv[1] # get token from command-line
bot = telepot.async.DelegatorBot(TOKEN, [
(per_chat_id(), create_open(MessageCounter, timeout=10)),
])
loop = asyncio.get_event_loop()
loop.create_task(bot.messageLoop())
print('Listening ...')
loop.run_forever()So far, we have been using getUpdates() to obtain new messages from Telegram servers - both notifyOnMessage() and messageLoop() call getUpdates() constantly under the hood. Another way to obtain new messages is through webhooks, in which case Telegram servers will send an HTTPS POST request to an URL you specify, containing a JSON-serialized Update object, whenever there is an update for the bot.
Setting up a webhook is more complicated than using getUpdates() because:
- You have to obtain an URL
- You have to obtain and set up an SSL certificate for the URL
- You have to set up a web server to handle the POST requests coming from Telegram servers
For a simple bot application, it is easy to use telepot directly from the web application. To make a smarter bot where you want to leverage telepot's more advanced features (e.g. to maintain separate "threads" of conversation using DelegatorBot and ChatHandler), we need a structured way to bring the web application and telepot together.
Webhook also presents a subtle problem: closely bunched updates may arrive out of order. That is, update_id 1000 may arrive ahead of update_id 999, if the two are issued by Telegram servers very closely. Unless a bot absolutely doesn't care about update order, it will have to re-order them in some way.
Since 5.0, telepot has a mechanism for web applications to interface with easily, and it takes care of re-ordering for you. The mechanism is simple: you call notifyOnMessage() or messageLoop() to initiate the bot's message handling as usual, but with an additional parameter source, which is a queue.
# for Python 2 and 3
try:
from Queue import Queue
except ImportError:
from queue import Queue
def handle(msg):
# ......
bot = telepot.Bot(TOKEN)
update_queue = Queue()
# get updates from queue, not from Telegram servers
bot.notifyOnMessage(handle, source=update_queue)The web application, upon receiving a POST request, dumps the data onto the queue, for the bot to retrieve at the other end. The bot will re-order the updates if necessary. Assuming Flask as the web application framework:
from flask import Flask, request
app = Flask(__name__)
@app.route('/webhook_path', methods=['GET', 'POST'])
def pass_update():
update_queue.put(request.data) # dump data to queue
return 'OK'It is beyond the scope of this document to detail the usage of web frameworks. Please look at the webhook examples for full demonstrations. Remember, you will have to set up the webhook URL, SSL certificate, and web server on your own.
Telegram website's introduction to deep linking may be a bit confusing to beginners. I try to give a clearer explanation here.
-
You have a database of users. Each user has an ID. Suppose you want your Telegram bot to communicate with user
123, but you don't know his Telegramchat_id(which the bot needs in order to send messages to him). How do you "entice" him to talk to the bot, thus revealing hischat_id? You put a link on a web page. -
But the link has to be "personalized". You want each user to press on a slightly different link, in order to distinguish them. One way to do that is to embed user ID in the link. However, user IDs are not something you want to expose, so you generate a (temporary) key associated with a user ID, and embed that key in the link. If user
123has the keyabcde, his personalized link will be:https://telegram.me/<bot_username>?start=abcde -
Someone clicks on the link, and is led to a conversation with your bot. When he presses the START button, your bot will receive a message:
/start abcde -
On receiving that message, the bot sees that
abcdeis associated with user123. Telegramchat_idcan also be extracted from the message. Knowing user123'schat_id, the bot can send him messages afterwards.
Telegram website's introduction refers often to "Memcache", by which they only mean a datastore that remembers key-user ID associations. In a simple experiment, a dictionary or associative array will do. In real world, you may use Memcached (the memory caching software) or a database table.
Here is a tutorial teaching you how to setup a bot on Raspberry Pi. This simple bot does nothing much but accepts two commands:
/roll- reply with a random integer between 1 and 6, like rolling a dice./time- reply with the current time, like a clock.
A starting point for your telepot programs.
Traditional, Simple »
Traditional, Routing table »
Traditional, Class-based »
Async, Simple »
Async, Routing table »
Async, Class-based »
Running on a Raspberry Pi with a few sensors attached, this bot accepts these commands:
/now- Report current temperature, humidity, and pressure/1m- Report every 1 minute/1h- Report every 1 hour/cancel- Cancel reporting
Running on a Raspberry Pi with a camera module attached, this bot accepts these commands:
/open- Open a port through the router to make the video stream accessible, and send you the URL (which includes the router's public IP address)/close- Close the port
Sooner or later, you want your bots to be able to send emoji. You may look up the unicode on the web, or from now on, you may just fire up Telegram and ask Emodi 😊
Traditional version »
Async version »
I am running this bot on a CentOS server. You should be able to talk to it 24/7. Intended for multiple users, the async version is being run.
By the way, I just discovered a Python emoji package. Use it.
Counts number of messages a user has sent. Illustrates the basic usage of DelegateBot and ChatHandler.
Traditional version »
Async version »
- Send the bot anything to start a game.
- The bot randomly picks an integer between 0-99.
- You make a guess.
- The bot tells you to go higher or lower.
- Repeat step 3 and 4, until guess is correct.
This example is able to serve many players at once. It illustrates the usage of DelegateBot and ChatHandler.
Traditional version »
Async version »
- People send messages to your bot.
- Your bot remembers the messages.
- You read the messages later.
It accepts the following commands from you, the owner, only:
/unread- tells you who has sent you messages and how many/next- read next sender's messages
This example can be a starting point for customer support type of bot accounts. For example, customers send questions to a bot account; staff answers the questions behind the scene, makes it look like the bot is answering questions.
It further illustrates the use of DelegateBot and ChatHandler, and how to spawn delegates differently according to the role of users.
This example only handles text messages and stores messages in memory. If the bot is killed, all messages are lost. It is an example after all.
Traditional version »
Async version »
Tracks a user's every actions, including normal messages and inline-related messages.
Traditional version »
Async version »
Only handles a user's inline-related messages.
Traditional version »
Async version »
When using DelegatorBot, each per_ZZZ() function is most sensibly paired with a certain kind of handler. This example demonstrates those patterns.
Traditional version »
Async version »
A few examples from above are duplicated, by using webhooks. For traditional Python, the web frontend used is Flask. For asynchronous Python, aiohttp.
Skeleton + Flask
Message Counter + Flask
Async Skeleton + aiohttp
Async Message Counter + aiohttp
Using Flask as the web frontend, this example serves a web link at <base_url>/link, and sets up a webhook at <base_url>/abc
- Open browser, visit:
<base_url>/link - Click on the link
- On Telegram conversation, click on the START button
- Bot should receive a message:
/start ghijk, whereghijkis the key embedded in the link, and the payload sent along with the/startcommand. You may use this key to identify the user, then his Telegramchat_id.