python apscheduler cron定時任務觸發介面自動化巡檢過程

$ pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler


class 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)