629 lines
25 KiB
Python
629 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
|
||
import requests
|
||
try:
|
||
from requests.packages.urllib3 import fields
|
||
format_header_param = fields.format_header_param
|
||
except ImportError:
|
||
format_header_param = None
|
||
import telebot
|
||
from telebot import types
|
||
from telebot import util
|
||
|
||
logger = telebot.logger
|
||
req_session = requests.session()
|
||
|
||
API_URL = "https://api.telegram.org/bot{0}/{1}"
|
||
FILE_URL = "https://api.telegram.org/file/bot{0}/{1}"
|
||
|
||
CONNECT_TIMEOUT = 3.5
|
||
READ_TIMEOUT = 9999
|
||
|
||
|
||
def _make_request(token, method_name, method='get', params=None, files=None, base_url=API_URL):
|
||
"""
|
||
Makes a request to the Telegram API.
|
||
:param token: The bot's API token. (Created with @BotFather)
|
||
:param method_name: Name of the API method to be called. (E.g. 'getUpdates')
|
||
:param method: HTTP method to be used. Defaults to 'get'.
|
||
:param params: Optional parameters. Should be a dictionary with key-value pairs.
|
||
:param files: Optional files.
|
||
:return: The result parsed to a JSON dictionary.
|
||
"""
|
||
request_url = base_url.format(token, method_name)
|
||
logger.debug("Request: method={0} url={1} params={2} files={3}".format(method, request_url, params, files))
|
||
read_timeout = READ_TIMEOUT
|
||
connect_timeout = CONNECT_TIMEOUT
|
||
if files and format_header_param:
|
||
fields.format_header_param = _no_encode(format_header_param)
|
||
if params:
|
||
if 'timeout' in params: read_timeout = params['timeout'] + 10
|
||
if 'connect-timeout' in params: connect_timeout = params['connect-timeout'] + 10
|
||
result = req_session.request(method, request_url, params=params, files=files, timeout=(connect_timeout, read_timeout))
|
||
logger.debug("The server returned: '{0}'".format(result.text.encode('utf8')))
|
||
return _check_result(method_name, result)['result']
|
||
|
||
|
||
def _check_result(method_name, result):
|
||
"""
|
||
Checks whether `result` is a valid API response.
|
||
A result is considered invalid if:
|
||
- The server returned an HTTP response code other than 200
|
||
- The content of the result is invalid JSON.
|
||
- The method call was unsuccessful (The JSON 'ok' field equals False)
|
||
|
||
:raises ApiException: if one of the above listed cases is applicable
|
||
:param method_name: The name of the method called
|
||
:param result: The returned result of the method request
|
||
:return: The result parsed to a JSON dictionary.
|
||
"""
|
||
if result.status_code != 200:
|
||
msg = 'The server returned HTTP {0} {1}. Response body:\n[{2}]' \
|
||
.format(result.status_code, result.reason, result.text.encode('utf8'))
|
||
raise ApiException(msg, method_name, result)
|
||
|
||
try:
|
||
result_json = result.json()
|
||
except:
|
||
msg = 'The server returned an invalid JSON response. Response body:\n[{0}]' \
|
||
.format(result.text.encode('utf8'))
|
||
raise ApiException(msg, method_name, result)
|
||
|
||
if not result_json['ok']:
|
||
msg = 'Error code: {0} Description: {1}' \
|
||
.format(result_json['error_code'], result_json['description'])
|
||
raise ApiException(msg, method_name, result)
|
||
return result_json
|
||
|
||
|
||
def get_me(token):
|
||
method_url = r'getMe'
|
||
return _make_request(token, method_url)
|
||
|
||
|
||
def get_file(token, file_id):
|
||
method_url = r'getFile'
|
||
return _make_request(token, method_url, params={'file_id': file_id})
|
||
|
||
|
||
def download_file(token, file_path):
|
||
url = FILE_URL.format(token, file_path)
|
||
result = req_session.get(url)
|
||
if result.status_code != 200:
|
||
msg = 'The server returned HTTP {0} {1}. Response body:\n[{2}]' \
|
||
.format(result.status_code, result.reason, result.text)
|
||
raise ApiException(msg, 'Download file', result)
|
||
return result.content
|
||
|
||
|
||
def send_message(token, chat_id, text, disable_web_page_preview=None, reply_to_message_id=None, reply_markup=None,
|
||
parse_mode=None, disable_notification=None):
|
||
"""
|
||
Use this method to send text messages. On success, the sent Message is returned.
|
||
:param token:
|
||
:param chat_id:
|
||
:param text:
|
||
:param disable_web_page_preview:
|
||
:param reply_to_message_id:
|
||
:param reply_markup:
|
||
:return:
|
||
"""
|
||
method_url = r'sendMessage'
|
||
payload = {'chat_id': str(chat_id), 'text': text}
|
||
if disable_web_page_preview:
|
||
payload['disable_web_page_preview'] = disable_web_page_preview
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if parse_mode:
|
||
payload['parse_mode'] = parse_mode
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
return _make_request(token, method_url, params=payload, method='post')
|
||
|
||
|
||
def set_webhook(token, url=None, certificate=None, max_connections=None, allowed_updates=None):
|
||
method_url = r'setWebhook'
|
||
payload = {
|
||
'url': url if url else "",
|
||
}
|
||
files = None
|
||
if certificate:
|
||
files = {'certificate': certificate}
|
||
if max_connections:
|
||
payload['max_connections'] = max_connections
|
||
if allowed_updates:
|
||
payload['allowed_updates'] = allowed_updates
|
||
return _make_request(token, method_url, params=payload, files=files)
|
||
|
||
|
||
def delete_webhook(token):
|
||
method_url = r'deleteWebhook'
|
||
return _make_request(token, method_url)
|
||
|
||
|
||
def get_webhook_info(token):
|
||
method_url = r'getWebhookInfo'
|
||
payload = {}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def get_updates(token, offset=None, limit=None, timeout=None, allowed_updates=None):
|
||
method_url = r'getUpdates'
|
||
payload = {}
|
||
if offset:
|
||
payload['offset'] = offset
|
||
if limit:
|
||
payload['limit'] = limit
|
||
if timeout:
|
||
payload['timeout'] = timeout
|
||
if allowed_updates:
|
||
payload['allowed_updates'] = allowed_updates
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def get_user_profile_photos(token, user_id, offset=None, limit=None):
|
||
method_url = r'getUserProfilePhotos'
|
||
payload = {'user_id': user_id}
|
||
if offset:
|
||
payload['offset'] = offset
|
||
if limit:
|
||
payload['limit'] = limit
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def get_chat(token, chat_id):
|
||
method_url = r'getChat'
|
||
payload = {'chat_id': chat_id}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def leave_chat(token, chat_id):
|
||
method_url = r'leaveChat'
|
||
payload = {'chat_id': chat_id}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def get_chat_administrators(token, chat_id):
|
||
method_url = r'getChatAdministrators'
|
||
payload = {'chat_id': chat_id}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def get_chat_members_count(token, chat_id):
|
||
method_url = r'getChatMembersCount'
|
||
payload = {'chat_id': chat_id}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def get_chat_member(token, chat_id, user_id):
|
||
method_url = r'getChatMember'
|
||
payload = {'chat_id': chat_id, 'user_id': user_id}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def forward_message(token, chat_id, from_chat_id, message_id, disable_notification=None):
|
||
method_url = r'forwardMessage'
|
||
payload = {'chat_id': chat_id, 'from_chat_id': from_chat_id, 'message_id': message_id}
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def send_photo(token, chat_id, photo, caption=None, reply_to_message_id=None, reply_markup=None,
|
||
disable_notification=None):
|
||
method_url = r'sendPhoto'
|
||
payload = {'chat_id': chat_id}
|
||
files = None
|
||
if not util.is_string(photo):
|
||
files = {'photo': photo}
|
||
else:
|
||
payload['photo'] = photo
|
||
if caption:
|
||
payload['caption'] = caption
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
return _make_request(token, method_url, params=payload, files=files, method='post')
|
||
|
||
|
||
def send_location(token, chat_id, latitude, longitude, reply_to_message_id=None, reply_markup=None,
|
||
disable_notification=None):
|
||
method_url = r'sendLocation'
|
||
payload = {'chat_id': chat_id, 'latitude': latitude, 'longitude': longitude}
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def send_venue(token, chat_id, latitude, longitude, title, address, foursquare_id=None, disable_notification=None,
|
||
reply_to_message_id=None, reply_markup=None):
|
||
method_url = r'sendVenue'
|
||
payload = {'chat_id': chat_id, 'latitude': latitude, 'longitude': longitude, 'title': title, 'address': address}
|
||
if foursquare_id:
|
||
payload['foursquare_id'] = foursquare_id
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def send_contact(token, chat_id, phone_number, first_name, last_name=None, disable_notification=None,
|
||
reply_to_message_id=None, reply_markup=None):
|
||
method_url = r'sendContact'
|
||
payload = {'chat_id': chat_id, 'phone_number': phone_number, 'first_name': first_name}
|
||
if last_name:
|
||
payload['last_name'] = last_name
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def send_chat_action(token, chat_id, action):
|
||
method_url = r'sendChatAction'
|
||
payload = {'chat_id': chat_id, 'action': action}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def send_video(token, chat_id, data, duration=None, caption=None, reply_to_message_id=None, reply_markup=None,
|
||
disable_notification=None, timeout=None):
|
||
method_url = r'sendVideo'
|
||
payload = {'chat_id': chat_id}
|
||
files = None
|
||
if not util.is_string(data):
|
||
files = {'video': data}
|
||
else:
|
||
payload['video'] = data
|
||
if duration:
|
||
payload['duration'] = duration
|
||
if caption:
|
||
payload['caption'] = caption
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if timeout:
|
||
payload['connect-timeout'] = timeout
|
||
return _make_request(token, method_url, params=payload, files=files, method='post')
|
||
|
||
|
||
def send_voice(token, chat_id, voice, caption=None, duration=None, reply_to_message_id=None, reply_markup=None,
|
||
disable_notification=None, timeout=None):
|
||
method_url = r'sendVoice'
|
||
payload = {'chat_id': chat_id}
|
||
files = None
|
||
if not util.is_string(voice):
|
||
files = {'voice': voice}
|
||
else:
|
||
payload['voice'] = voice
|
||
if caption:
|
||
payload['caption'] = caption
|
||
if duration:
|
||
payload['duration'] = duration
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if timeout:
|
||
payload['connect-timeout'] = timeout
|
||
return _make_request(token, method_url, params=payload, files=files, method='post')
|
||
|
||
|
||
def send_video_note(token, chat_id, data, duration=None, length=None, reply_to_message_id=None, reply_markup=None,
|
||
disable_notification=None, timeout=None):
|
||
method_url = r'sendVideoNote'
|
||
payload = {'chat_id': chat_id}
|
||
files = None
|
||
if not util.is_string(data):
|
||
files = {'video_note': data}
|
||
files['video_note']['length'] = 639 # seems like it is MAX length size
|
||
else:
|
||
payload['video_note'] = data
|
||
if duration:
|
||
payload['duration'] = duration
|
||
if length:
|
||
payload['length'] = length
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if timeout:
|
||
payload['connect-timeout'] = timeout
|
||
return _make_request(token, method_url, params=payload, files=files, method='post')
|
||
|
||
|
||
def send_audio(token, chat_id, audio, caption=None, duration=None, performer=None, title=None, reply_to_message_id=None,
|
||
reply_markup=None, disable_notification=None, timeout=None):
|
||
method_url = r'sendAudio'
|
||
payload = {'chat_id': chat_id}
|
||
files = None
|
||
if not util.is_string(audio):
|
||
files = {'audio': audio}
|
||
else:
|
||
payload['audio'] = audio
|
||
if caption:
|
||
payload['caption'] = caption
|
||
if duration:
|
||
payload['duration'] = duration
|
||
if performer:
|
||
payload['performer'] = performer
|
||
if title:
|
||
payload['title'] = title
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if timeout:
|
||
payload['connect-timeout'] = timeout
|
||
return _make_request(token, method_url, params=payload, files=files, method='post')
|
||
|
||
|
||
def send_data(token, chat_id, data, data_type, reply_to_message_id=None, reply_markup=None, disable_notification=None,
|
||
timeout=None, caption=None):
|
||
method_url = get_method_by_type(data_type)
|
||
payload = {'chat_id': chat_id}
|
||
files = None
|
||
if not util.is_string(data):
|
||
files = {data_type: data}
|
||
else:
|
||
payload[data_type] = data
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if timeout:
|
||
payload['connect-timeout'] = timeout
|
||
if caption:
|
||
payload['caption'] = caption
|
||
return _make_request(token, method_url, params=payload, files=files, method='post')
|
||
|
||
|
||
def get_method_by_type(data_type):
|
||
if data_type == 'document':
|
||
return r'sendDocument'
|
||
if data_type == 'sticker':
|
||
return r'sendSticker'
|
||
|
||
|
||
def kick_chat_member(token, chat_id, user_id):
|
||
method_url = 'kickChatMember'
|
||
payload = {'chat_id': chat_id, 'user_id': user_id}
|
||
return _make_request(token, method_url, params=payload, method='post')
|
||
|
||
|
||
def unban_chat_member(token, chat_id, user_id):
|
||
method_url = 'unbanChatMember'
|
||
payload = {'chat_id': chat_id, 'user_id': user_id}
|
||
return _make_request(token, method_url, params=payload, method='post')
|
||
|
||
|
||
# Updating messages
|
||
|
||
def edit_message_text(token, text, chat_id=None, message_id=None, inline_message_id=None, parse_mode=None,
|
||
disable_web_page_preview=None, reply_markup=None):
|
||
method_url = r'editMessageText'
|
||
payload = {'text': text}
|
||
if chat_id:
|
||
payload['chat_id'] = chat_id
|
||
if message_id:
|
||
payload['message_id'] = message_id
|
||
if inline_message_id:
|
||
payload['inline_message_id'] = inline_message_id
|
||
if parse_mode:
|
||
payload['parse_mode'] = parse_mode
|
||
if disable_web_page_preview:
|
||
payload['disable_web_page_preview'] = disable_web_page_preview
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def edit_message_caption(token, caption, chat_id=None, message_id=None, inline_message_id=None, reply_markup=None):
|
||
method_url = r'editMessageCaption'
|
||
payload = {'caption': caption}
|
||
if chat_id:
|
||
payload['chat_id'] = chat_id
|
||
if message_id:
|
||
payload['message_id'] = message_id
|
||
if inline_message_id:
|
||
payload['inline_message_id'] = inline_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def edit_message_reply_markup(token, chat_id=None, message_id=None, inline_message_id=None, reply_markup=None):
|
||
method_url = r'editMessageReplyMarkup'
|
||
payload = {}
|
||
if chat_id:
|
||
payload['chat_id'] = chat_id
|
||
if message_id:
|
||
payload['message_id'] = message_id
|
||
if inline_message_id:
|
||
payload['inline_message_id'] = inline_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
def delete_message(token, chat_id=None, message_id=None):
|
||
method_url = r'deleteMessage'
|
||
payload = {'chat_id': chat_id, 'message_id': message_id}
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
# Game
|
||
|
||
def send_game(token, chat_id, game_short_name, disable_notification=None, reply_to_message_id=None, reply_markup=None):
|
||
method_url = r'sendGame'
|
||
payload = {'chat_id': chat_id, 'game_short_name': game_short_name}
|
||
if disable_notification:
|
||
payload['disable_notification'] = disable_notification
|
||
if reply_to_message_id:
|
||
payload['reply_to_message_id'] = reply_to_message_id
|
||
if reply_markup:
|
||
payload['reply_markup'] = _convert_markup(reply_markup)
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
# https://core.telegram.org/bots/api#setgamescore
|
||
def set_game_score(token, user_id, score, force=None, disable_edit_message=None, chat_id=None, message_id=None,
|
||
inline_message_id=None):
|
||
"""
|
||
Use this method to set the score of the specified user in a game. On success, if the message was sent by the bot, returns the edited Message, otherwise returns True. Returns an error, if the new score is not greater than the user's current score in the chat.
|
||
:param token: Bot's token (you don't need to fill this)
|
||
:param user_id: User identifier
|
||
:param score: New score, must be non-negative
|
||
:param force: (Optional) Pass True, if the high score is allowed to decrease. This can be useful when fixing mistakes or banning cheaters
|
||
:param disable_edit_message: (Optional) Pass True, if the game message should not be automatically edited to include the current scoreboard
|
||
:param chat_id: (Optional, required if inline_message_id is not specified) Unique identifier for the target chat (or username of the target channel in the format @channelusername)
|
||
:param message_id: (Optional, required if inline_message_id is not specified) Unique identifier of the sent message
|
||
:param inline_message_id: (Optional, required if chat_id and message_id are not specified) Identifier of the inline message
|
||
:return:
|
||
"""
|
||
method_url = r'setGameScore'
|
||
payload = {'user_id': user_id, 'score': score}
|
||
if force:
|
||
payload['force'] = force
|
||
if chat_id:
|
||
payload['chat_id'] = chat_id
|
||
if message_id:
|
||
payload['message_id'] = message_id
|
||
if inline_message_id:
|
||
payload['inline_message_id'] = inline_message_id
|
||
if disable_edit_message:
|
||
payload['disable_edit_message'] = disable_edit_message
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
# https://core.telegram.org/bots/api#getgamehighscores
|
||
def get_game_high_scores(token, user_id, chat_id=None, message_id=None, inline_message_id=None):
|
||
"""
|
||
Use this method to get data for high score tables. Will return the score of the specified user and several of his neighbors in a game. On success, returns an Array of GameHighScore objects.
|
||
This method will currently return scores for the target user, plus two of his closest neighbors on each side. Will also return the top three users if the user and his neighbors are not among them. Please note that this behavior is subject to change.
|
||
:param token: Bot's token (you don't need to fill this)
|
||
:param user_id: Target user id
|
||
:param chat_id: (Optional, required if inline_message_id is not specified) Unique identifier for the target chat (or username of the target channel in the format @channelusername)
|
||
:param message_id: (Optional, required if inline_message_id is not specified) Unique identifier of the sent message
|
||
:param inline_message_id: (Optional, required if chat_id and message_id are not specified) Identifier of the inline message
|
||
:return:
|
||
"""
|
||
method_url = r'getGameHighScores'
|
||
payload = {'user_id': user_id}
|
||
if chat_id:
|
||
payload['chat_id'] = chat_id
|
||
if message_id:
|
||
payload['message_id'] = message_id
|
||
if inline_message_id:
|
||
payload['inline_message_id'] = inline_message_id
|
||
return _make_request(token, method_url, params=payload)
|
||
|
||
|
||
# InlineQuery
|
||
|
||
def answer_callback_query(token, callback_query_id, text=None, show_alert=None, url=None, cache_time=None):
|
||
"""
|
||
Use this method to send answers to callback queries sent from inline keyboards. The answer will be displayed to the user as a notification at the top of the chat screen or as an alert. On success, True is returned.
|
||
Alternatively, the user can be redirected to the specified Game URL. For this option to work, you must first create a game for your bot via BotFather and accept the terms. Otherwise, you may use links like telegram.me/your_bot?start=XXXX that open your bot with a parameter.
|
||
:param token: Bot's token (you don't need to fill this)
|
||
:param callback_query_id: Unique identifier for the query to be answered
|
||
:param text: (Optional) Text of the notification. If not specified, nothing will be shown to the user, 0-200 characters
|
||
:param show_alert: (Optional) If true, an alert will be shown by the client instead of a notification at the top of the chat screen. Defaults to false.
|
||
:param url: (Optional) URL that will be opened by the user's client. If you have created a Game and accepted the conditions via @Botfather, specify the URL that opens your game – note that this will only work if the query comes from a callback_game button.
|
||
Otherwise, you may use links like telegram.me/your_bot?start=XXXX that open your bot with a parameter.
|
||
:param cache_time: (Optional) The maximum amount of time in seconds that the result of the callback query may be cached client-side. Telegram apps will support caching starting in version 3.14. Defaults to 0.
|
||
:return:
|
||
"""
|
||
method_url = 'answerCallbackQuery'
|
||
payload = {'callback_query_id': callback_query_id}
|
||
if text:
|
||
payload['text'] = text
|
||
if show_alert:
|
||
payload['show_alert'] = show_alert
|
||
if url:
|
||
payload['url'] = url
|
||
if cache_time:
|
||
payload['cache_time'] = cache_time
|
||
return _make_request(token, method_url, params=payload, method='post')
|
||
|
||
|
||
def answer_inline_query(token, inline_query_id, results, cache_time=None, is_personal=None, next_offset=None,
|
||
switch_pm_text=None, switch_pm_parameter=None):
|
||
method_url = 'answerInlineQuery'
|
||
payload = {'inline_query_id': inline_query_id, 'results': _convert_inline_results(results)}
|
||
if cache_time:
|
||
payload['cache_time'] = cache_time
|
||
if is_personal:
|
||
payload['is_personal'] = is_personal
|
||
if next_offset is not None:
|
||
payload['next_offset'] = next_offset
|
||
if switch_pm_text:
|
||
payload['switch_pm_text'] = switch_pm_text
|
||
if switch_pm_parameter:
|
||
payload['switch_pm_parameter'] = switch_pm_parameter
|
||
return _make_request(token, method_url, params=payload, method='post')
|
||
|
||
|
||
def _convert_inline_results(results):
|
||
ret = ''
|
||
for r in results:
|
||
if isinstance(r, types.JsonSerializable):
|
||
ret = ret + r.to_json() + ','
|
||
if len(ret) > 0:
|
||
ret = ret[:-1]
|
||
return '[' + ret + ']'
|
||
|
||
|
||
def _convert_markup(markup):
|
||
if isinstance(markup, types.JsonSerializable):
|
||
return markup.to_json()
|
||
return markup
|
||
|
||
|
||
def _no_encode(func):
|
||
def wrapper(key, val):
|
||
if key == 'filename':
|
||
return '{0}={1}'.format(key, val)
|
||
else:
|
||
return func(key, val)
|
||
return wrapper
|
||
|
||
|
||
class ApiException(Exception):
|
||
"""
|
||
This class represents an Exception thrown when a call to the Telegram API fails.
|
||
In addition to an informative message, it has a `function_name` and a `result` attribute, which respectively
|
||
contain the name of the failed function and the returned result that made the function to be considered as
|
||
failed.
|
||
"""
|
||
|
||
def __init__(self, msg, function_name, result):
|
||
super(ApiException, self).__init__("A request to the Telegram API was unsuccessful. {0}".format(msg))
|
||
self.function_name = function_name
|
||
self.result = result
|