khl.py khl.py
主页
文档
示例
  • khl.py (opens new window)
  • khl.py.docs (opens new window)
KOOK (opens new window)
主页
文档
示例
  • khl.py (opens new window)
  • khl.py.docs (opens new window)
KOOK (opens new window)
  • 机器人

    • 初始化
    • 命令
    • 定时任务
      • 1.interval
      • 2.cron
      • 3.date
      • 4.on_startup
      • 5.on_shutdown
    • 在玩状态
  • 消息

    • 消息类型
    • 上传和发送文件或图片
    • 卡片消息
  • 频道

    • 获取频道对象
    • 主动发送消息
  • 用户

    • 获取机器人用户对象
    • 获取用户对象和发送私聊信息
    • 机器人下线
  • 服务器

    • 获取服务器对象
    • 服务器角色操作
    • 设置用户昵称
  • 事件

    • 事件基础处理
  • 其他

    • KOOK客户端获取各类ID
    • 多文件注册机器人命令
    • 配置python的venv虚拟环境
    • 获取图片链接
    • 尝试自己调用api
    • OpenSSL本地证书无法获取错误
  • docs
  • 机器人
musnows
2023-05-28
目录

定时任务

机器人定时任务包含3中不同的方式,其是在APScheduler模块上进行的封装

# 1.interval

这是最常用的task类型,其作用是让机器人隔一段时间就来执行一下某一个函数

@bot.task.add_interval() # 在这里传入参数,设定任务时间
async def task_interval_def(): # 函数不需要入参
    ...

具体的参数可以参考编译器的代码补全,所有的参数类型都是字面意思

image-20230905150549277

比如按如下设定,那么机器人就会在开机后开始计时,每1分钟执行一次;

@bot.task.add_interval(minutes=1)
async def task1():
	...

你也可以传入多个时间,最终的执行时间是参数的和;

@bot.task.add_interval(minutes=26,seconds=20)
async def task2():
	...

# 2.cron

cron类型task是让机器人到指定时间运行函数,比如:

  • 每天的1点0分
  • 每周周1的0点
  • 每月1日的12点
  • ...

你可以通过不同的参数设定,来实现在不同的时间段执行对应的函数;

@bot.task.add_cron() # 传入时间设定参数
async def task_cron_def(): # 函数不需要入参
    ...

请注意,为了明确该函数执行的时间,必须要给add_cron传入timezone时区参数,如果想使用东八区,请设置参数为 Asia/Shanghai 字符串(其余时区自行百度)

# 每天早上8点0分执行函数 (东八区)
@bot.task.add_cron(hour=8, minute=0, timezone="Asia/Shanghai")
async def auto_skin_notify_task():
	...

具体的传参参考IDE的代码补全

image-20230905151237548

在APScheduler的类中可以详细看到每一个入参应该如何传入参数;

其中day_of_week参数既可以传入每周的字符串,如mon代表周一、sun代表周日;又可以传入0到6代表周一到周日。请注意,在这个函数中,每一周的第一天始终是周一!

class CronTrigger(BaseTrigger):
    """
    Triggers when current time matches all specified time constraints,
    similarly to how the UNIX cron scheduler works.

    :param int|str year: 4-digit year
    :param int|str month: month (1-12)
    :param int|str day: day of month (1-31)
    :param int|str week: ISO week (1-53)
    :param int|str day_of_week: number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
    :param int|str hour: hour (0-23)
    :param int|str minute: minute (0-59)
    :param int|str second: second (0-59)
    :param datetime|str start_date: earliest possible date/time to trigger on (inclusive)
    :param datetime|str end_date: latest possible date/time to trigger on (inclusive)
    :param datetime.tzinfo|str timezone: time zone to use for the date/time calculations (defaults
        to scheduler timezone)
    :param int|None jitter: delay the job execution by ``jitter`` seconds at most

    .. note:: The first weekday is always **monday**.
    """

再次提醒,不管你使用那个参数设定运行时间,都需要传入时区参数明确目标时区。

# 3.date

这个task和前两者不同,它只会在设定的时间执行一次任务;

同时,它也可以不传入参数,则函数会在机器人启动后立即执行(注意区分start_up任务)

@bot.task.add_date()
async def test_date_task():
    ...

比如按如下方式传入参数,那么机器人就会在启动后过10秒执行一次这个函数

from datetime import datetime,timedelta

@bot.task.add_date(( datetime.now() + timedelta(seconds=10)),timezone='Asia/Shanghai')
async def test_date_task():
    ...

同样的,为了明确运行时间,也需要传入timezone时区参数;如果你担心出错,那就无差别地将所有task都传入时区参数,保证无误;

# 4.on_startup

除了上面3个基于APScheduler的任务类型之外,还有一个专门的启动任务,其使用方法如下

@bot.on_startup
async def bot_start_task(bot: Bot):
	...

这个任务会在机器人启动前执行!

  • add_date不传参数时,会在机器人启动后执行,此时机器人已经可以响应命令
  • on_startup会在机器人启动之前执行,此时机器人不可以响应命令

你可以根据自己的需要,选择合适的task类型

# 5.on_shutdown

顾名思义,这个函数会在进程收到信号退出的时候执行,比如键入CTRL+C(在LINUX下是2号信号)机器人在进程退出之前执行。

适合在内部做一些垃圾回收操作,比如释放数据库链接、将dict的内容以json方式保存到本地文件等操作...

@bot.on_shutdown
async def test_shutdown_task(b:Bot):
    ...

除了使用on_shutdown,你还可以给机器人设立一个只有开发者可以执行的机器人退出命令,并在命令内部实现垃圾回收+安全退出操作。为了避免其他人错误调用这个命令,你可以通过判断消息来源的用户ID的方式,让该命令只能由一部分用户调用。

import os,aiohttp
BOT_TOKEN = "你的机器人token"
MASTER_ID = "预先定义的管理员用户id"

async def bot_offline():
    """下线机器人的api调用"""
    kook_headers = {f'Authorization': f"Bot {BOT_TOKEN}"}
    url = "https://www.kookapp.cn" + "/api/v3/user/offline"
    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=kook_headers) as response:
            res = json.loads(await response.text())
    return res


@bot.command(name='kill')
async def kill_bot_cmd(msg: Message, *arg):
    """机器人退出命令"""
    try:
        print(f"recv /kill cmd from Au:{msg.author_id}")
        # 判断当前用户是否为预定义的管理员用户
        if msg.author_id != MASTER_ID:
            return await msg.reply("您没有权限执行本命令")

        # 进行垃圾回收操作,如释放数据库链接等
        ...

        # 提示用户即将退出机器人
        await msg.reply(f"[KILL] 保存全局变量成功,bot下线")
        res = await bot_offline() # 调用接口下线bot
        print(f"Au:{msg.author_id} | [KILL] bot-off | {res}\n")
        os._exit(0)  # 正常退出程序
    except Exception as result:
        print(f"error in /kill cmd",traceback.format_exc())
        os.abort()  # 有异常也强制退出程序
上次更新: 2023/11/06, 23:13:37
命令
在玩状态

← 命令 在玩状态→

Theme by Vdoing | Copyright © 2023-2025 khl.py | MIT License
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式