futool-tiny-datahub/datahub/scheduletask/schedule.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2023/4/12 15:15
# @Author  : old tom
# @File    : schedule.py
# @Project : futool-tiny-datahub
# @Desc    : 定时任务配置
from apscheduler.triggers.cron import CronTrigger
# from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler

# 后台指定调度，配合fastapi使用
# BlockingScheduler: 调用start函数后会阻塞当前线程。当调度器是你应用中唯一要运行的东西时（如上例）使用。
# BackgroundScheduler: 调用start后主线程不会阻塞。当你不运行任何其他框架时使用，并希望调度器在你应用的后台执行。
# 每隔 1分钟 运行一次 job 方法
# scheduler.add_job(tick, trigger=CronExpTrigger.parse_crontab('0/1 * * * * * *'), kwargs={
#     "name": "bob"
# }
sch = BlockingScheduler()


class CronExpTrigger(CronTrigger):
    """
    重写cron触发器,支持6\7位 cron表达式
    7位：* * * * * * *
    秒、分、时、天、月、周、年
    """

    @classmethod
    def parse_crontab(cls, expr, timezone=None):
        values = expr.split()
        if len(values) == 6:
            # 6位转7位-->? 转 *-->末尾补充 *
            values[5] = '*'
            values = values + ['*']
        if len(values) != 7:
            raise ValueError('Wrong number of fields; got {}, expected 7'.format(len(values)))
        return cls(second=values[0], minute=values[1], hour=values[2], day=values[3], month=values[4],
                   day_of_week=values[5], year=values[6], timezone=timezone)