# -*- coding: utf-8 -*- # TODO InlineQueryResultArticle, InlineQueryResultPhoto, InlineQueryResultGif, InlineQueryResultMpeg4Gif, InlineQueryResultVideo import json import six class JsonSerializable: """ Subclasses of this class are guaranteed to be able to be converted to JSON format. All subclasses of this class must override to_json. """ def to_json(self): """ Returns a JSON string representation of this class. This function must be overridden by subclasses. :return: a JSON formatted string. """ raise NotImplementedError class JsonDeserializable: """ Subclasses of this class are guaranteed to be able to be created from a json-style dict or json formatted string. All subclasses of this class must override de_json. """ @classmethod def de_json(cls, json_type): """ Returns an instance of this class from the given json dict or string. This function must be overridden by subclasses. :return: an instance of this class created from the given json dict or string. """ raise NotImplementedError @staticmethod def check_json(json_type): """ Checks whether json_type is a dict or a string. If it is already a dict, it is returned as-is. If it is not, it is converted to a dict by means of json.loads(json_type) :param json_type: :return: """ if type(json_type) == dict: return json_type elif type(json_type) == str: return json.loads(json_type) else: raise ValueError("json_type should be a json dict or string.") def __str__(self): d = {} for x, y in six.iteritems(self.__dict__): if hasattr(y, '__dict__'): d[x] = y.__dict__ else: d[x] = y return six.text_type(d) class Update(JsonDeserializable): @classmethod def de_json(cls, json_type): obj = cls.check_json(json_type) update_id = obj['update_id'] message = None inline_query = None chosen_inline_result = None if 'message' in obj: message = Message.de_json(obj['message']) if 'inline_query' in obj: inline_query = InlineQuery.de_json(obj['inline_query']) if 'chosen_inline_result' in obj: chosen_inline_result = ChosenInlineResult.de_json(obj['chosen_inline_result']) return cls(update_id, message, inline_query, chosen_inline_result) def __init__(self, update_id, message, inline_query, chosen_inline_result): self.update_id = update_id self.message = message self.inline_query = inline_query self.chosen_inline_result = chosen_inline_result class User(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) id = obj['id'] first_name = obj['first_name'] last_name = obj.get('last_name') username = obj.get('username') return cls(id, first_name, last_name, username) def __init__(self, id, first_name, last_name=None, username=None): self.id = id self.first_name = first_name self.username = username self.last_name = last_name class GroupChat(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) id = obj['id'] title = obj['title'] return cls(id, title) def __init__(self, id, title): self.id = id self.title = title class Chat(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) id = obj['id'] type = obj['type'] title = obj.get('title') username = obj.get('username') first_name = obj.get('first_name') last_name = obj.get('last_name') return cls(id, type, title, username, first_name, last_name) def __init__(self, id, type, title=None, username=None, first_name=None, last_name=None): self.type = type self.last_name = last_name self.first_name = first_name self.username = username self.id = id self.title = title class Message(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) message_id = obj['message_id'] from_user = None if 'from' in obj: from_user = User.de_json(obj['from']) chat = Chat.de_json(obj['chat']) date = obj['date'] content_type = None opts = {} if 'forward_from' in obj: opts['forward_from'] = User.de_json(obj['forward_from']) if 'forward_date' in obj: opts['forward_date'] = obj['forward_date'] if 'reply_to_message' in obj: opts['reply_to_message'] = Message.de_json(obj['reply_to_message']) if 'text' in obj: opts['text'] = obj['text'] content_type = 'text' if 'audio' in obj: opts['audio'] = Audio.de_json(obj['audio']) content_type = 'audio' if 'voice' in obj: opts['voice'] = Audio.de_json(obj['voice']) content_type = 'voice' if 'document' in obj: opts['document'] = Document.de_json(obj['document']) content_type = 'document' if 'photo' in obj: opts['photo'] = Message.parse_photo(obj['photo']) content_type = 'photo' if 'sticker' in obj: opts['sticker'] = Sticker.de_json(obj['sticker']) content_type = 'sticker' if 'video' in obj: opts['video'] = Video.de_json(obj['video']) content_type = 'video' if 'location' in obj: opts['location'] = Location.de_json(obj['location']) content_type = 'location' if 'contact' in obj: opts['contact'] = Contact.de_json(json.dumps(obj['contact'])) content_type = 'contact' if 'new_chat_participant' in obj: opts['new_chat_participant'] = User.de_json(obj['new_chat_participant']) content_type = 'new_chat_participant' if 'left_chat_participant' in obj: opts['left_chat_participant'] = User.de_json(obj['left_chat_participant']) content_type = 'left_chat_participant' if 'new_chat_title' in obj: opts['new_chat_title'] = obj['new_chat_title'] content_type = 'new_chat_title' if 'new_chat_photo' in obj: opts['new_chat_photo'] = obj['new_chat_photo'] content_type = 'new_chat_photo' if 'delete_chat_photo' in obj: opts['delete_chat_photo'] = obj['delete_chat_photo'] content_type = 'delete_chat_photo' if 'group_chat_created' in obj: opts['group_chat_created'] = obj['group_chat_created'] content_type = 'group_chat_created' if 'caption' in obj: opts['caption'] = obj['caption'] return cls(message_id, from_user, date, chat, content_type, opts) @classmethod def parse_chat(cls, chat): if 'first_name' not in chat: return GroupChat.de_json(chat) else: return User.de_json(chat) @classmethod def parse_photo(cls, photo_size_array): ret = [] for ps in photo_size_array: ret.append(PhotoSize.de_json(ps)) return ret def __init__(self, message_id, from_user, date, chat, content_type, options): self.chat = chat self.date = date self.from_user = from_user self.message_id = message_id self.content_type = content_type self.forward_from = None self.forward_date = None self.reply_to_message = None self.text = None self.audio = None self.voice = None self.document = None self.photo = None self.sticker = None self.video = None self.location = None self.contact = None self.new_chat_participant = None self.left_chat_participant = None self.new_chat_title = None self.new_chat_photo = None self.delete_chat_photo = None self.group_chat_created = None self.caption = None for key in options: setattr(self, key, options[key]) class PhotoSize(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) file_id = obj['file_id'] width = obj['width'] height = obj['height'] file_size = obj.get('file_size') return cls(file_id, width, height, file_size) def __init__(self, file_id, width, height, file_size=None): self.file_size = file_size self.height = height self.width = width self.file_id = file_id class Audio(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) file_id = obj['file_id'] duration = obj['duration'] performer = obj.get('performer') title = obj.get('title') mime_type = obj.get('mime_type') file_size = obj.get('file_size') return cls(file_id, duration, performer, title, mime_type, file_size) def __init__(self, file_id, duration, performer=None, title=None, mime_type=None, file_size=None): self.file_id = file_id self.duration = duration self.performer = performer self.title = title self.mime_type = mime_type self.file_size = file_size class Voice(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) file_id = obj['file_id'] duration = obj['duration'] mime_type = obj.get('mime_type') file_size = obj.get('file_size') return cls(file_id, duration, mime_type, file_size) def __init__(self, file_id, duration, mime_type=None, file_size=None): self.file_id = file_id self.duration = duration self.mime_type = mime_type self.file_size = file_size class Document(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) file_id = obj['file_id'] thumb = None if 'thumb' in obj and 'file_id' in obj['thumb']: thumb = PhotoSize.de_json(obj['thumb']) file_name = obj.get('file_name') mime_type = obj.get('mime_type') file_size = obj.get('file_size') return cls(file_id, thumb, file_name, mime_type, file_size) def __init__(self, file_id, thumb, file_name=None, mime_type=None, file_size=None): self.file_id = file_id self.thumb = thumb self.file_name = file_name self.mime_type = mime_type self.file_size = file_size class Sticker(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) file_id = obj['file_id'] width = obj['width'] height = obj['height'] thumb = None if 'thumb' in obj: thumb = PhotoSize.de_json(obj['thumb']) file_size = obj.get('file_size') return cls(file_id, width, height, thumb, file_size) def __init__(self, file_id, width, height, thumb, file_size=None): self.file_id = file_id self.width = width self.height = height self.thumb = thumb self.file_size = file_size class Video(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) file_id = obj['file_id'] width = obj['width'] height = obj['height'] duration = obj['duration'] thumb = None if 'thumb' in obj: thumb = PhotoSize.de_json(obj['thumb']) mime_type = obj.get('mime_type') file_size = obj.get('file_size') return cls(file_id, width, height, duration, thumb, mime_type, file_size) def __init__(self, file_id, width, height, duration, thumb=None, mime_type=None, file_size=None): 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 class Contact(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) phone_number = obj['phone_number'] first_name = obj['first_name'] last_name = obj.get('last_name') user_id = obj.get('user_id') return cls(phone_number, first_name, last_name, user_id) def __init__(self, phone_number, first_name, last_name=None, user_id=None): self.phone_number = phone_number self.first_name = first_name self.last_name = last_name self.user_id = user_id class Location(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) longitude = obj['longitude'] latitude = obj['latitude'] return cls(longitude, latitude) def __init__(self, longitude, latitude): self.longitude = longitude self.latitude = latitude class UserProfilePhotos(JsonDeserializable): @classmethod def de_json(cls, json_string): obj = cls.check_json(json_string) total_count = obj['total_count'] photos = [[PhotoSize.de_json(y) for y in x] for x in obj['photos']] return cls(total_count, photos) def __init__(self, total_count, photos): self.total_count = total_count self.photos = photos class File(JsonDeserializable): @classmethod def de_json(cls, json_type): obj = cls.check_json(json_type) file_id = obj['file_id'] file_size = obj.get('file_size') file_path = obj.get('file_path') return cls(file_id, file_size, file_path) def __init__(self, file_id, file_size, file_path): self.file_id = file_id self.file_size = file_size self.file_path = file_path class ForceReply(JsonSerializable): def __init__(self, selective=None): self.selective = selective def to_json(self): json_dict = {'force_reply': True} if self.selective: json_dict['selective'] = True return json.dumps(json_dict) class ReplyKeyboardHide(JsonSerializable): def __init__(self, selective=None): self.selective = selective def to_json(self): json_dict = {'hide_keyboard': True} if self.selective: json_dict['selective'] = True return json.dumps(json_dict) class ReplyKeyboardMarkup(JsonSerializable): def __init__(self, resize_keyboard=None, one_time_keyboard=None, selective=None, row_width=3): self.resize_keyboard = resize_keyboard self.one_time_keyboard = one_time_keyboard self.selective = selective self.row_width = row_width self.keyboard = [] def add(self, *args): """ This function adds strings to the keyboard, while not exceeding row_width. E.g. ReplyKeyboardMarkup#add("A", "B", "C") yields the json result {keyboard: [["A"], ["B"], ["C"]]} when row_width is set to 1. When row_width is set to 2, the following is the result of this function: {keyboard: [["A", "B"], ["C"]]} See https://core.telegram.org/bots/api#replykeyboardmarkup :param args: strings to append to the keyboard """ i = 1 row = [] for string in args: row.append(string) if i % self.row_width == 0: self.keyboard.append(row) row = [] i += 1 if len(row) > 0: self.keyboard.append(row) def row(self, *args): """ Adds a list of strings to the keyboard. This function does not consider row_width. ReplyKeyboardMarkup#row("A")#row("B", "C")#to_json() outputs '{keyboard: [["A"], ["B", "C"]]}' See https://core.telegram.org/bots/api#replykeyboardmarkup :param args: strings :return: self, to allow function chaining. """ self.keyboard.append(args) return self def to_json(self): """ Converts this object to its json representation following the Telegram API guidelines described here: https://core.telegram.org/bots/api#replykeyboardmarkup :return: """ json_dict = {'keyboard': self.keyboard} if self.one_time_keyboard: json_dict['one_time_keyboard'] = True if self.resize_keyboard: json_dict['resize_keyboard'] = True if self.selective: json_dict['selective'] = True return json.dumps(json_dict) class InlineQuery(JsonDeserializable): @classmethod def de_json(cls, json_type): obj = cls.check_json(json_type) id = obj['id'] from_user = User.de_json(obj['from']) query = obj['query'] offset = obj['offset'] return cls(id, from_user, query, offset) def __init__(self, id, from_user, query, offset): """ This object represents an incoming inline query. When the user sends an empty query, your bot could return some default or trending results. :param id: string Unique identifier for this query :param from_user: User Sender :param query: String Text of the query :param offset: String Offset of the results to be returned, can be controlled by the bot :return: InlineQuery Object """ self.id = id self.from_user = from_user self.query = query self.offset = offset class ChosenInlineResult(JsonDeserializable): @classmethod def de_json(cls, json_type): obj = cls.check_json(json_type) result_id = obj['result_id'] from_user = User.de_json(obj['from']) query = obj['query'] return cls(result_id, from_user, query) def __init__(self, result_id, from_user, query): """ This object represents a result of an inline query that was chosen by the user and sent to their chat partner. :param result_id: string The unique identifier for the result that was chosen. :param from_user: User The user that chose the result. :param query: String The query that was used to obtain the result. :return: ChosenInlineResult Object. """ self.result_id = result_id self.from_user = from_user self.query = query class InlineQueryResultArticle(JsonSerializable): def __init__(self, id, title, message_text, parse_mode=None, disable_web_page_preview=None, url=None, hide_url=None, description=None, thumb_url=None, thumb_width=None, thumb_height=None): self.type = 'article' self.id = id self.title = title self.message_text = message_text self.parse_mode = parse_mode self.disable_web_page_preview = disable_web_page_preview self.url = url self.hide_url = hide_url self.description = description self.thumb_url = thumb_url self.thumb_width = thumb_width self.thumb_height = thumb_height def to_json(self): json_dict = {'type': self.type, 'id': self.id, 'title': self.title, 'message_text': self.message_text} if self.parse_mode: json_dict['parse_mode'] = self.parse_mode if self.disable_web_page_preview: json_dict['disable_web_page_preview'] = self.disable_web_page_preview if self.url: json_dict['url'] = self.url if self.hide_url: json_dict['hide_url'] = self.hide_url if self.description: json_dict['description'] = self.description if self.thumb_url: json_dict['thumb_url'] = self.thumb_url if self.thumb_width: json_dict['thumb_width'] = self.thumb_width if self.thumb_height: json_dict['thumb_height'] = self.thumb_height return json.dumps(json_dict) class InlineQueryResultPhoto(JsonSerializable): def __init__(self, id, photo_url, mime_type=None, photo_width=None, photo_height=None, thumb_url=None, title=None, description=None, caption=None, message_text=None, parse_mode=None, disable_web_page_preview=None): self.type = 'photo' self.id = id self.photo_url = photo_url self.mime_type = mime_type self.photo_width = photo_width self.photo_height = photo_height self.thumb_url = thumb_url self.title = title self.description = description self.caption = caption self.message_text = message_text self.parse_mode = parse_mode self.disable_web_page_preview = disable_web_page_preview def to_json(self): json_dict = {'type': self.type, 'id': self.id, 'photo_url': self.photo_url} if self.mime_type: json_dict['mime_type'] = self.mime_type if self.photo_width: json_dict['photo_width'] = self.photo_width if self.photo_height: json_dict['photo_height'] = self.photo_height if self.thumb_url: json_dict['thumb_url'] = self.thumb_url if self.title: json_dict['title'] = self.title if self.description: json_dict['description'] = self.description if self.caption: json_dict['caption'] = self.caption if self.message_text: json_dict['message_text'] = self.message_text if self.parse_mode: json_dict['parse_mode'] = self.parse_mode if self.disable_web_page_preview: json_dict['disable_web_page_preview'] = self.disable_web_page_preview return json.dumps(json_dict) class InlineQueryResultGif(JsonSerializable): def __init__(self, id, gif_url, gif_width=None, gif_height=None, thumb_url=None, title=None, caption=None, message_text=None, parse_mode=None, disable_web_page_preview=None): self.type = 'gif' self.id = id self.gif_url = gif_url self.gif_width = gif_width self.gif_height = gif_height self.thumb_url = thumb_url self.title = title self.caption = caption self.message_text = message_text self.parse_mode = parse_mode self.disable_web_page_preview = disable_web_page_preview def to_json(self): json_dict = {'type': self.type, 'id': self.id, 'gif_url': self.gif_url} if self.gif_height: json_dict['gif_height'] = self.gif_height if self.gif_width: json_dict['gif_width'] = self.gif_width if self.thumb_url: json_dict['thumb_url'] = self.thumb_url if self.title: json_dict['title'] = self.title if self.caption: json_dict['caption'] = self.caption if self.message_text: json_dict['message_text'] = self.message_text if self.parse_mode: json_dict['parse_mode'] = self.parse_mode if self.disable_web_page_preview: json_dict['disable_web_page_preview'] = self.disable_web_page_preview return json.dumps(json_dict) class InlineQueryResultMpeg4Gif(JsonSerializable): def __init__(self, id, mpeg4_url, mpeg4_width=None, mpeg4_height=None, thumb_url=None, title=None, caption=None, message_text=None, parse_mode=None, disable_web_page_preview=None): self.type = 'mpeg4_gif' self.id = id self.mpeg4_url = mpeg4_url self.mpeg4_width = mpeg4_width self.mpeg4_height = mpeg4_height self.thumb_url = thumb_url self.title = title self.caption = caption self.message_text = message_text self.parse_mode = parse_mode self.disable_web_page_preview = disable_web_page_preview def to_json(self): json_dict = {'type': self.type, 'id': self.id, 'mpeg4_url': self.mpeg4_url} if self.mpeg4_width: json_dict['mpeg4_width'] = self.mpeg4_width if self.mpeg4_height: json_dict['mpeg4_height'] = self.mpeg4_height if self.thumb_url: json_dict['thumb_url'] = self.thumb_url if self.title: json_dict['title'] = self.title if self.caption: json_dict['caption'] = self.caption if self.message_text: json_dict['message_text'] = self.message_text if self.parse_mode: json_dict['parse_mode'] = self.parse_mode if self.disable_web_page_preview: json_dict['disable_web_page_preview'] = self.disable_web_page_preview return json.dumps(json_dict) class InlineQueryResultVideo(JsonSerializable): def __init__(self, id, video_url, mime_type, message_text=None, parse_mode=None, disable_web_page_preview=None, video_width=None, video_height=None, video_duration=None, thumb_url=None, title=None, description=None): self.type = 'video' self.id = id self.video_url = video_url self.mime_type = mime_type self.message_text = message_text self.parse_mode = parse_mode self.disable_web_page_preview = disable_web_page_preview self.video_width = video_width self.video_height = video_height self.video_duration = video_duration self.thumb_url = thumb_url self.title = title self.description = description def to_json(self): json_dict = {'type': self.type, 'id': self.id, 'video_url': self.video_url, 'mime_type': self.mime_type} if self.message_text: json_dict['message_text'] = self.message_text if self.parse_mode: json_dict['parse_mode'] = self.parse_mode if self.disable_web_page_preview: json_dict['disable_web_page_preview'] = self.disable_web_page_preview if self.video_width: json_dict['video_width'] = self.video_width if self.video_height: json_dict['video_height'] = self.video_height if self.video_duration: json_dict['video_duration'] = self.video_duration if self.thumb_url: json_dict['thumb_url'] = self.thumb_url if self.title: json_dict['title'] = self.title if self.description: json_dict['description'] = self.description return json.dumps(json_dict)