skip to Main Content

I have a custom code that does its routine and I want to send a message to myself in Telegram if something goes wrong. In my case I use python-telegram-bot library along with apscheduler and its listeners, where certain events could be catched.
I came up with such working code, but my question is: is it possible to make it better, namely without using global variable? This was done to overcome the problem that listeners do not accept arguments needed for bot to send a message.

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
from telegram.ext import Updater, CommandHandler
import copy
import my_custom_library

saved_update = None

def my_listener(event): # not related with bot
    if event.exception:
        if saved_update is not None:
            alert(saved_update, 'Scheduler threw event.exception.') # should have bot related args
    else:
        record = event.retval # get returned value from my_custom_library.repetitive_function
        try:
            processed_record = my_custom_library.my_unsafe_business_logic(record) # something might go wrong here
            my_custom_library.add_to_db(processed_record) # and here
        except Exception as e:
            if saved_update is not None:
                alert(saved_update, e) # should have bot related args

def start(update, context):
    global saved_update
    saved_update = copy.deepcopy(update) # this is what I don't like
    update.message.reply_text('You have subscribed for notifications.')

def alert(update, reason):
    update.message.reply_text('Something went wrong: {}'.format(reason))

def main():
    scheduler = BackgroundScheduler()
    scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
    scheduler.add_job(my_custom_library.repetitive_function, args=(my_args,), trigger='interval', minutes=1)
    scheduler.start()

    # bot
    updater = Updater(TOKEN, use_context=True)
    dp = updater.dispatcher
    dp.add_handler(CommandHandler("start", callback=start))
    updater.start_polling()
    updater.idle()


if __name__ == '__main__':
    main()

2

Answers


  1. The Telegram Bot API is fairly simple, you just ned to send an HTTP GET Request to this URL:
    https://api.telegram.org/bot_token_/sendMessage?chat_id=123&text=Hello%20World!

    Just create a bot with Botfather and send the Bot a message.
    With the specified Token from Botfather and this URL:
    https://api.telegram.org/bot_token_/getUpdates

    You can get the messages which were sent to the Bot and the chat_id.

    The simplest way would be to use the requests module and send the output of the updater to the first URL as the text parameter.

    Login or Signup to reply.
  2. In my scripts I often use callbacks. It’s a clean solution that provides separation between the two scripts. The send_message function can accept kwargs, so you can essentially create a dictionary, and update it when sending the message. The first part (channel ID) is something you know on the bot-side, and the second part (the text itself) is something you know one the third-party side.

    In your third-party, provide a set_cb function that accepts a callback and a dictionary. Like so:

    def set_cb(self, callback, params):
        self.callback = callback
        self.callback_params = params
    

    In your bot script, set the callback before updater.idle()

    # Start the Bot
    updater = Updater("TOKEN")
    updater.start_polling()
    
    # Set the callback
    r.set_cb(updater.bot.send_message, {"chat_id": 123456})
    

    Then, in your third-party, once you want to send a message, simply add the message text call the following:

    self.callback_params.update({"text": text})
    self.callback(**self.callback_params) # converts the dict to kwargs
    
    Login or Signup to reply.
Please signup or login to give your own answer.
Back To Top
Search