python apscheduler cron定时任务触发接口自动化巡检怎么实现

$ pip install apscheduler

from apscheduler.schedulers.blocking import BlockingSchedulerclass Timing:    def __init__(self, start_date, end_date, hour=None):        self.start_date = start_date        self.end_date = end_date        self.hour = hour    def cron(self, job, *value_list):        """cron格式 在特定时间周期性地触发"""        # year (int 或 str) – 年，4位数字        # month (int 或 str) – 月 (范围1-12)        # day (int 或 str) – 日 (范围1-31)        # week (int 或 str) – 周 (范围1-53)        # day_of_week (int 或 str) – 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)        # hour (int 或 str) – 时 (范围0-23)        # minute (int 或 str) – 分 (范围0-59)        # second (int 或 str) – 秒 (范围0-59)        # start_date (datetime 或 str) – 最早开始日期(包含)        # end_date (datetime 或 str) – 分 最晚结束时间(包含)        # timezone (datetime.tzinfo 或str) – 指定时区        scheduler = BlockingScheduler()        scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,                          args=[*value_list])        scheduler.start()    def interval(self, job, *value_list):        """interval格式 周期触发任务"""        # weeks (int) - 间隔几周        # days (int)  - 间隔几天        # hours (int) - 间隔几小时        # minutes (int) - 间隔几分钟        # seconds (int) - 间隔多少秒        # start_date (datetime 或 str) - 开始日期        # end_date (datetime 或 str) - 结束日期        # timezone (datetime.tzinfo 或str) - 时区        scheduler = BlockingScheduler()        # 在 2019-08-29 22:15:00至2019-08-29 22:17:00期间，每隔1分30秒 运行一次 job 方法        scheduler.add_job(job, 'interval', minutes=1, seconds=30, start_date=self.start_date,                          end_date=self.end_date, args=[*value_list])        scheduler.start()    @staticmethod    def date(job, *value_list):        """date格式 特定时间点触发"""        # run_date (datetime 或 str) - 作业的运行日期或时间        # timezone (datetime.tzinfo 或 str)  - 指定时区        scheduler = BlockingScheduler()        # 在 2019-8-30 01:00:01 运行一次 job 方法        scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=[*value_list])        scheduler.start()

if __name__ == '__main__':    file_list = ["test_shiyan.py", "MeetSpringFestival.py"]    # run_py(file_list)    case_list = ["test_case_01", "test_case_02"]    # run_case(test_sample, case_list)    dingDing_list = [2, case_list, test_sample]    # run_dingDing(*dingDing_list)    Timing('2022-02-15 00:00:00', '2022-02-16 00:00:00', '0-23').cron(run_dingDing, *dingDing_list)

def cron(self, job, *value_list):        """cron格式 在特定时间周期性地触发"""        scheduler.add_job(job, 'cron', start_date=self.start_date, end_date=self.end_date, hour=self.hour,                                  args=[*value_list])

 class Config(object):     SCHEDULER_JOBSTORES = {        'default': RedisJobStore(db=3,host='0.0.0.0', port=6378,password='******'),    }     SCHEDULER_EXECUTORS = {        'default': {'type': 'processpool', 'max_workers': 50}  #用进程池提升任务处理效率    }     SCHEDULER_JOB_DEFAULTS = {        'coalesce': True,   #积攒的任务只跑一次        'max_instances': 1000, #支持1000个实例并发       'misfire_grace_time':600 #600秒的任务超时容错    }     SCHEDULER_API_ENABLED = True

 {"id": "9586_site_status","name": "9586_site_status","func": "monitor_scheduler:monitor_site_status","args": [9586,"http://sl.jxcn.cn/",1000,100,200,"",0,2],"kwargs": {},"trigger": "interval","start_date": "2018-09-14T00:00:00+08:00","end_date": "2018-12-31T00:00:00+08:00","minutes": 15,"misfire_grace_time": 10,"max_instances": 3000,"next_run_time": "2018-10-24T18:00:00+08:00"}

 scheduler.modify_job(func=func, id=id, args=args, trigger=trigger, minutes=minutes,start_date=start_date,end_date=end_date,misfire_grace_time=600)