我们从Python开源项目中,提取了以下2个代码示例,用于说明如何使用telegram.ext()。
def __init__(self, queue, mutex, slaves): """ Initialization. Args: queue (queue.Queue): global message queue slaves (dict): Dictionary of slaves """ super().__init__(queue, mutex) self.slaves = slaves try: self.bot = telegram.ext.Updater(getattr(config, self.channel_id)['token']) except (AttributeError, KeyError): raise ValueError("Token is not properly defined. Please define it in `config.py`.") mimetypes.init(files=["mimetypes"]) self.admins = getattr(config, self.channel_id)['admins'] self.logger = logging.getLogger("plugins.%s.TelegramChannel" % self.channel_id) self.me = self.bot.bot.get_me() self.bot.dispatcher.add_handler(WhitelistHandler(self.admins)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("link", self.link_chat_show_list, pass_args=True)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("chat", self.start_chat_list, pass_args=True)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("recog", self.recognize_speech, pass_args=True)) self.bot.dispatcher.add_handler(telegram.ext.CallbackQueryHandler(self.callback_query_dispatcher)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("start", self.start, pass_args=True)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("extra", self.extra_help)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("help", self.help)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("unlink_all", self.unlink_all)) self.bot.dispatcher.add_handler(telegram.ext.CommandHandler("info", self.info)) self.bot.dispatcher.add_handler( telegram.ext.RegexHandler(r"^/(?P<id>[0-9]+)_(?P<command>[a-z0-9_-]+)", self.extra_call, pass_groupdict=True)) self.bot.dispatcher.add_handler(telegram.ext.MessageHandler( telegram.ext.Filters.text | telegram.ext.Filters.photo | telegram.ext.Filters.sticker | telegram.ext.Filters.document | telegram.ext.Filters.venue | telegram.ext.Filters.location | telegram.ext.Filters.audio | telegram.ext.Filters.voice | telegram.ext.Filters.video, self.msg )) self.bot.dispatcher.add_error_handler(self.error) self.MUTE_CHAT_ID = self.MUTE_CHAT_ID.format(chat_id=self.channel_id) # Truncate string by bytes # Written by Mark Tolonen # http://stackoverflow.com/a/13738452/1989455
def _download_file(self, tg_msg, file_obj, msg_type): """ Download media file from telegram platform. Args: tg_msg: Telegram message instance file_obj: File object msg_type: Type of message Returns: tuple of str[2]: Full path of the file, MIME type """ path = os.path.join("storage", self.channel_id) if not os.path.exists(path): os.makedirs(path) size = getattr(file_obj, "file_size", None) file_id = file_obj.file_id if size and size > telegram.constants.MAX_FILESIZE_DOWNLOAD: raise EFBMessageError("Attachment is too large. Maximum 20 MB. (AT01)") f = self.bot.bot.getFile(file_id) fname = "%s_%s_%s_%s" % (msg_type, tg_msg.chat.id, tg_msg.message_id, int(time.time())) fullpath = os.path.join(path, fname) f.download(fullpath) mime = getattr(file_obj, "mime_type", magic.from_file(fullpath, mime=True)) if type(mime) is bytes: mime = mime.decode() guess_ext = mimetypes.guess_extension(mime) or ".unknown" if guess_ext == ".unknown": self.logger.warning("File %s with mime %s has no matching extensions.", fullpath, mime) ext = ".jpeg" if mime == "image/jpeg" else guess_ext os.rename(fullpath, "%s%s" % (fullpath, ext)) fullpath = "%s%s" % (fullpath, ext) return fullpath, mime