如何在 Python-telegram-bot 中使用 Jobqueue
How to use Jobqueue in Python-telegram-bot
我可以通过阅读 docs 很容易地制作一个机器人,但是 Jobqueue 没有按照它写的那样工作。 run_daily
方法使用 datetime.time
对象在特定时间发送消息,但这段代码既不发送消息也不显示任何错误。它只是保持 运行
import datetime
from telegram import bot
from telegram.ext import Updater
def callback_minute(bot, job):
bot.send_message(chat_id=475838704, text='PlEaSe wOrK!')
def main():
updater = Updater()
bot = updater.bot
job = updater.job_queue
dispatcher = updater.dispatcher
job.run_daily(callback_minute, time=datetime.time(6,33,00))
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
也许这会有所帮助:
from telegram.ext import Updater, CommandHandler
def daily_job(bot, update, job_queue):
""" Running on Mon, Tue, Wed, Thu, Fri = tuple(range(5)) """
bot.send_message(chat_id=<YOUR CHAT ID>, text='Setting a daily notifications!')
t = datetime.time(10, 00, 00, 000000)
job_queue.run_daily(notify_assignees, t, days=tuple(range(5)), context=update)
def notify_assignees(bot, job):
bot.send_message(chat_id=<CHAT ID>, text="Some text!")
updater = Updater(<BOT_TOKEN>)
updater.dispatcher.add_handler(CommandHandler('notify', daily_job, pass_job_queue=True))
updater.start_polling()
然后对机器人说 /notify
JobQueue Extention from python-telegram-bot
的简单使用示例
from telegram.ext import Updater, CommandHandler
def callback_alarm(bot, job):
bot.send_message(chat_id=job.context, text='Wait for another 10 Seconds')
def callback_timer(bot, update, job_queue):
bot.send_message(chat_id=update.message.chat_id,
text='Wait for 10 seconds')
job_queue.run_repeating(callback_alarm, 10, context=update.message.chat_id)
def Stop_timer(bot, update, job_queue):
bot.send_message(chat_id=update.message.chat_id,
text='Stopped!')
job_queue.stop()
updater = Updater("YOUR_TOKEN")
updater.dispatcher.add_handler(CommandHandler('start', callback_timer, pass_job_queue=True))
updater.dispatcher.add_handler(CommandHandler('stop', Stop_timer, pass_job_queue=True))
updater.start_polling()
/start
命令将启动 JobQueue 并以 5 秒的间隔发送一条消息,队列可以通过 /stop
命令停止。
使用以下代码:
from telegram.ext import CommandHandler, Updater
my_token = 'YOUR TOKEN'
updater = Updater(my_token, use_context=True)
job_queue = updater.job_queue
def send_message_job(context):
context.bot.send_message(chat_id='@YOUR CHANELL ID',text='job executed')
job_queue.run_repeating(send_message_job,interval=10.0,first=0.0)
我可以通过阅读 docs 很容易地制作一个机器人,但是 Jobqueue 没有按照它写的那样工作。 run_daily
方法使用 datetime.time
对象在特定时间发送消息,但这段代码既不发送消息也不显示任何错误。它只是保持 运行
import datetime
from telegram import bot
from telegram.ext import Updater
def callback_minute(bot, job):
bot.send_message(chat_id=475838704, text='PlEaSe wOrK!')
def main():
updater = Updater()
bot = updater.bot
job = updater.job_queue
dispatcher = updater.dispatcher
job.run_daily(callback_minute, time=datetime.time(6,33,00))
updater.start_polling()
updater.idle()
if __name__ == '__main__':
main()
也许这会有所帮助:
from telegram.ext import Updater, CommandHandler
def daily_job(bot, update, job_queue):
""" Running on Mon, Tue, Wed, Thu, Fri = tuple(range(5)) """
bot.send_message(chat_id=<YOUR CHAT ID>, text='Setting a daily notifications!')
t = datetime.time(10, 00, 00, 000000)
job_queue.run_daily(notify_assignees, t, days=tuple(range(5)), context=update)
def notify_assignees(bot, job):
bot.send_message(chat_id=<CHAT ID>, text="Some text!")
updater = Updater(<BOT_TOKEN>)
updater.dispatcher.add_handler(CommandHandler('notify', daily_job, pass_job_queue=True))
updater.start_polling()
然后对机器人说 /notify
JobQueue Extention from python-telegram-bot
的简单使用示例from telegram.ext import Updater, CommandHandler
def callback_alarm(bot, job):
bot.send_message(chat_id=job.context, text='Wait for another 10 Seconds')
def callback_timer(bot, update, job_queue):
bot.send_message(chat_id=update.message.chat_id,
text='Wait for 10 seconds')
job_queue.run_repeating(callback_alarm, 10, context=update.message.chat_id)
def Stop_timer(bot, update, job_queue):
bot.send_message(chat_id=update.message.chat_id,
text='Stopped!')
job_queue.stop()
updater = Updater("YOUR_TOKEN")
updater.dispatcher.add_handler(CommandHandler('start', callback_timer, pass_job_queue=True))
updater.dispatcher.add_handler(CommandHandler('stop', Stop_timer, pass_job_queue=True))
updater.start_polling()
/start
命令将启动 JobQueue 并以 5 秒的间隔发送一条消息,队列可以通过 /stop
命令停止。
使用以下代码:
from telegram.ext import CommandHandler, Updater
my_token = 'YOUR TOKEN'
updater = Updater(my_token, use_context=True)
job_queue = updater.job_queue
def send_message_job(context):
context.bot.send_message(chat_id='@YOUR CHANELL ID',text='job executed')
job_queue.run_repeating(send_message_job,interval=10.0,first=0.0)