Initial commit for inline bot support

This commit is contained in:
Jannes Höke 2016-01-04 17:31:06 +01:00
parent 71f82d57d9
commit e2e98e86db
12 changed files with 818 additions and 15 deletions

View file

@ -0,0 +1,7 @@
telegram.choseninlineresult module
==================================
.. automodule:: telegram.choseninlineresult
:members:
:undoc-members:
:show-inheritance:

View file

@ -0,0 +1,7 @@
telegram.inlinequery module
===========================
.. automodule:: telegram.inlinequery
:members:
:undoc-members:
:show-inheritance:

View file

@ -0,0 +1,7 @@
telegram.inlinequeryresult module
=================================
.. automodule:: telegram.inlinequeryresult
:members:
:undoc-members:
:show-inheritance:

View file

@ -12,6 +12,9 @@ Submodules
telegram.updater
telegram.dispatcher
telegram.jobqueue
telegram.inlinequery
telegram.inlinequeryresult
telegram.choseninlineresult
telegram.chataction
telegram.contact
telegram.document

106
examples/inlinebot.py Normal file
View file

@ -0,0 +1,106 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Simple Bot to reply to Telegram messages
# This program is dedicated to the public domain under the CC0 license.
"""
This Bot uses the Updater class to handle the bot.
First, a few handler functions are defined. Then, those functions are passed to
the Dispatcher and registered at their respective places.
Then, the bot is started and runs until we press Ctrl-C on the command line.
Usage:
Basic inline bot example. Applies different text transformations.
Press Ctrl-C on the command line or send a signal to the process to stop the
bot.
"""
from random import getrandbits
import re
from telegram import Updater, Update, InlineQueryResultArticle, ParseMode
import logging
# Enable logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO)
logger = logging.getLogger(__name__)
# Define a few command handlers. These usually take the two arguments bot and
# update. Error handlers also receive the raised TelegramError object in error.
def start(bot, update):
bot.sendMessage(update.message.chat_id, text='Hi!')
def help(bot, update):
bot.sendMessage(update.message.chat_id, text='Help!')
def escape_markdown(text):
"""Helper function to escape telegram markup symbols"""
escape_chars = '\*_`\['
return re.sub(r'([%s])' % escape_chars, r'\\\1', text)
def inlinequery(bot, update):
if update.inline_query is not None and update.inline_query.query:
query = update.inline_query.query
results = list()
results.append(InlineQueryResultArticle(
id=hex(getrandbits(64))[2:],
title="Caps",
message_text=query.upper()))
results.append(InlineQueryResultArticle(
id=hex(getrandbits(64))[2:],
title="Bold",
message_text="*%s*" % escape_markdown(query),
parse_mode=ParseMode.MARKDOWN))
results.append(InlineQueryResultArticle(
id=hex(getrandbits(64))[2:],
title="Italic",
message_text="_%s_" % escape_markdown(query),
parse_mode=ParseMode.MARKDOWN))
bot.answerInlineQuery(update.inline_query.id, results=results)
def error(bot, update, error):
logger.warn('Update "%s" caused error "%s"' % (update, error))
def main():
# Create the Updater and pass it your bot's token.
updater = Updater("TOKEN")
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.addTelegramCommandHandler("start", start)
dp.addTelegramCommandHandler("help", help)
# on noncommand i.e message - echo the message on Telegram
dp.addTelegramInlineHandler(inlinequery)
# log all errors
dp.addErrorHandler(error)
# Start the Bot
updater.start_polling()
# Block until the user presses Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()
if __name__ == '__main__':
main()

View file

@ -46,6 +46,10 @@ from .nullhandler import NullHandler
from .emoji import Emoji
from .parsemode import ParseMode
from .message import Message
from .inlinequery import InlineQuery
from .choseninlineresult import ChosenInlineResult
from .inlinequeryresult import InlineQueryResultArticle, InlineQueryResultGif,\
InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo
from .update import Update
from .bot import Bot
from .dispatcher import Dispatcher
@ -57,4 +61,7 @@ __all__ = ['Bot', 'Updater', 'Dispatcher', 'Emoji', 'TelegramError',
'ReplyKeyboardMarkup', 'UserProfilePhotos', 'ChatAction',
'Location', 'Contact', 'Video', 'Sticker', 'Document', 'File',
'Audio', 'PhotoSize', 'Chat', 'Update', 'ParseMode', 'Message',
'User', 'TelegramObject', 'NullHandler', 'Voice', 'JobQueue']
'User', 'TelegramObject', 'NullHandler', 'Voice', 'JobQueue',
'InlineQuery', 'ChosenInlineResult', 'InlineQueryResultArticle',
'InlineQueryResultGif', 'InlineQueryResultPhoto',
'InlineQueryResultMpeg4Gif', 'InlineQueryResultVideo']

View file

@ -584,6 +584,56 @@ class Bot(TelegramObject):
return url, data
@log
def answerInlineQuery(self,
inline_query_id,
results,
cache_time=None,
is_personal=None,
next_offset=None):
"""Use this method to reply to an inline query.
Args:
inline_query_id (int):
Unique identifier for answered query
results (list[InlineQueryResult]):
A list of results for the inline query
Keyword Args:
cache_time (Optional[int]): The maximum amount of time the result
of the inline query may be cached on the server
is_personal (Optional[bool]): Pass True, if results may be cached
on the server side only for the user that sent the query. By
default, results may be returned to any user who sends the same
query
next_offset (Optional[str]): Pass the offset that a client should
send in the next query with the same text to receive more
results. Pass an empty string if there are no more results or
if you don't support pagination. Offset length can't exceed 64
bytes.
Returns:
A boolean if answering was successful
"""
url = '%s/answerInlineQuery' % self.base_url
results = [res.to_dict() for res in results]
data = {'inline_query_id': inline_query_id,
'results': results}
if cache_time is not None:
data['cache_time'] = cache_time
if is_personal is not None:
data['is_personal'] = is_personal
if next_offset is not None:
data['next_offset'] = next_offset
result = request.post(url, data)
return result
@log
def getUserProfilePhotos(self,
user_id,

View file

@ -0,0 +1,79 @@
#!/usr/bin/env python
# pylint: disable=R0902,R0912,R0913
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""
This module contains a object that represents a Telegram ChosenInlineResult
"""
from telegram import TelegramObject
class ChosenInlineResult(TelegramObject):
"""This object represents a Telegram ChosenInlineResult.
Note:
* In Python `from` is a reserved word, use `from_user` instead.
Attributes:
result_id (int):
from_user (:class:`telegram.User`):
query (str):
Args:
result_id (int):
from_user (:class:`telegram.User`):
query (str):
"""
def __init__(self,
result_id,
from_user,
query):
# Required
self.result_id = int(result_id)
self.from_user = from_user
self.query = query
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.ChosenInlineResult:
"""
if not data:
return None
return ChosenInlineResult(**data)
def to_dict(self):
"""
Returns:
dict:
"""
data = super(ChosenInlineResult, self).to_dict()
# Required
data['from'] = data.pop('from_user', None)
return data

View file

@ -130,6 +130,7 @@ class Dispatcher:
self.bot = bot
self.update_queue = update_queue
self.telegram_message_handlers = []
self.telegram_inline_handlers = []
self.telegram_command_handlers = {}
self.telegram_regex_handlers = {}
self.string_regex_handlers = {}
@ -236,21 +237,23 @@ class Dispatcher:
handled = True
# Telegram update (regex)
if isinstance(update, Update):
if isinstance(update, Update) and update.message is not None:
self.dispatchRegex(update)
handled = True
# Telegram update (command)
if isinstance(update, Update) \
and update.message.text.startswith('/'):
self.dispatchTelegramCommand(update)
handled = True
# Telegram update (message)
elif isinstance(update, Update):
self.dispatchTelegramMessage(update)
handled = True
# Telegram update (command)
if update.message.text.startswith('/'):
self.dispatchTelegramCommand(update)
# Telegram update (message)
else:
self.dispatchTelegramMessage(update)
handled = True
elif isinstance(update, Update) and \
(update.inline_query is not None or
update.chosen_inline_result is not None):
self.dispatchTelegramInline(update)
handled = True
# Update not recognized
if not handled:
self.dispatchError(update, TelegramError(
@ -268,6 +271,17 @@ class Dispatcher:
self.telegram_message_handlers.append(handler)
def addTelegramInlineHandler(self, handler):
"""
Registers an inline query handler in the Dispatcher.
Args:
handler (function): A function that takes (Bot, Update, *args) as
arguments.
"""
self.telegram_inline_handlers.append(handler)
def addTelegramCommandHandler(self, command, handler):
"""
Registers a command handler in the Dispatcher.
@ -397,6 +411,17 @@ class Dispatcher:
if handler in self.telegram_message_handlers:
self.telegram_message_handlers.remove(handler)
def removeTelegramInlineHandler(self, handler):
"""
De-registers an inline query handler.
Args:
handler (any):
"""
if handler in self.telegram_inline_handlers:
self.telegram_inline_handlers.remove(handler)
def removeTelegramCommandHandler(self, command, handler):
"""
De-registers a command handler.
@ -574,6 +599,17 @@ class Dispatcher:
self.dispatchTo(self.telegram_message_handlers, update)
def dispatchTelegramInline(self, update):
"""
Dispatches an update that contains an inline update.
Args:
update (telegram.Update): The Telegram update that contains the
message.
"""
self.dispatchTo(self.telegram_inline_handlers, update)
def dispatchError(self, update, error):
"""
Dispatches an error.

82
telegram/inlinequery.py Normal file
View file

@ -0,0 +1,82 @@
#!/usr/bin/env python
# pylint: disable=R0902,R0912,R0913
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""This module contains a object that represents a Telegram InlineQuery"""
from telegram import TelegramObject, User
class InlineQuery(TelegramObject):
"""This object represents a Telegram InlineQuery.
Note:
* In Python `from` is a reserved word, use `from_user` instead.
Attributes:
id (int):
from_user (:class:`telegram.User`):
query (str):
offset (str):
Args:
id (int):
from_user (:class:`telegram.User`):
query (str):
offset (str):
"""
def __init__(self,
id,
from_user,
query,
offset):
# Required
self.id = int(id)
self.from_user = from_user
self.query = query
self.offset = offset
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQuery:
"""
if not data:
return None
data['from_user'] = User.de_json(data.pop('from'))
return InlineQuery(**data)
def to_dict(self):
"""
Returns:
dict:
"""
data = super(InlineQuery, self).to_dict()
# Required
data['from'] = data.pop('from_user', None)
return data

View file

@ -0,0 +1,410 @@
#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
"""
This module contains the classes that represent Telegram InlineQueryResults
"""
from telegram import TelegramObject
class InlineQueryResult(TelegramObject):
"""This object represents a Telegram InlineQueryResult.
Attributes:
type (str):
id (str):
Args:
type (str):
id (str):
"""
def __init__(self,
type,
id):
# Required
self.type = str(type)
self.id = str(id)
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQueryResult:
"""
if not data:
return None
return InlineQueryResult(**data)
class InlineQueryResultArticle(InlineQueryResult):
"""This object represents a Telegram InlineQueryResultArticle.
Attributes:
id (str):
title (str):
message_text (str):
parse_mode (str):
disable_web_page_preview (bool):
url (str):
hide_url (bool):
description (str):
thumb_url (str):
thumb_width (int):
thumb_height (int):
Args:
id (str):
title (str):
message_text (str):
Keyword Args:
parse_mode (Optional[str]):
disable_web_page_preview (Optional[bool]):
url (Optional[str]):
hide_url (Optional[bool]):
description (Optional[str]):
thumb_url (Optional[str]):
thumb_width (Optional[int]):
thumb_height (Optional[int]):
"""
def __init__(self,
id,
title,
message_text,
**kwargs):
# Required
super(InlineQueryResultArticle, self).__init__('article', id)
self.title = title
self.message_text = message_text
# Optional
self.parse_mode = kwargs.get('parse_mode', '')
self.disable_web_page_preview = kwargs.get('disable_web_page_preview',
False)
self.url = kwargs.get('url', '')
self.hide_url = kwargs.get('hide_url', False)
self.description = kwargs.get('description', '')
self.thumb_url = kwargs.get('thumb_url', '')
self.parse_mode = kwargs.get('parse_mode', '')
if 'thumb_width' in kwargs:
self.thumb_width = int(kwargs['thumb_width'])
if 'thumb_height' in kwargs:
self.thumb_height = int(kwargs['thumb_height'])
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQueryResultArticle:
"""
if not data:
return None
return InlineQueryResultArticle(**data)
class InlineQueryResultPhoto(InlineQueryResult):
"""This object represents a Telegram InlineQueryResultPhoto.
Attributes:
id (str):
photo_url (str):
mime_type (str):
photo_width (int):
photo_height (int):
thumb_url (str):
title (str):
description (str):
caption (str):
message_text (str):
parse_mode (str):
disable_web_page_preview (bool):
Args:
id (str):
photo_url (str):
Keyword Args:
mime_type (Optional[str]):
photo_width (Optional[int]):
photo_height (Optional[int]):
thumb_url (Optional[str]):
title (Optional[str]):
description (Optional[str]):
caption (Optional[str]):
message_text (Optional[str]):
parse_mode (Optional[str]):
disable_web_page_preview (Optional[bool]):
"""
def __init__(self,
id,
photo_url,
**kwargs):
# Required
super(InlineQueryResultPhoto, self).__init__('photo', id)
self.photo_url = photo_url
# Optional
self.mime_type = kwargs.get('mime_type', 'image/jpeg')
if 'photo_width' in kwargs:
self.photo_width = int(kwargs['photo_width'])
if 'photo_height' in kwargs:
self.photo_height = int(kwargs['photo_height'])
self.thumb_url = kwargs.get('thumb_url', '')
self.title = kwargs.get('title', '')
self.description = kwargs.get('description', '')
self.caption = kwargs.get('caption', '')
self.message_text = kwargs.get('message_text', '')
self.parse_mode = kwargs.get('parse_mode', '')
self.disable_web_page_preview = kwargs.get('disable_web_page_preview',
False)
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQueryResultPhoto:
"""
if not data:
return None
return InlineQueryResultPhoto(**data)
class InlineQueryResultGif(InlineQueryResult):
"""This object represents a Telegram InlineQueryResultGif.
Attributes:
id (str):
gif_url (str):
gif_width (int):
gif_height (int):
thumb_url (str):
title (str):
caption (str):
message_text (str):
parse_mode (str):
disable_web_page_preview (bool):
Args:
id (str):
gif_url (str):
Keyword Args:
gif_width (Optional[int]):
gif_height (Optional[int]):
thumb_url (Optional[str]):
title (Optional[str]):
caption (Optional[str]):
message_text (Optional[str]):
parse_mode (Optional[str]):
disable_web_page_preview (Optional[bool]):
"""
def __init__(self,
id,
gif_url,
**kwargs):
# Required
super(InlineQueryResultGif, self).__init__('gif', id)
self.gif_url = gif_url
# Optional
if 'gif_width' in kwargs:
self.gif_width = int(kwargs['gif_width'])
if 'gif_height' in kwargs:
self.gif_height = int(kwargs['gif_height'])
self.thumb_url = kwargs.get('thumb_url', '')
self.title = kwargs.get('title', '')
self.caption = kwargs.get('caption', '')
self.message_text = kwargs.get('message_text', '')
self.parse_mode = kwargs.get('parse_mode', '')
self.disable_web_page_preview = kwargs.get('disable_web_page_preview',
False)
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQueryResultGif:
"""
if not data:
return None
return InlineQueryResultGif(**data)
class InlineQueryResultMpeg4Gif(InlineQueryResult):
"""This object represents a Telegram InlineQueryResultMpeg4Gif.
Attributes:
id (str):
mpeg4_url (str):
mpeg4_width (int):
mpeg4_height (int):
thumb_url (str):
title (str):
caption (str):
message_text (str):
parse_mode (str):
disable_web_page_preview (bool):
Args:
id (str):
mpeg4_url (str):
Keyword Args:
mpeg4_width (Optional[int]):
mpeg4_height (Optional[int]):
thumb_url (Optional[str]):
title (Optional[str]):
caption (Optional[str]):
message_text (Optional[str]):
parse_mode (Optional[str]):
disable_web_page_preview (Optional[bool]):
"""
def __init__(self,
id,
mpeg4_url,
**kwargs):
# Required
super(InlineQueryResultMpeg4Gif, self).__init__('mpeg4_gif', id)
self.mpeg4_url = mpeg4_url
# Optional
if 'mpeg4_width' in kwargs:
self.mpeg4_width = int(kwargs['mpeg4_width'])
if 'mpeg4_height' in kwargs:
self.mpeg4_height = int(kwargs['mpeg4_height'])
self.thumb_url = kwargs.get('thumb_url', '')
self.title = kwargs.get('title', '')
self.caption = kwargs.get('caption', '')
self.message_text = kwargs.get('message_text', '')
self.parse_mode = kwargs.get('parse_mode', '')
self.disable_web_page_preview = kwargs.get('disable_web_page_preview',
False)
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQueryResultMpeg4Gif:
"""
if not data:
return None
return InlineQueryResultMpeg4Gif(**data)
class InlineQueryResultVideo(InlineQueryResult):
"""This object represents a Telegram InlineQueryResultVideo.
Attributes:
id (str):
video_url (str):
mime_type (str):
video_width (int):
video_height (int):
video_duration (int):
thumb_url (str):
title (str):
description (str):
caption (str):
message_text (str):
parse_mode (str):
disable_web_page_preview (bool):
Args:
id (str):
video_url (str):
mime_type (str):
Keyword Args:
video_width (Optional[int]):
video_height (Optional[int]):
video_duration (Optional[int]):
thumb_url (Optional[str]):
title (Optional[str]):
description (Optional[str]):
caption (Optional[str]):
message_text (Optional[str]):
parse_mode (Optional[str]):
disable_web_page_preview (Optional[bool]):
"""
def __init__(self,
id,
video_url,
mime_type,
**kwargs):
# Required
super(InlineQueryResultVideo, self).__init__('video', id)
self.video_url = video_url
self.mime_type = mime_type
# Optional
if 'video_width' in kwargs:
self.video_width = int(kwargs['video_width'])
if 'video_height' in kwargs:
self.video_height = int(kwargs['video_height'])
if 'video_duration' in kwargs:
self.video_duration = int(kwargs['video_duration'])
self.thumb_url = kwargs.get('thumb_url', '')
self.title = kwargs.get('title', '')
self.description = kwargs.get('description', '')
self.caption = kwargs.get('caption', '')
self.message_text = kwargs.get('message_text', '')
self.parse_mode = kwargs.get('parse_mode', '')
self.disable_web_page_preview = kwargs.get('disable_web_page_preview',
False)
@staticmethod
def de_json(data):
"""
Args:
data (dict):
Returns:
telegram.InlineQueryResultVideo:
"""
if not data:
return None
return InlineQueryResultVideo(**data)

View file

@ -19,7 +19,7 @@
"""This module contains a object that represents a Telegram Update"""
from telegram import Message, TelegramObject
from telegram import Message, TelegramObject, InlineQuery, ChosenInlineResult
class Update(TelegramObject):
@ -28,6 +28,8 @@ class Update(TelegramObject):
Attributes:
update_id (int):
message (:class:`telegram.Message`):
inline_query (:class:`telegram.InlineQuery`):
chosen_inline_result (:class:`telegram.ChosenInlineResult`):
Args:
update_id (int):
@ -35,6 +37,8 @@ class Update(TelegramObject):
Keyword Args:
message (Optional[:class:`telegram.Message`]):
inline_query (Optional[:class:`telegram.InlineQuery`]):
chosen_inline_result (Optional[:class:`telegram.ChosenInlineResult`])
"""
def __init__(self,
update_id,
@ -43,12 +47,14 @@ class Update(TelegramObject):
self.update_id = int(update_id)
# Optionals
self.message = kwargs.get('message')
self.inline_query = kwargs.get('inline_query')
self.chosen_inline_result = kwargs.get('chosen_inline_result')
@staticmethod
def de_json(data):
"""
Args:
data (str):
data (dict):
Returns:
telegram.Update:
@ -56,6 +62,9 @@ class Update(TelegramObject):
if not data:
return None
data['message'] = Message.de_json(data['message'])
data['message'] = Message.de_json(data.get('message'))
data['inline_query'] = InlineQuery.de_json(data.get('inline_query'))
data['chosen_inline_result'] = \
ChosenInlineResult.de_json(data.get('chosen_inline_result'))
return Update(**data)