二:定时任务
main.py
import asyncio
import datetime
from mirai import Mirai, WebSocketAdapter, Startup, Shutdown
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
if __name__ == '__main__':
bot = Mirai(
qq=12345678, # 改成你的机器人的 QQ 号
adapter=WebSocketAdapter(
verify_key='yirimirai', host='localhost', port=8080
)
)
_task = None
@bot.on(Startup)
async def start_scheduler(_):
async def timer():
today_finished = False # 设置变量标识今天是会否完成任务,防止重复发送
while True:
await asyncio.sleep(1)
now = datetime.dtaetime.now()
if now.hour == 7 and now.minute == 30 and not today_finished: # 每天早上 7:30 发送早安
await bot.send_group_message(12345678, "早安")
today_finished = True
if now.hour == 7 and now.minute == 31:
today_finished = False # 早上 7:31,重置今天是否完成任务的标识
global _task
_task = asyncio.create_task(timer())
@bot.on(Shutdown)
async def stop_scheduler(_):
# 退出时停止定时任务
if _task and not task.done():
_task.cancel()
scheduler = AsyncIOScheduler()
@bot.on(Startup)
def start_scheduler(_):
scheduler.start() # 启动定时器
@bot.on(Shutdown)
def stop_scheduler(_):
scheduler.shutdown(True) # 结束定时器
@scheduler.scheduled_job(CronTrigger(hour='*', minute=0))
async def timer():
await bot.send_group_message(12345678, f"为您报时: {datetime.datetime.now().hour}:00")
bot.run()