定时任务

约 1049 字大约 4 分钟

2025-03-27

BasePlugin 提供了注册定时任务的功能.

注册定时任务

解析时间参数为调度配置字典，支持格式:
- 一次性任务: 'YYYY-MM-DD HH:MM:SS' 或 GitHub Action格式 'YYYY:MM:DD-HH:MM:SS'
- 每日任务: 'HH:MM'
- 间隔任务: 
    * 基础单位: '120s', '2h30m', '0.5d'
    * 冒号分隔: '00:15:30' (时:分:秒)
    * 自然语言: '2天3小时5秒'

一般在 BasePlugin.on_load() 方法中时注册定时任务.

函数原型参考

class BasePlugin:
    @final
    def add_scheduled_task(
        self,
        job_func: Callable,
        name: str,
        interval: Union[str, int, float],
        conditions: Optional[List[Callable[[], bool]]] = None,
        max_runs: Optional[int] = None,
        args: Optional[Tuple] = None,
        kwargs: Optional[Dict] = None,
        args_provider: Optional[Callable[[], Tuple]] = None,
        kwargs_provider: Optional[Callable[[], Dict[str, Any]]] = None,
    ) -> bool:
        """
        添加定时任务

        Args:
            job_func (Callable): 要执行的任务函数
            name (str): 任务唯一标识名称
            interval (Union[str, int, float]): 调度时间参数
            conditions (Optional[List[Callable]]): 执行条件列表
            max_runs (Optional[int]): 最大执行次数
            args (Optional[Tuple]): 静态位置参数
            kwargs (Optional[Dict]): 静态关键字参数
            args_provider (Optional[Callable]): 动态位置参数生成函数
            kwargs_provider (Optional[Callable]): 动态关键字参数生成函数

        Returns:
            bool: 是否添加成功

        Raises:
            ValueError: 当参数冲突或时间格式无效时
        """

实际案例

又见早八提醒插件.

class ZaoBa(BasePlugin):
    name = "ZaoBa" # 插件名称
    version = "0.0.1" # 插件版本

    async def on_load(self):
        # 插件加载时执行的操作, 可缺省
        self.add_scheduled_task(
            job_func=self.zaoba, 
            name="早八问候", 
            interval="08:00",
            max_runs=10, 
            args=("早八人", ),
        )
        self.add_scheduled_task(
            job_func=self.remind, 
            name="起床提醒", 
            interval="30s", 
            max_runs=10, 
            args_provider=lambda:(
                datetime.datetime.now().hour, 
                datetime.datetime.now().minute
            ),
        )
        self.add_scheduled_task(
            job_func=self.exam, 
            name="考试提醒",
            interval="2025-03-27 13:12:20",
            kwargs={"subject":"物理"}
        )
    
    def zaoba(self, extra):
        print("你好, 早八人")
        print(extra)
    
    def remind(self, hour, minute):
        print(f"起床了, 已经 {hour} 点 {minute} 分了")
        
    def exam(self, subject):
        print(f"要考试了, 是 {subject}")

不同类型的任务

通过 interval 参数来明确任务类型.

一次性任务: 传入字符串格式为 YYYY-MM-DD HH:MM:SS 或 GitHub Action 格式 YYYY:MM:DD-HH:MM:SS. 在这个时间点执行一次.
间隔任务: 传入字符串格式为 120s, 2h30m, 0.5d 或冒号分隔格式 00:15:30 (时:分:秒) 或自然语言格式 2天3小时5秒. 从当前开始, 经过指定时间间隔后执行一次.
每日任务: 传入字符串格式为 HH:MM. 每天在这个时间点执行一次.

任务的时间精度有限, 大约有 5s 的误差.

任务回调函数

job_func 参数为任务回调函数, 该函数会在任务触发时执行. 会传入一些参数, 参数传入规则如下:

提示

args 参数必须写为元组, Python 语法中, (expr) 被解析为一个表达式, 而不是元组, (expr,) 才会被解析为元组.

args 位置参数:
- 通过 args 静态指定, 在 add_scheduled_task 中传入一个 args 元组可以静态指定参数.
- 通过 args_provider 动态指定, 在 add_scheduled_task 中传入一个 args_provider 函数可以动态指定参数. 函数的返回值必须是元组, 会在任务触发时调用这个函数并生成 job_func 的位置参数.
- 两者互斥
kwargs 命名参数:
- 通过 kwargs 静态指定, 在 add_scheduled_task 中传入一个 kwargs 字典可以静态指定参数.
- 通过 kwargs_provider 动态指定, 在 add_scheduled_task 中传入一个 kwargs_provider 函数可以动态指定参数. 函数的返回值必须是字典, 会在任务触发时调用这个函数并生成 job_func 的命名参数.
- 两者互斥