Python telegram 模块，error() 实例源码

def poll(self):
        """
        Message polling process.
        """
        self.polling_from_tg()
        while True:
            try:
                m = self.queue.get()
                if m is None:
                    break
                self.logger.info("Got message from queue\nType: %s\nText: %s\n----" % (m.type, m.text))
                threading.Thread(target=self.process_msg, args=(m,)).start()
                self.queue.task_done()
                self.logger.info("Msg sent to TG, task_done marked.")
            except Exception as e:
                self.logger.error("Error occurred during message polling")
                self.logger.error(repr(e))
                self.bot.stop()
                self.poll()

        self.logger.debug("Gracefully stopping %s (%s).", self.channel_name, self.channel_id)
        self.bot.stop()
        self.logger.debug("%s (%s) gracefully stopped.", self.channel_name, self.channel_id)

def _get_episode_info(self, tvdb_id: int, episode: Episode) -> tv_info.EpisodeInfo:
        attempt = 1
        sleep_tm = 1

        while 1:
            try:
                return tv_info.EpisodeInfo(self.tvdb, tvdb_id, episode.season, episode.num)

            except (ConnectionError, BadData):
                self._logger.warning(
                    'tvdb server in trouble; attempt={} sleep={}'.format(attempt, sleep_tm),
                    exc_info=True)
                time.sleep(sleep_tm)
                attempt += 1
                if sleep_tm < 60:
                    sleep_tm *= 2

            except Exception:
                self._logger.exception('unknown error')
                return

def send_tg(self, msg: str):
        for i in self.chat_ids:
            attempt = 1
            sleep_tm = 1
            while 1:
                try:
                    self.bot.sendMessage(i, msg, parse_mode='Markdown')

                except telegram.error.NetworkError:
                    self._logger.warning(
                        'telegram servers in trouble; attempt={} sleep={}'.format(
                            attempt, sleep_tm))
                    time.sleep(sleep_tm)
                    attempt += 1
                    if sleep_tm < 60:
                        sleep_tm *= 2
                    continue

                except telegram.TelegramError:
                    self._logger.exception('failed sending telegram')

                break

def signalHandler(self, signal, frame):
        # always disable buzzer
        if self.config['buzzer']['enable']:
            gpio = self.config['buzzer']['gpio']
            self.GPIO.output(gpio, 0)
            self.GPIO.cleanup()

        msg = 'Caught signal %d, terminating now.' % signal
        self.logger.error(msg)
        for owner_id in self.config['telegram']['owner_ids']:
            try:
                self.bot.sendMessage(chat_id=owner_id, text=msg)
            except Exception as e:
                pass
        sys.exit(1)

def webhook (request, bot_token):

    #verifico la validità del token
    bot = DjangoTelegramBot.getBot(bot_id=bot_token, safe=False)
    if bot is None:
        logger.warn('Request for not found token : {}'.format(bot_token))
        return JsonResponse({})

    try:
        data = json.loads(request.body.decode("utf-8"))

    except:
        logger.warn('Telegram bot <{}> receive invalid request : {}'.format(bot.username, repr(request)))
        return JsonResponse({})

    dispatcher = DjangoTelegramBot.getDispatcher(bot_token, safe=False)
    if dispatcher is None:
        logger.error('Dispatcher for bot <{}> not found : {}'.format(bot.username, bot_token))
        return JsonResponse({})

    try:
        update = telegram.Update.de_json(data, bot)
        dispatcher.process_update(update)
        logger.debug('Bot <{}> : Processed update {}'.format(bot.username, update))
    # Dispatch any errors
    except TelegramError as te:
        logger.warn("Bot <{}> : Error was raised while processing Update.".format(bot.username))
        dispatcher.dispatchError(update, te)

    # All other errors should not stop the thread, just print them
    except:
        logger.error("Bot <{}> : An uncaught error was raised while processing an update\n{}".format(bot.username, sys.exc_info()[0]))

    finally:
        return JsonResponse({})

def error(bot, update, error):
    logger.warn('Update "%s" caused error "%s"' % (update, error))

def main():
    #define bot
    bot = telegram.Bot(TOKEN)

    # Create the EventHandler and pass it your bot's token.
    updater = Updater(TOKEN)

    # Get the dispatcher to register handlers
    #bot.GetUpdate()
    dp = updater.dispatcher
    # on different commands - answer in Telegram
    print("Loading Modules")

    #init module
    for mod in commands:
        print(mod.name)
        dp.add_handler(CommandHandler(mod.name,mod.cb,pass_args=mod.args))

    dp.add_handler(MessageHandler(Filters.status_update.new_chat_members,onJoinHandler))

    # log all errors
    dp.add_error_handler(error)

    # Start the Bot
    updater.start_polling(clean=True)

    # Run the bot until the you presses Ctrl-C or the process receives SIGINT,
    # SIGTERM or SIGABRT. This should be used most of the time, since
    # start_polling() is non-blocking and will stop the bot gracefully.
    updater.idle()

def telegram_posts_sender(self):
        while True:
            time_to_wait = (self.wait_seconds // 3) + 1
            posts_to_sent = Crawler.get_post_to_send()
            grouped_posts = groupby(posts_to_sent, key=lambda x: x.to_send_id)
            pending_msgs_to_send = False
            for chat_id, posts in grouped_posts:
                pending_msgs_to_send = True
                posts = list(posts)[:self.post_chunk]
                if settings.DEBUG:
                    print('Sending {} new posts to chat {}'.format(len(posts), chat_id))
                for post in posts:
                    try:
                        self.telegram.sendMessage(chat_id=chat_id, text=post.description)
                        if post.image:
                            try:
                                self.telegram.sendPhoto(chat_id=chat_id, photo=post.image)
                            except BadRequest:
                                self.send_error('ERROR sending picture to {}'.format(post))
                        post.status = 'SENT'
                    except RetryAfter as e:
                        # Flood control exceeded. Retry in 175 seconds
                        self.send_error('RetryAfter error, waiting {} seconds'.format(e.retry_after))
                        time_to_wait = max(e.retry_after, time_to_wait)
                        post.status = 'ERROR'
                    except Exception as e:
                        exc_type, exc_value, exc_traceback = sys.exc_info()
                        error_stack = 'Send_updates ERROR: {}\n'.format(e)
                        error_stack += "".join(traceback.format_exception(exc_type, exc_value, exc_traceback))
                        self.send_error(error_stack)
                        post.status = 'ERROR'
                Crawler.save_posts(posts)

            if pending_msgs_to_send:
                sleep_time = 3
            else:
                sleep_time = time_to_wait
            gevent.sleep(sleep_time)

def send_error(self, text):
        """
        Sending text using telegram and error channel
        :param text: text to be sent
        """
        try:
            self.telegram.sendMessage(chat_id=settings.ERRORS_TELEGRAM_CHAT_ID, text=text)
        except Exception as e:
            print('Error trying to send the error to a ErrorChat {}'.format(e))
        print(text)

def send_help(bot, chat_id, command, error=""):
    if command not in help_messages:
        raise Exception('Unknown command: ' + command)

    message = ""
    if error != "":
        message = "Error: " + error + "\n"

    message += help_messages[command]
    try:
        __actual_send_message(chat_id=chat_id, bot=bot, text=message, parse_mode=ParseMode.MARKDOWN)
    except TelegramError as e:
        raise e

def send_error(bot, chat_id, name):
    if name not in error_messages:
        raise Exception('Unknown error messages: ' + name)

    __actual_send_message(bot, chat_id=chat_id, text=error_messages[name])

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 _reply_error(bot, update, errmsg):
        """
        A wrap that directly reply a message with error details.

        Returns:
            telegram.Message: Message sent
        """
        return bot.send_message(update.message.chat.id, errmsg, reply_to_message_id=update.message.message_id)

def check_token(token: str):
    from telegram.error import InvalidToken
    import string
    if token is None:
        return
    if len(token) != 48:
        raise InvalidToken()

    for symbol in token:
        if symbol not in string.hexdigits:
            raise InvalidToken()

def main():
    from pip import main as pip_main
    print("Setting up SpinEverydayBot...")

    ver = version.split()[0].split('.')
    if not (int(ver[0]) >= 3 and int(ver[1]) >= 6):
        exit("ERROR: You need to install Python 3.6 or newer to use this bot")

    try:
        import telegram
    except ImportError:
        print("WARNING: 'python-telegram-bot' package is not installed. Installing...")
        if pip_main(['install', 'python-telegram-bot']) != 0:
            exit("""ERROR: An error occurred while installing packages.
Check that you're running this script as root or administrator""")
        import telegram

    del telegram

    try:
        import TeleSocketClient
        del TeleSocketClient
    except ImportError:
        if input("Do you want to use TeleSocket Service? (y/N): ").lower().strip() == "y":
            print("WARNING: 'TeleSocketClient' package is not installed. Installing...")
            if pip_main(['install', 'TeleSocketClient']) != 0:
                exit("""ERROR: An error occurred while installing packages.
            Check that you're running this script as root or administrator""")
            import TeleSocketClient
            del TeleSocketClient

    if not path.exists("config.py"):
        print("WARNING: 'config.py' doesn't exist, copying example...")
        copy("config_example.py", "config.py")
        input("Now change example values to yours in file 'config.py' and press <Enter>")

    check_config()
    print("Setup finished")
    print("Now you can run your bot via 'python3.6 bot.py' command")
    exit(0)