我有一个执行例行程序的自定义代码,如果出现问题,我想在Telegram中向自己发送一条消息。就我而言,我将python-telegram-bot
库与apscheduler
及其侦听器一起使用,可以捕获某些事件。我想出了这样的工作代码,但是我的问题是:是否可以使它更好,即不使用全局变量?这样做是为了克服侦听器不接受bot发送消息所需的参数的问题。
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()
Telegram Bot API非常简单,您刚刚可以向该URL发送HTTP GET请求:https://api.telegram.org/bot_token_/sendMessage?chat_id=123&text=Hello%20World!
只需与Botfather创建一个机器人并向Bot发送消息。使用Botfather指定的令牌和以下URL:https://api.telegram.org/bot_token_/getUpdates
您可以获取发送到Bot和chat_id的消息。
最简单的方法是使用请求模块,并将更新程序的输出作为文本参数发送到第一个URL。