This commit is contained in:
Leandro Toledo 2015-08-28 11:11:32 -03:00
commit ce58f72566
24 changed files with 1076 additions and 373 deletions

View file

@ -22,8 +22,6 @@ __version__ = '2.7.1'
from .base import TelegramObject
from .user import User
from .message import Message
from .update import Update
from .groupchat import GroupChat
from .photosize import PhotoSize
from .audio import Audio
@ -39,10 +37,12 @@ from .replymarkup import ReplyMarkup
from .replykeyboardmarkup import ReplyKeyboardMarkup
from .replykeyboardhide import ReplyKeyboardHide
from .forcereply import ForceReply
from .inputfile import InputFile
from .error import TelegramError
from .inputfile import InputFile
from .nullhandler import NullHandler
from .emoji import Emoji
from .message import Message
from .update import Update
from .bot import Bot
__all__ = ['Bot', 'Emoji', 'TelegramError', 'InputFile', 'ReplyMarkup',

View file

@ -16,37 +16,83 @@
# 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 Audio"""
from telegram import TelegramObject
class Audio(TelegramObject):
"""This object represents a Telegram Audio.
Attributes:
file_id (str):
duration (int):
performer (str):
title (str):
mime_type (str):
file_size (int):
Args:
file_id (str):
duration (int):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
performer (Optional[str]):
title (Optional[str]):
mime_type (Optional[str]):
file_size (Optional[int]):
"""
def __init__(self,
file_id,
duration,
performer=None,
title=None,
mime_type=None,
file_size=None):
**kwargs):
# Required
self.file_id = file_id
self.duration = duration
self.performer = performer
self.title = title
self.mime_type = mime_type
self.file_size = file_size
self.duration = int(duration)
# Optionals
self.performer = kwargs.get('performer', '')
self.title = kwargs.get('title', '')
self.mime_type = kwargs.get('mime_type', '')
self.file_size = int(kwargs.get('file_size', 0))
@staticmethod
def de_json(data):
return Audio(file_id=data.get('file_id', None),
duration=data.get('duration', None),
performer=data.get('performer', None),
title=data.get('title', None),
mime_type=data.get('mime_type', None),
file_size=data.get('file_size', None))
"""
Args:
data (str):
Returns:
telegram.Audio:
"""
if not data:
return None
audio = dict()
# Required
audio['file_id'] = data['file_id']
audio['duration'] = data['duration']
# Optionals
audio['performer'] = data.get('performer')
audio['title'] = data.get('title')
audio['mime_type'] = data.get('mime_type')
audio['file_size'] = data.get('file_size', 0)
return Audio(**audio)
def to_dict(self):
data = {'file_id': self.file_id,
'duration': self.duration}
"""
Returns:
dict:
"""
data = dict()
# Required
data['file_id'] = self.file_id
data['duration'] = self.duration
# Optionals
if self.performer:
data['performer'] = self.performer
if self.title:
@ -55,4 +101,5 @@ class Audio(TelegramObject):
data['mime_type'] = self.mime_type
if self.file_size:
data['file_size'] = self.file_size
return data

View file

@ -41,4 +41,4 @@ class TelegramObject(object):
@abstractmethod
def to_dict(self):
return
return None

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# pylint: disable=R0903
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <leandrotoeldodesouza@gmail.com>
@ -16,8 +17,12 @@
# 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 ChatAction"""
class ChatAction(object):
"""This object represents a Telegram ChatAction."""
TYPING = 'typing'
UPLOAD_PHOTO = 'upload_photo'
RECORD_VIDEO = 'record_video'

View file

@ -16,33 +16,78 @@
# 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 Contact"""
from telegram import TelegramObject
class Contact(TelegramObject):
"""This object represents a Telegram Contact.
Attributes:
phone_number (str):
first_name (str):
last_name (str):
user_id (int):
Args:
phone_number (str):
first_name (str):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
last_name (Optional[str]):
user_id (Optional[int]):
"""
def __init__(self,
phone_number,
first_name,
last_name=None,
user_id=None):
**kwargs):
# Required
self.phone_number = phone_number
self.first_name = first_name
self.last_name = last_name
self.user_id = user_id
# Optionals
self.last_name = kwargs.get('last_name', '')
self.user_id = int(kwargs.get('user_id', 0))
@staticmethod
def de_json(data):
return Contact(phone_number=data.get('phone_number', None),
first_name=data.get('first_name', None),
last_name=data.get('last_name', None),
user_id=data.get('user_id', None))
"""
Args:
data (str):
Returns:
telegram.Contact:
"""
if not data:
return None
contact = dict()
# Required
contact['phone_number'] = data['phone_number']
contact['first_name'] = data['first_name']
# Optionals
contact['last_name'] = data.get('last_name')
contact['user_id'] = data.get('user_id', 0)
return Contact(**contact)
def to_dict(self):
data = {'phone_number': self.phone_number,
'first_name': self.first_name}
"""
Returns:
dict:
"""
data = dict()
# Required
data['phone_number'] = self.phone_number
data['first_name'] = self.first_name
# Optionals
if self.last_name:
data['last_name'] = self.last_name
if self.user_id:
data['user_id'] = self.user_id
return data

View file

@ -16,39 +16,77 @@
# 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 Document"""
from telegram import TelegramObject
from telegram import PhotoSize, TelegramObject
class Document(TelegramObject):
"""This object represents a Telegram Document.
Attributes:
file_id (str):
thumb (:class:`telegram.PhotoSize`):
file_name (str):
mime_type (str):
file_size (int):
Args:
file_id (str):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
thumb (Optional[:class:`telegram.PhotoSize`]):
file_name (Optional[str]):
mime_type (Optional[str]):
file_size (Optional[int]):
"""
def __init__(self,
file_id,
thumb=None,
file_name=None,
mime_type=None,
file_size=None):
**kwargs):
# Required
self.file_id = file_id
self.thumb = thumb
self.file_name = file_name
self.mime_type = mime_type
self.file_size = file_size
# Optionals
self.thumb = kwargs.get('thumb')
self.file_name = kwargs.get('file_name', '')
self.mime_type = kwargs.get('mime_type', '')
self.file_size = int(kwargs.get('file_size', 0))
@staticmethod
def de_json(data):
if 'thumb' in data:
from telegram import PhotoSize
thumb = PhotoSize.de_json(data['thumb'])
else:
thumb = None
"""
Args:
data (str):
return Document(file_id=data.get('file_id', None),
thumb=thumb,
file_name=data.get('file_name', None),
mime_type=data.get('mime_type', None),
file_size=data.get('file_size', None))
Returns:
telegram.Document:
"""
if not data:
return None
document = dict()
# Required
document['file_id'] = data['file_id']
# Optionals
document['thumb'] = PhotoSize.de_json(data.get('thumb'))
document['file_name'] = data.get('file_name')
document['mime_type'] = data.get('mime_type')
document['file_size'] = data.get('file_size', 0)
return Document(**document)
def to_dict(self):
data = {'file_id': self.file_id}
"""
Returns:
dict:
"""
data = dict()
# Required
data['file_id'] = self.file_id
# Optionals
if self.thumb:
data['thumb'] = self.thumb.to_dict()
if self.file_name:
@ -57,4 +95,5 @@ class Document(TelegramObject):
data['mime_type'] = self.mime_type
if self.file_size:
data['file_size'] = self.file_size
return data

View file

@ -1,5 +1,6 @@
#!/usr/bin/env python
# flake8: noqa
# pylint: disable=C0103,C0301,R0903
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <leandrotoeldodesouza@gmail.com>
@ -17,8 +18,12 @@
# 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 an Emoji"""
class Emoji(object):
"""This object represents an Emoji."""
GRINNING_FACE_WITH_SMILING_EYES = b'\xF0\x9F\x98\x81'
FACE_WITH_TEARS_OF_JOY = b'\xF0\x9F\x98\x82'
SMILING_FACE_WITH_OPEN_MOUTH = b'\xF0\x9F\x98\x83'
@ -155,16 +160,26 @@ class Emoji(object):
SQUARED_SOS = b'\xF0\x9F\x86\x98'
SQUARED_UP_WITH_EXCLAMATION_MARK = b'\xF0\x9F\x86\x99'
SQUARED_VS = b'\xF0\x9F\x86\x9A'
REGIONAL_INDICATOR_SYMBOL_LETTER_D_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_E = b'\xF0\x9F\x87\xA9\xF0\x9F\x87\xAA'
REGIONAL_INDICATOR_SYMBOL_LETTER_G_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_B = b'\xF0\x9F\x87\xAC\xF0\x9F\x87\xA7'
REGIONAL_INDICATOR_SYMBOL_LETTER_C_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_N = b'\xF0\x9F\x87\xA8\xF0\x9F\x87\xB3'
REGIONAL_INDICATOR_SYMBOL_LETTER_J_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_P = b'\xF0\x9F\x87\xAF\xF0\x9F\x87\xB5'
REGIONAL_INDICATOR_SYMBOL_LETTER_K_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_R = b'\xF0\x9F\x87\xB0\xF0\x9F\x87\xB7'
REGIONAL_INDICATOR_SYMBOL_LETTER_F_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_R = b'\xF0\x9F\x87\xAB\xF0\x9F\x87\xB7'
REGIONAL_INDICATOR_SYMBOL_LETTER_E_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_S = b'\xF0\x9F\x87\xAA\xF0\x9F\x87\xB8'
REGIONAL_INDICATOR_SYMBOL_LETTER_I_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_T = b'\xF0\x9F\x87\xAE\xF0\x9F\x87\xB9'
REGIONAL_INDICATOR_SYMBOL_LETTER_U_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_S = b'\xF0\x9F\x87\xBA\xF0\x9F\x87\xB8'
REGIONAL_INDICATOR_SYMBOL_LETTER_R_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_U = b'\xF0\x9F\x87\xB7\xF0\x9F\x87\xBA'
REGIONAL_INDICATOR_SYMBOL_LETTER_D_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_E\
= b'\xF0\x9F\x87\xA9\xF0\x9F\x87\xAA'
REGIONAL_INDICATOR_SYMBOL_LETTER_G_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_B\
= b'\xF0\x9F\x87\xAC\xF0\x9F\x87\xA7'
REGIONAL_INDICATOR_SYMBOL_LETTER_C_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_N\
= b'\xF0\x9F\x87\xA8\xF0\x9F\x87\xB3'
REGIONAL_INDICATOR_SYMBOL_LETTER_J_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_P\
= b'\xF0\x9F\x87\xAF\xF0\x9F\x87\xB5'
REGIONAL_INDICATOR_SYMBOL_LETTER_K_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_R\
= b'\xF0\x9F\x87\xB0\xF0\x9F\x87\xB7'
REGIONAL_INDICATOR_SYMBOL_LETTER_F_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_R\
= b'\xF0\x9F\x87\xAB\xF0\x9F\x87\xB7'
REGIONAL_INDICATOR_SYMBOL_LETTER_E_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_S\
= b'\xF0\x9F\x87\xAA\xF0\x9F\x87\xB8'
REGIONAL_INDICATOR_SYMBOL_LETTER_I_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_T\
= b'\xF0\x9F\x87\xAE\xF0\x9F\x87\xB9'
REGIONAL_INDICATOR_SYMBOL_LETTER_U_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_S\
= b'\xF0\x9F\x87\xBA\xF0\x9F\x87\xB8'
REGIONAL_INDICATOR_SYMBOL_LETTER_R_PLUS_REGIONAL_INDICATOR_SYMBOL_LETTER_U\
= b'\xF0\x9F\x87\xB7\xF0\x9F\x87\xBA'
SQUARED_KATAKANA_KOKO = b'\xF0\x9F\x88\x81'
SQUARED_KATAKANA_SA = b'\xF0\x9F\x88\x82'
SQUARED_CJK_UNIFIED_IDEOGRAPH_7121 = b'\xF0\x9F\x88\x9A'

83
telegram/enchancedbot.py Normal file
View file

@ -0,0 +1,83 @@
import telegram
class NoSuchCommandException(BaseException):
pass
class CommandDispatcher:
def __init__(self,):
self.commands = list()
self.default = None
def addCommand(self, command, callback):
self.commands.append((command, callback))
def setDefault(self, callback):
self.default = callback
def dispatch(self, update):
if hasattr(update.message, 'text'):
text = update.message.text
else:
text = ''
user_id = update.message.from_user.id
com = text.split('@')[0]
for command, callback in self.commands:
if com == command:
return callback(command, user_id)
if self.default is not None:
return self.default(text, user_id)
else:
raise NoSuchCommandException()
class EnhancedBot(telegram.Bot):
"""The Bot class with command dispatcher added.
>>> bot = EnhancedBot(token=TOKEN)
>>> @bot.command('/start')
... def start(command, user_id):
... # should return a tuple: (text, reply_id, custom_keyboard)
... return ("Hello, there! Your id is {}".format(user_id), None, None)
>>> while True:
... bot.processUpdates()
... time.sleep(3)
"""
def __init__(self, token):
self.dispatcher = CommandDispatcher()
telegram.Bot.__init__(self, token=token)
self.offset = 0 #id of the last processed update
def command(self, *names, default=False):
"""Decorator for adding callbacks for commands."""
def inner_command(callback):
for name in names:
self.dispatcher.addCommand(name, callback)
if default:
self.dispatcher.setDefault(callback)
return callback # doesn't touch the callback, so we can use it
return inner_command
def processUpdates(self):
updates = self.getUpdates(offset=self.offset)
for update in updates:
print('processing update: {}'.format(str(update.to_dict())))
self.offset = update.update_id + 1
if not hasattr(update, 'message'):
continue
try:
answer, reply_to, reply_markup = self.dispatcher.dispatch(update)
except Exception as e:
print('error occured') # TODO logging
print(update.to_dict())
raise e
if answer is not None:
self.sendMessage(chat_id=update.message.chat_id,
text=answer,
reply_to_message_id=reply_to,
reply_markup=reply_markup)

View file

@ -16,11 +16,16 @@
# 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 Error"""
class TelegramError(Exception):
"""Base class for Telegram errors."""
"""This object represents a Telegram Error."""
@property
def message(self):
'''Returns the first argument used to construct this error.'''
"""
Returns:
str:
"""
return self.args[0]

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# pylint: disable=C0103,W0622
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <leandrotoeldodesouza@gmail.com>
@ -16,23 +17,59 @@
# 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 GroupChat"""
from telegram import TelegramObject
class GroupChat(TelegramObject):
"""This object represents a Telegram GroupChat.
Attributes:
id (int):
title (str):
Args:
id (int):
title (str):
"""
def __init__(self,
id,
title):
self.id = id
# Required
self.id = int(id)
self.title = title
@staticmethod
def de_json(data):
return GroupChat(id=data.get('id', None),
title=data.get('title', None))
"""
Args:
data (str):
Returns:
telegram.GroupChat:
"""
if not data:
return None
groupchat = dict()
# Required
groupchat['id'] = data['id']
groupchat['title'] = data['title']
return GroupChat(**groupchat)
def to_dict(self):
data = {'id': self.id,
'title': self.title}
"""
Returns:
dict:
"""
data = dict()
# Required
data['id'] = self.id
data['title'] = self.title
return data

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# pylint: disable=W0622,E0611
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <leandrotoeldodesouza@gmail.com>
@ -16,6 +17,7 @@
# 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 InputFile"""
try:
from email.generator import _make_boundary as choose_boundary
@ -29,7 +31,7 @@ import os
import sys
import imghdr
from .error import TelegramError
from telegram import TelegramError
DEFAULT_MIME_TYPE = 'application/octet-stream'
USER_AGENT = 'Python Telegram Bot' \
@ -37,6 +39,8 @@ USER_AGENT = 'Python Telegram Bot' \
class InputFile(object):
"""This object represents a Telegram InputFile."""
def __init__(self,
data):
self.data = data
@ -71,14 +75,26 @@ class InputFile(object):
@property
def headers(self):
"""
Returns:
str:
"""
return {'User-agent': USER_AGENT,
'Content-type': self.content_type}
@property
def content_type(self):
"""
Returns:
str:
"""
return 'multipart/form-data; boundary=%s' % self.boundary
def to_form(self):
"""
Returns:
str:
"""
form = []
form_boundary = '--' + self.boundary
@ -105,9 +121,14 @@ class InputFile(object):
form.append('--' + self.boundary + '--')
form.append('')
return self._parse(form)
return InputFile._parse(form)
def _parse(self, form):
@staticmethod
def _parse(form):
"""
Returns:
str:
"""
if sys.version_info > (3,):
# on Python 3 form needs to be byte encoded
encoded_form = []
@ -125,11 +146,10 @@ class InputFile(object):
"""Check if the content file is an image by analyzing its headers.
Args:
stream:
A str representing the content of a file.
stream (str): A str representing the content of a file.
Returns:
The str mimetype of an image.
str: The str mimetype of an image.
"""
image = imghdr.what(None, stream)
if image:
@ -142,8 +162,7 @@ class InputFile(object):
"""Check if the request is a file request.
Args:
data:
A dict of (str, unicode) key/value pairs
data (str): A dict of (str, unicode) key/value pairs
Returns:
bool

View file

@ -16,23 +16,59 @@
# 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 Location"""
from telegram import TelegramObject
class Location(TelegramObject):
"""This object represents a Telegram Sticker.
Attributes:
longitude (float):
latitude (float):
Args:
longitude (float):
latitude (float):
"""
def __init__(self,
longitude,
latitude):
self.longitude = longitude
self.latitude = latitude
# Required
self.longitude = float(longitude)
self.latitude = float(latitude)
@staticmethod
def de_json(data):
return Location(longitude=data.get('longitude', None),
latitude=data.get('latitude', None))
"""
Args:
data (str):
Returns:
telegram.Location:
"""
if not data:
return None
location = dict()
# Required
location['longitude'] = data['longitude']
location['latitude'] = data['latitude']
return Location(**location)
def to_dict(self):
data = {'longitude': self.longitude,
'latitude': self.latitude}
"""
Returns:
dict:
"""
data = dict()
# Required
data['longitude'] = self.longitude
data['latitude'] = self.latitude
return data

View file

@ -1,4 +1,5 @@
#!/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 <leandrotoeldodesouza@gmail.com>
@ -16,214 +17,192 @@
# 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 Message"""
from telegram import TelegramObject
from datetime import datetime
from time import mktime
from telegram import (Audio, Contact, Document, GroupChat, Location, PhotoSize,
Sticker, TelegramObject, User, Video, Voice)
class Message(TelegramObject):
"""This object represents a Telegram Message.
Note:
* In Python `from` is a reserved word, use `from_user` instead.
Attributes:
message_id (int):
from_user (:class:`telegram.User`):
date (:class:`datetime.datetime`):
forward_from (:class:`telegram.User`):
forward_date (:class:`datetime.datetime`):
reply_to_message (:class:`telegram.Message`):
text (str):
audio (:class:`telegram.Audio`):
document (:class:`telegram.Document`):
photo (List[:class:`telegram.PhotoSize`]):
sticker (:class:`telegram.Sticker`):
video (:class:`telegram.Video`):
voice (:class:`telegram.Voice`):
caption (str):
contact (:class:`telegram.Contact`):
location (:class:`telegram.Location`):
new_chat_participant (:class:`telegram.User`):
left_chat_participant (:class:`telegram.User`):
new_chat_title (str):
new_chat_photo (List[:class:`telegram.PhotoSize`]):
delete_chat_photo (bool):
group_chat_created (bool):
Args:
message_id (int):
from_user (:class:`telegram.User`):
date (:class:`datetime.datetime`):
chat (:class:`telegram.User` or :class:`telegram.GroupChat`):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
forward_from (Optional[:class:`telegram.User`]):
forward_date (Optional[:class:`datetime.datetime`]):
reply_to_message (Optional[:class:`telegram.Message`]):
text (Optional[str]):
audio (Optional[:class:`telegram.Audio`]):
document (Optional[:class:`telegram.Document`]):
photo (Optional[List[:class:`telegram.PhotoSize`]]):
sticker (Optional[:class:`telegram.Sticker`]):
video (Optional[:class:`telegram.Video`]):
voice (Optional[:class:`telegram.Voice`]):
caption (Optional[str]):
contact (Optional[:class:`telegram.Contact`]):
location (Optional[:class:`telegram.Location`]):
new_chat_participant (Optional[:class:`telegram.User`]):
left_chat_participant (Optional[:class:`telegram.User`]):
new_chat_title (Optional[str]):
new_chat_photo (Optional[List[:class:`telegram.PhotoSize`]):
delete_chat_photo (Optional[bool]):
group_chat_created (Optional[bool]):
"""
def __init__(self,
message_id,
from_user,
date,
chat,
forward_from=None,
forward_date=None,
reply_to_message=None,
text=None,
audio=None,
document=None,
photo=None,
sticker=None,
video=None,
voice=None,
caption=None,
contact=None,
location=None,
new_chat_participant=None,
left_chat_participant=None,
new_chat_title=None,
new_chat_photo=None,
delete_chat_photo=None,
group_chat_created=None):
self.message_id = message_id
**kwargs):
# Required
self.message_id = int(message_id)
self.from_user = from_user
self.date = date
self.chat = chat
self.forward_from = forward_from
self.forward_date = forward_date
self.reply_to_message = reply_to_message
self.text = text
self.audio = audio
self.document = document
self.photo = photo
self.sticker = sticker
self.video = video
self.voice = voice
self.caption = caption
self.contact = contact
self.location = location
self.new_chat_participant = new_chat_participant
self.left_chat_participant = left_chat_participant
self.new_chat_title = new_chat_title
self.new_chat_photo = new_chat_photo
self.delete_chat_photo = delete_chat_photo
self.group_chat_created = group_chat_created
# Optionals
self.forward_from = kwargs.get('forward_from')
self.forward_date = kwargs.get('forward_date')
self.reply_to_message = kwargs.get('reply_to_message')
self.text = kwargs.get('text', '')
self.audio = kwargs.get('audio')
self.document = kwargs.get('document')
self.photo = kwargs.get('photo')
self.sticker = kwargs.get('sticker')
self.video = kwargs.get('video')
self.voice = kwargs.get('voice')
self.caption = kwargs.get('caption', '')
self.contact = kwargs.get('contact')
self.location = kwargs.get('location')
self.new_chat_participant = kwargs.get('new_chat_participant')
self.left_chat_participant = kwargs.get('left_chat_participant')
self.new_chat_title = kwargs.get('new_chat_title', '')
self.new_chat_photo = kwargs.get('new_chat_photo')
self.delete_chat_photo = bool(kwargs.get('delete_chat_photo', False))
self.group_chat_created = bool(kwargs.get('group_chat_created', False))
@property
def chat_id(self):
"""int: Short for :attr:`Message.chat.id`"""
return self.chat.id
@staticmethod
def de_json(data):
if 'from' in data: # from is a reserved word, use from_user instead.
from telegram import User
from_user = User.de_json(data['from'])
else:
from_user = None
"""
Args:
data (str):
if 'date' in data:
date = datetime.fromtimestamp(data['date'])
else:
date = None
Returns:
telegram.Message:
"""
if not data:
return None
if 'chat' in data:
if 'first_name' in data['chat']:
from telegram import User
chat = User.de_json(data['chat'])
if 'title' in data['chat']:
from telegram import GroupChat
chat = GroupChat.de_json(data['chat'])
else:
chat = None
message = dict()
if 'forward_from' in data:
from telegram import User
forward_from = User.de_json(data['forward_from'])
else:
forward_from = None
# Required
message['message_id'] = data['message_id']
message['from_user'] = User.de_json(data['from'])
message['date'] = datetime.fromtimestamp(data['date'])
# Optionals
if 'first_name' in data.get('chat', ''):
message['chat'] = User.de_json(data.get('chat'))
elif 'title' in data.get('chat', ''):
message['chat'] = GroupChat.de_json(data.get('chat'))
message['forward_from'] = \
User.de_json(data.get('forward_from'))
message['forward_date'] = \
Message._fromtimestamp(data.get('forward_date'))
message['reply_to_message'] = \
Message.de_json(data.get('reply_to_message'))
message['text'] = \
data.get('text')
message['audio'] = \
Audio.de_json(data.get('audio'))
message['document'] = \
Document.de_json(data.get('document'))
message['photo'] = \
[PhotoSize.de_json(x) for x in data.get('photo', [])]
message['sticker'] = \
Sticker.de_json(data.get('sticker'))
message['video'] = \
Video.de_json(data.get('video'))
message['voice'] = \
Voice.de_json(data.get('voice'))
message['caption'] = \
data.get('caption')
message['contact'] = \
Contact.de_json(data.get('contact'))
message['location'] = \
Location.de_json(data.get('location'))
message['new_chat_participant'] = \
User.de_json(data.get('new_chat_participant'))
message['left_chat_participant'] = \
User.de_json(data.get('left_chat_participant'))
message['new_chat_title'] = \
data.get('new_chat_title')
message['new_chat_photo'] = \
[PhotoSize.de_json(x) for x in data.get('new_chat_photo', [])]
message['delete_chat_photo'] = \
data.get('delete_chat_photo')
message['group_chat_created'] = \
data.get('group_chat_created')
if 'forward_date' in data:
forward_date = datetime.fromtimestamp(data['forward_date'])
else:
forward_date = None
if 'reply_to_message' in data:
reply_to_message = Message.de_json(data['reply_to_message'])
else:
reply_to_message = None
if 'audio' in data:
from telegram import Audio
audio = Audio.de_json(data['audio'])
else:
audio = None
if 'document' in data:
from telegram import Document
document = Document.de_json(data['document'])
else:
document = None
if 'photo' in data:
from telegram import PhotoSize
photo = [PhotoSize.de_json(x) for x in data['photo']]
else:
photo = None
if 'sticker' in data:
from telegram import Sticker
sticker = Sticker.de_json(data['sticker'])
else:
sticker = None
if 'video' in data:
from telegram import Video
video = Video.de_json(data['video'])
else:
video = None
if 'voice' in data:
from telegram import Voice
voice = Voice.de_json(data['voice'])
else:
voice = None
if 'contact' in data:
from telegram import Contact
contact = Contact.de_json(data['contact'])
else:
contact = None
if 'location' in data:
from telegram import Location
location = Location.de_json(data['location'])
else:
location = None
if 'new_chat_participant' in data:
from telegram import User
new_chat_participant = User.de_json(data['new_chat_participant'])
else:
new_chat_participant = None
if 'left_chat_participant' in data:
from telegram import User
left_chat_participant = User.de_json(data['left_chat_participant'])
else:
left_chat_participant = None
if 'new_chat_photo' in data:
from telegram import PhotoSize
new_chat_photo = \
[PhotoSize.de_json(x) for x in data['new_chat_photo']]
else:
new_chat_photo = None
return Message(message_id=data.get('message_id', None),
from_user=from_user,
date=date,
chat=chat,
forward_from=forward_from,
forward_date=forward_date,
reply_to_message=reply_to_message,
text=data.get('text', ''),
audio=audio,
document=document,
photo=photo,
sticker=sticker,
video=video,
voice=voice,
caption=data.get('caption', ''),
contact=contact,
location=location,
new_chat_participant=new_chat_participant,
left_chat_participant=left_chat_participant,
new_chat_title=data.get('new_chat_title', None),
new_chat_photo=new_chat_photo,
delete_chat_photo=data.get('delete_chat_photo', None),
group_chat_created=data.get('group_chat_created', None))
return Message(**message)
def to_dict(self):
data = {'message_id': self.message_id,
'from': self.from_user.to_dict(),
'chat': self.chat.to_dict()}
try:
# Python 3.3+ supports .timestamp()
data['date'] = int(self.date.timestamp())
if self.forward_date:
data['forward_date'] = int(self.forward_date.timestamp())
except AttributeError:
# _totimestamp() for Python 3 (< 3.3) and Python 2
data['date'] = self._totimestamp(self.date)
if self.forward_date:
data['forward_date'] = self._totimestamp(self.forward_date)
"""
Returns:
dict:
"""
data = dict()
# Required
data['message_id'] = self.message_id
data['from'] = self.from_user.to_dict()
data['date'] = self._totimestamp(self.date)
data['chat'] = self.chat.to_dict()
# Optionals
if self.forward_from:
data['forward_from'] = self.forward_from.to_dict()
if self.forward_date:
data['forward_date'] = self._totimestamp(self.forward_date)
if self.reply_to_message:
data['reply_to_message'] = self.reply_to_message.to_dict()
if self.text:
@ -259,8 +238,38 @@ class Message(TelegramObject):
data['delete_chat_photo'] = self.delete_chat_photo
if self.group_chat_created:
data['group_chat_created'] = self.group_chat_created
return data
@staticmethod
def _totimestamp(dt):
return int(mktime(dt.timetuple()))
def _fromtimestamp(unixtime):
"""
Args:
unixtime (int):
Returns:
datetime.datetime:
"""
if not unixtime:
return None
return datetime.fromtimestamp(unixtime)
@staticmethod
def _totimestamp(dt_obj):
"""
Args:
dt_obj (:class:`datetime.datetime`):
Returns:
int:
"""
if not dt_obj:
return None
try:
# Python 3.3+
return int(dt_obj.timestamp())
except AttributeError:
# Python 3 (< 3.3) and Python 2
return int(mktime(dt_obj.timetuple()))

View file

@ -16,10 +16,17 @@
# 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 logging NullHandler"""
import logging
class NullHandler(logging.Handler):
"""This object represents a logging NullHandler."""
def emit(self, record):
"""
Args:
record (str):
"""
pass

View file

@ -16,32 +16,78 @@
# 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 PhotoSize"""
from telegram import TelegramObject
class PhotoSize(TelegramObject):
"""This object represents a Telegram PhotoSize.
Attributes:
file_id (str):
width (int):
height (int):
file_size (int):
Args:
file_id (str):
width (int):
height (int):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
file_size (Optional[int]):
"""
def __init__(self,
file_id,
width,
height,
file_size=None):
**kwargs):
# Required
self.file_id = file_id
self.width = width
self.height = height
self.file_size = file_size
self.width = int(width)
self.height = int(height)
# Optionals
self.file_size = int(kwargs.get('file_size', 0))
@staticmethod
def de_json(data):
return PhotoSize(file_id=data.get('file_id', None),
width=data.get('width', None),
height=data.get('height', None),
file_size=data.get('file_size', None))
"""
Args:
data (str):
Returns:
telegram.PhotoSize:
"""
if not data:
return None
photosize = dict()
# Required
photosize['file_id'] = data['file_id']
photosize['width'] = data['width']
photosize['height'] = data['height']
# Optionals
photosize['file_size'] = data.get('file_size', 0)
return PhotoSize(**photosize)
def to_dict(self):
data = {'file_id': self.file_id,
'width': self.width,
'height': self.height}
"""
Returns:
dict:
"""
data = dict()
# Required
data['file_id'] = self.file_id
data['width'] = self.width
data['height'] = self.height
# Optionals
if self.file_size:
data['file_size'] = self.file_size
return data

View file

@ -16,42 +16,84 @@
# 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 Sticker"""
from telegram import TelegramObject
from telegram import PhotoSize, TelegramObject
class Sticker(TelegramObject):
"""This object represents a Telegram Sticker.
Attributes:
file_id (str):
width (int):
height (int):
thumb (:class:`telegram.PhotoSize`):
file_size (int):
Args:
file_id (str):
width (int):
height (int):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
thumb (Optional[:class:`telegram.PhotoSize`]):
file_size (Optional[int]):
"""
def __init__(self,
file_id,
width,
height,
thumb=None,
file_size=None):
**kwargs):
# Required
self.file_id = file_id
self.width = width
self.height = height
self.thumb = thumb
self.file_size = file_size
self.width = int(width)
self.height = int(height)
# Optionals
self.thumb = kwargs.get('thumb')
self.file_size = int(kwargs.get('file_size', 0))
@staticmethod
def de_json(data):
if 'thumb' in data:
from telegram import PhotoSize
thumb = PhotoSize.de_json(data['thumb'])
else:
thumb = None
"""
Args:
data (str):
return Sticker(file_id=data.get('file_id', None),
width=data.get('width', None),
height=data.get('height', None),
thumb=thumb,
file_size=data.get('file_size', None))
Returns:
telegram.Sticker:
"""
if not data:
return None
sticker = dict()
# Required
sticker['file_id'] = data['file_id']
sticker['width'] = data['width']
sticker['height'] = data['height']
# Optionals
sticker['thumb'] = PhotoSize.de_json(data['thumb'])
sticker['file_size'] = data.get('file_size', 0)
return Sticker(**sticker)
def to_dict(self):
data = {'file_id': self.file_id,
'width': self.width,
'height': self.height,
'thumb': self.thumb.to_dict()}
"""
Returns:
dict:
"""
data = dict()
# Required
data['file_id'] = self.file_id
data['width'] = self.width
data['height'] = self.height
# Optionals
if self.thumb:
data['thumb'] = self.thumb.to_dict()
if self.file_size:
data['file_size'] = self.file_size
return data

View file

@ -16,30 +16,60 @@
# 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 Update"""
from telegram import TelegramObject
from telegram import Message, TelegramObject
class Update(TelegramObject):
"""This object represents a Telegram Update.
Attributes:
update_id (int):
message (:class:`telegram.Message`):
Args:
update_id (int):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
message (Optional[:class:`telegram.Message`]):
"""
def __init__(self,
update_id,
message=None):
**kwargs):
# Required
self.update_id = update_id
self.message = message
# Optionals
self.message = kwargs.get('message')
@staticmethod
def de_json(data):
if 'message' in data:
from telegram import Message
message = Message.de_json(data['message'])
else:
message = None
"""
Args:
data (str):
return Update(update_id=data.get('update_id', None),
message=message)
Returns:
telegram.Update:
"""
update = dict()
update['update_id'] = data['update_id']
update['message'] = Message.de_json(data['message'])
return Update(**update)
def to_dict(self):
data = {'update_id': self.update_id}
"""
Returns:
dict:
"""
data = dict()
# Required
data['update_id'] = self.update_id
# Optionals
if self.message:
data['message'] = self.message.to_dict()
return data

View file

@ -1,4 +1,5 @@
#!/usr/bin/env python
# pylint: disable=C0103,W0622
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <leandrotoeldodesouza@gmail.com>
@ -16,23 +17,44 @@
# 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 User"""
from telegram import TelegramObject
class User(TelegramObject):
"""This object represents a Telegram Sticker.
Attributes:
id (int):
first_name (str):
last_name (str):
username (str):
Args:
id (int):
first_name (str):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
last_name (Optional[str]):
username (Optional[str]):
"""
def __init__(self,
id,
first_name,
last_name=None,
username=None):
self.id = id
**kwargs):
# Required
self.id = int(id)
self.first_name = first_name
self.last_name = last_name
self.username = username
# Optionals
self.last_name = kwargs.get('last_name', '')
self.username = kwargs.get('username', '')
@property
def name(self):
"""str: """
if self.username:
return '@%s' % self.username
if self.last_name:
@ -41,16 +63,41 @@ class User(TelegramObject):
@staticmethod
def de_json(data):
return User(id=data.get('id', None),
first_name=data.get('first_name', None),
last_name=data.get('last_name', None),
username=data.get('username', None))
"""
Args:
data (str):
Returns:
telegram.User:
"""
if not data:
return None
user = dict()
# Required
user['id'] = data['id']
user['first_name'] = data['first_name']
# Optionals
user['last_name'] = data.get('last_name')
user['username'] = data.get('username')
return User(**user)
def to_dict(self):
data = {'id': self.id,
'first_name': self.first_name}
"""
Returns:
dict:
"""
data = dict()
# Required
data['id'] = self.id
data['first_name'] = self.first_name
# Optionals
if self.last_name:
data['last_name'] = self.last_name
if self.username:
data['username'] = self.username
return data

View file

@ -16,36 +16,64 @@
# 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
UserProfilePhotos"""
from telegram import TelegramObject
from telegram import PhotoSize, TelegramObject
class UserProfilePhotos(TelegramObject):
"""This object represents a Telegram UserProfilePhotos.
Attributes:
total_count (int):
photos (List[List[:class:`telegram.PhotoSize`]]):
Args:
total_count (int):
photos (List[List[:class:`telegram.PhotoSize`]]):
"""
def __init__(self,
total_count,
photos):
self.total_count = total_count
# Required
self.total_count = int(total_count)
self.photos = photos
@staticmethod
def de_json(data):
if 'photos' in data:
from telegram import PhotoSize
photos = []
for photo in data['photos']:
photos.append([PhotoSize.de_json(x) for x in photo])
else:
photos = None
"""
Args:
data (str):
return UserProfilePhotos(total_count=data.get('total_count', None),
photos=photos)
Returns:
telegram.UserProfilePhotos:
"""
if not data:
return None
upf = dict()
# Required
upf['total_count'] = data['total_count']
upf['photos'] = []
for photo in data['photos']:
upf['photos'].append([PhotoSize.de_json(x) for x in photo])
return UserProfilePhotos(**upf)
def to_dict(self):
data = {}
if self.total_count:
data['total_count'] = self.total_count
if self.photos:
data['photos'] = []
for photo in self.photos:
data['photos'].append([x.to_dict() for x in photo])
"""
Returns:
dict:
"""
data = dict()
# Required
data['total_count'] = self.total_count
data['photos'] = []
for photo in self.photos:
data['photos'].append([x.to_dict() for x in photo])
return data

View file

74
telegram/utils/botan.py Normal file
View file

@ -0,0 +1,74 @@
#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015 Leandro Toledo de Souza <leandrotoeldodesouza@gmail.com>
#
# 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/].
import json
try:
from urllib.parse import urlencode
from urllib.request import urlopen, Request
from urllib.error import HTTPError, URLError
except ImportError:
from urllib import urlencode
from urllib2 import urlopen, Request
from urllib2 import HTTPError, URLError
DEFAULT_BASE_URL = \
'https://api.botan.io/track?token=%(token)&uid=%(uid)&name=%(name)'
USER_AGENT = 'Python Telegram Bot' \
' (https://github.com/leandrotoledo/python-telegram-bot)'
CONTENT_TYPE = 'application/json'
class Botan(Object):
def __init__(self,
token,
base_url=None):
self.token = token
if base_url is None:
self.base_url = DEFAULT_BASE_URL % {'token': self.token}
else:
self.base_url = base_url % {'token': self.token}
def track(self,
uid,
text,
name = 'Message'):
url = self.base_url % {'uid': uid,
'message': text,
'name': name}
self._post(url, message)
def _post(self,
url,
data):
headers = {'User-agent': USER_AGENT,
'Content-type': CONTENT_TYPE}
try:
request = Request(
url,
data=urlencode(data).encode(),
headers=headers
)
return urlopen(request).read()
except IOError as e:
raise TelegramError(str(e))
except HTTPError as e:
raise TelegramError(str(e))

View file

@ -16,52 +16,96 @@
# 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 Video"""
from telegram import TelegramObject
from telegram import PhotoSize, TelegramObject
class Video(TelegramObject):
"""This object represents a Telegram Video.
Attributes:
file_id (str):
width (int):
height (int):
duration (int):
thumb (:class:`telegram.PhotoSize`):
mime_type (str):
file_size (int):
Args:
file_id (str):
width (int):
height (int):
duration (int):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
thumb (Optional[:class:`telegram.PhotoSize`]):
mime_type (Optional[str]):
file_size (Optional[int]):
"""
def __init__(self,
file_id,
width,
height,
duration,
thumb=None,
mime_type=None,
file_size=None):
**kwargs):
# Required
self.file_id = file_id
self.width = width
self.height = height
self.duration = duration
self.thumb = thumb
self.mime_type = mime_type
self.file_size = file_size
self.width = int(width)
self.height = int(height)
self.duration = int(duration)
# Optionals
self.thumb = kwargs.get('thumb')
self.mime_type = kwargs.get('mime_type', '')
self.file_size = int(kwargs.get('file_size', 0))
@staticmethod
def de_json(data):
if 'thumb' in data:
from telegram import PhotoSize
thumb = PhotoSize.de_json(data['thumb'])
else:
thumb = None
"""
Args:
data (str):
return Video(file_id=data.get('file_id', None),
width=data.get('width', None),
height=data.get('height', None),
duration=data.get('duration', None),
thumb=thumb,
mime_type=data.get('mime_type', None),
file_size=data.get('file_size', None))
Returns:
telegram.Video:
"""
if not data:
return None
video = dict()
# Required
video['file_id'] = data['file_id']
video['width'] = data['width']
video['height'] = data['height']
video['duration'] = data['duration']
# Optionals
video['thumb'] = PhotoSize.de_json(data.get('thumb'))
video['mime_type'] = data.get('mime_type')
video['file_size'] = data.get('file_size', 0)
return Video(**video)
def to_dict(self):
data = {'file_id': self.file_id,
'width': self.width,
'height': self.height,
'duration': self.duration}
"""
Returns:
dict:
"""
data = dict()
# Required
data['file_id'] = self.file_id
data['width'] = self.width
data['height'] = self.height
data['duration'] = self.duration
# Optionals
if self.thumb:
data['thumb'] = self.thumb.to_dict()
if self.mime_type:
data['mime_type'] = self.mime_type
if self.file_size:
data['file_size'] = self.file_size
return data

View file

@ -16,34 +16,78 @@
# 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 Voice"""
from telegram import TelegramObject
class Voice(TelegramObject):
"""This object represents a Telegram Voice.
Attributes:
file_id (str):
duration (int):
mime_type (str):
file_size (int):
Args:
file_id (str):
**kwargs: Arbitrary keyword arguments.
Keyword Args:
duration (Optional[int]):
mime_type (Optional[str]):
file_size (Optional[int]):
"""
def __init__(self,
file_id,
duration=None,
mime_type=None,
file_size=None):
**kwargs):
# Required
self.file_id = file_id
self.duration = duration
self.mime_type = mime_type
self.file_size = file_size
# Optionals
self.duration = int(kwargs.get('duration', 0))
self.mime_type = kwargs.get('mime_type', '')
self.file_size = int(kwargs.get('file_size', 0))
@staticmethod
def de_json(data):
return Voice(file_id=data.get('file_id', None),
duration=data.get('duration', None),
mime_type=data.get('mime_type', None),
file_size=data.get('file_size', None))
"""
Args:
data (str):
Returns:
telegram.Voice:
"""
if not data:
return None
voice = dict()
# Required
voice['file_id'] = data['file_id']
# Optionals
voice['duration'] = data.get('duration', 0)
voice['mime_type'] = data.get('mime_type')
voice['file_size'] = data.get('file_size', 0)
return Voice(**voice)
def to_dict(self):
data = {'file_id': self.file_id}
"""
Returns:
dict:
"""
data = dict()
# Required
data['file_id'] = self.file_id
# Optionals
if self.duration:
data['duration'] = self.duration
if self.mime_type:
data['mime_type'] = self.mime_type
if self.file_size:
data['file_size'] = self.file_size
return data

View file

@ -62,8 +62,9 @@ class BotTest(unittest.TestCase):
'''Test the telegram.Bot getUpdates method'''
print('Testing getUpdates')
updates = self._bot.getUpdates()
self.assertTrue(self.is_json(updates[0].to_json()))
self.assertIsInstance(updates[0], telegram.Update)
if updates:
self.assertTrue(self.is_json(updates[0].to_json()))
self.assertIsInstance(updates[0], telegram.Update)
def testForwardMessage(self):
'''Test the telegram.Bot forwardMessage method'''