mirror of
https://github.com/python-telegram-bot/python-telegram-bot.git
synced 2024-12-22 14:35:00 +01:00
move isFileRequest to InputFile as @staticmethod, new tests written, README updated
This commit is contained in:
leandrotoledo
parent
7d8fcc5fb9
commit
431510e02f
3 changed files with 82 additions and 66 deletions
|
|
@ -533,27 +533,6 @@ class Bot(object):
|
|||
|
||||
return True
|
||||
|
||||
def _isFileRequest(self,
|
||||
data=None):
|
||||
"""Check if the request is a file request
|
||||
Args:
|
||||
data:
|
||||
A dict of (str, unicode) key/value pairs
|
||||
|
||||
Returns:
|
||||
bool
|
||||
"""
|
||||
if data:
|
||||
file_types = ['audio', 'document', 'photo', 'video']
|
||||
file_type = [i for i in data.keys() if i in file_types]
|
||||
|
||||
if file_type:
|
||||
file_content = data[file_type[0]]
|
||||
return isinstance(file_content, file) or \
|
||||
str(file_content).startswith('http')
|
||||
|
||||
return False
|
||||
|
||||
def _requestUrl(self,
|
||||
url,
|
||||
method,
|
||||
|
|
@ -574,7 +553,7 @@ class Bot(object):
|
|||
|
||||
if method == 'POST':
|
||||
try:
|
||||
if self._isFileRequest(data):
|
||||
if InputFile.is_inputfile(data):
|
||||
data = InputFile(data)
|
||||
|
||||
request = urllib2.Request(
|
||||
|
|
@ -619,25 +598,11 @@ class Bot(object):
|
|||
|
||||
try:
|
||||
data = json.loads(json_data.decode())
|
||||
self._checkForTelegramError(data)
|
||||
if not data['ok']:
|
||||
raise TelegramError(data['description'])
|
||||
except ValueError:
|
||||
if '<title>403 Forbidden</title>' in json_data:
|
||||
raise TelegramError({'message': 'API must be authenticated'})
|
||||
raise TelegramError({'message': 'JSON decoding'})
|
||||
|
||||
return data['result']
|
||||
|
||||
def _checkForTelegramError(self,
|
||||
data):
|
||||
"""Raises a TelegramError if Telegram returns an error message.
|
||||
|
||||
Args:
|
||||
data:
|
||||
A Python dict created from the Telegram JSON response.
|
||||
|
||||
Raises:
|
||||
TelegramError wrapping the Telegram error message if one exists.
|
||||
"""
|
||||
|
||||
if not data['ok']:
|
||||
raise TelegramError(data['description'])
|
||||
|
|
|
|||
|
|
@ -7,6 +7,12 @@ import os
|
|||
import re
|
||||
import urllib2
|
||||
|
||||
from .error import TelegramError
|
||||
|
||||
DEFAULT_MIME_TYPE = 'application/octet-stream'
|
||||
USER_AGENT = 'Python Telegram Bot' \
|
||||
' (https://github.com/leandrotoledo/python-telegram-bot)'
|
||||
|
||||
|
||||
class InputFile(object):
|
||||
def __init__(self,
|
||||
|
|
@ -31,17 +37,16 @@ class InputFile(object):
|
|||
self.input_file_content = self.input_file.read()
|
||||
self.filename = os.path.basename(self.input_file.name)
|
||||
self.mimetype = mimetypes.guess_type(self.filename)[0] or \
|
||||
'application/octet-stream'
|
||||
DEFAULT_MIME_TYPE
|
||||
|
||||
if 'http' in self.input_file:
|
||||
self.input_file_content = urllib2.urlopen(self.input_file).read()
|
||||
self.mimetype = InputFile.image_type(self.input_file_content)
|
||||
self.mimetype = InputFile.is_image(self.input_file_content)
|
||||
self.filename = self.mimetype.replace('/', '.')
|
||||
|
||||
|
||||
|
||||
@property
|
||||
def headers(self):
|
||||
return {'User-agent': 'Python Telegram Bot (https://github.com/leandrotoledo/python-telegram-bot)',
|
||||
return {'User-agent': USER_AGENT,
|
||||
'Content-type': self.content_type}
|
||||
|
||||
@property
|
||||
|
|
@ -66,7 +71,7 @@ class InputFile(object):
|
|||
form_boundary,
|
||||
str('Content-Disposition: form-data; name="%s"; filename="%s"' % (
|
||||
self.input_name, self.filename
|
||||
)),
|
||||
)),
|
||||
'Content-Type: %s' % self.mimetype,
|
||||
'',
|
||||
self.input_file_content
|
||||
|
|
@ -78,15 +83,54 @@ class InputFile(object):
|
|||
return '\r\n'.join(form)
|
||||
|
||||
@staticmethod
|
||||
def image_type(stream):
|
||||
header = stream[:10]
|
||||
def is_image(stream):
|
||||
"""Check if the content file is an image by analyzing its headers.
|
||||
|
||||
if re.match(r'GIF8', header):
|
||||
return 'image/gif'
|
||||
Args:
|
||||
stream:
|
||||
A str representing the content of a file.
|
||||
|
||||
if re.match(r'\x89PNG', header):
|
||||
return 'image/png'
|
||||
Returns:
|
||||
The str mimetype of an image.
|
||||
"""
|
||||
try:
|
||||
header = stream[:10]
|
||||
|
||||
if re.match(r'\xff\xd8\xff\xe0\x00\x10JFIF', header) or \
|
||||
re.match(r'\xff\xd8\xff\xe1(.*){2}Exif', header):
|
||||
return 'image/jpeg'
|
||||
if re.match(r'GIF8', header):
|
||||
return 'image/gif'
|
||||
|
||||
if re.match(r'\x89PNG', header):
|
||||
return 'image/png'
|
||||
|
||||
if re.match(r'\xff\xd8\xff\xe0\x00\x10JFIF', header) or \
|
||||
re.match(r'\xff\xd8\xff\xe1(.*){2}Exif', header):
|
||||
return 'image/jpeg'
|
||||
except IndexError as e:
|
||||
raise TelegramError(str(e))
|
||||
|
||||
raise TelegramError({'message': 'Could not parse file content'})
|
||||
|
||||
@staticmethod
|
||||
def is_inputfile(data):
|
||||
"""Check if the request is a file request
|
||||
Args:
|
||||
data:
|
||||
A dict of (str, unicode) key/value pairs
|
||||
|
||||
Returns:
|
||||
bool
|
||||
"""
|
||||
if data:
|
||||
file_types = ['audio', 'document', 'photo', 'video']
|
||||
file_type = [i for i in data.keys() if i in file_types]
|
||||
|
||||
if file_type:
|
||||
file_content = data[file_type[0]]
|
||||
|
||||
if file_type[0] == 'photo':
|
||||
return isinstance(file_content, file) or \
|
||||
str(file_content).startswith('http')
|
||||
|
||||
return isinstance(file_content, file)
|
||||
|
||||
return False
|
||||
|
|
|
|||
|
|
@ -16,11 +16,11 @@ class BotTest(unittest.TestCase):
|
|||
def testGetMe(self):
|
||||
'''Test the telegram.Bot getMe method'''
|
||||
print('Testing getMe')
|
||||
user = self._bot.getMe()
|
||||
self.assertEqual(120405045, user.id)
|
||||
self.assertEqual('Toledo\'s Palace Bot', user.first_name)
|
||||
self.assertEqual(None, user.last_name)
|
||||
self.assertEqual('ToledosPalaceBot', user.username)
|
||||
bot = self._bot.getMe()
|
||||
self.assertEqual(120405045, bot.id)
|
||||
self.assertEqual('Toledo\'s Palace Bot', bot.first_name)
|
||||
self.assertEqual(None, bot.last_name)
|
||||
self.assertEqual('ToledosPalaceBot', bot.username)
|
||||
|
||||
def testSendMessage(self):
|
||||
'''Test the telegram.Bot sendMessage method'''
|
||||
|
|
@ -33,7 +33,7 @@ class BotTest(unittest.TestCase):
|
|||
'''Test the telegram.Bot getUpdates method'''
|
||||
print('Testing getUpdates')
|
||||
updates = self._bot.getUpdates()
|
||||
self.assertEqual(129566614, updates[0].update_id)
|
||||
self.assertIsInstance(updates[0], telegram.Update)
|
||||
|
||||
def testForwardMessage(self):
|
||||
'''Test the telegram.Bot forwardMessage method'''
|
||||
|
|
@ -58,6 +58,13 @@ class BotTest(unittest.TestCase):
|
|||
chat_id=12173560)
|
||||
self.assertEqual('AgADAQADr6cxGzU8LQe6q0dMJD2rHYkP2ykABFymiQqJgjxRGGMAAgI', message.photo[0].file_id)
|
||||
|
||||
def testSendURLPhoto(self):
|
||||
'''Test the telegram.Bot sendPhoto method'''
|
||||
print('Testing sendPhoto - URL')
|
||||
message = self._bot.sendPhoto(photo=str('http://dummyimage.com/600x400/000/fff.jpg&text=telegram'),
|
||||
chat_id=12173560)
|
||||
self.assertEqual(822, message.photo[0].file_size)
|
||||
|
||||
def testSendAudio(self):
|
||||
'''Test the telegram.Bot sendAudio method'''
|
||||
print('Testing sendAudio - File')
|
||||
|
|
@ -86,13 +93,6 @@ class BotTest(unittest.TestCase):
|
|||
chat_id=12173560)
|
||||
self.assertEqual('BQADAQADHAADNTwtBxZxUGKyxYbYAg', message.document.file_id)
|
||||
|
||||
def testResendSticker(self):
|
||||
'''Test the telegram.Bot sendSticker method'''
|
||||
print('Testing sendSticker - Resend')
|
||||
message = self._bot.sendSticker(sticker=str('BQADAQADHAADyIsGAAFZfq1bphjqlgI'),
|
||||
chat_id=12173560)
|
||||
self.assertEqual(39518, message.sticker.file_size)
|
||||
|
||||
def testSendVideo(self):
|
||||
'''Test the telegram.Bot sendVideo method'''
|
||||
print('Testing sendVideo - File')
|
||||
|
|
@ -107,6 +107,13 @@ class BotTest(unittest.TestCase):
|
|||
chat_id=12173560)
|
||||
self.assertEqual(4, message.video.duration)
|
||||
|
||||
def testResendSticker(self):
|
||||
'''Test the telegram.Bot sendSticker method'''
|
||||
print('Testing sendSticker - Resend')
|
||||
message = self._bot.sendSticker(sticker=str('BQADAQADHAADyIsGAAFZfq1bphjqlgI'),
|
||||
chat_id=12173560)
|
||||
self.assertEqual(39518, message.sticker.file_size)
|
||||
|
||||
def testSendLocation(self):
|
||||
'''Test the telegram.Bot sendLocation method'''
|
||||
print('Testing sendLocation')
|
||||
|
|
|
|||
Loading…
Reference in a new issue