优化定任务 增加时间点任务

Controller/TestApi.py

@@ -56,7 +56,7 @@ from Object.ResponseObject import ResponseObject
 from Object.TokenObject import TokenObject
 from Service.CommonService import CommonService
 from Service.VodHlsService import SplitVodHlsObject
-from Object.ApschedulerObject import add_job, del_job
+from Object.ApschedulerObject import ApschedulerObject
 
 ACCESS_KEY = "AKIA2E67UIMD3CYTIWPA"
 SECRET_KEY = "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"
@@ -146,10 +146,12 @@ class testView(View):
             return self.ali_text_review(request_dict, response)
         elif operation == "ali_image_review":  # 阿里云图片审核
             return self.ali_image_review(request_dict, response)
-        elif operation == "add_apscheduler":  # 新增定时任务
-            return self.add_apscheduler_job(request_dict, response)
-        elif operation == "update_apscheduler":  # 修改定时任务
-            return self.update_apscheduler_job(request_dict, response)
+        elif operation == "add_cron_apscheduler":  # 新增定时任务
+            return self.add_cron_job(request_dict, response)
+        elif operation == "add_date_apscheduler":  # 新增定时任务
+            return self.add_date_job(request_dict, response)
+        elif operation == "del_apscheduler":  # 删除定时任务
+            return self.del_apscheduler_job(request_dict, response)
         else:
             return response.json(414)
 
@@ -1002,15 +1004,27 @@ class testView(View):
             )
 
     @staticmethod
-    def add_apscheduler_job(request_dict, response):
+    def add_cron_job(request_dict, response):
         task_id = request_dict.get('task_id', None)
         hour = request_dict.get('hour', None)
         minute = request_dict.get('minute', None)
-        add_job(task_id, hour, minute)
+        day_of_week = request_dict.get('day_of_week', None)
+        apscheduler_obj = ApschedulerObject()
+        apscheduler_obj.cron_job(task_id, day_of_week, hour, minute)
         return response.json(0)
 
     @staticmethod
-    def update_apscheduler_job(request_dict, response):
+    def add_date_job(request_dict, response):
         task_id = request_dict.get('task_id', None)
-        del_job(task_id)
+        time_stamp = request_dict.get('time_stamp', None)
+        time_stamp = int(time_stamp)
+        apscheduler_obj = ApschedulerObject()
+        apscheduler_obj.date_job(task_id, time_stamp)
+        return response.json(0)
+
+    @staticmethod
+    def del_apscheduler_job(request_dict, response):
+        task_id = request_dict.get('task_id', None)
+        apscheduler_obj = ApschedulerObject()
+        apscheduler_obj.del_job(task_id)
         return response.json(0)

Object/ApschedulerObject.py

@@ -3,20 +3,28 @@ from apscheduler.schedulers.background import BackgroundScheduler
 from django_apscheduler.jobstores import DjangoJobStore
 from Ansjer.config import LOGGER
 from django_apscheduler.models import DjangoJob
+import datetime
 
 
-def auto_hello():
-    now_time = time.time()
-    print('hello world:[{}]'.format(now_time))
+class ApschedulerObject:
+    def __init__(self):
+        self.scheduler = BackgroundScheduler()
+        self.scheduler.add_jobstore(DjangoJobStore(), 'default')
+        self.scheduler.start()
 
+    @staticmethod
+    def auto_hello():  # 任务
+        now_time = time.time()
+        print('hello world:[{}]'.format(now_time))
 
-def add_job(task_id, hour, minute):
-    scheduler = BackgroundScheduler()
-    scheduler.add_jobstore(DjangoJobStore(), 'default')
-    scheduler.add_job(auto_hello, 'cron', day='*', hour=hour, minute=minute, replace_existing=True, id=task_id,
-                      max_instances=1, coalesce=True)
-    scheduler.start()
+    def cron_job(self, task_id, day_of_week, hour, minute):  # 定时任务
+        self.scheduler.add_job(self.auto_hello, 'cron', day_of_week=day_of_week, hour=hour, minute=minute,
+                               replace_existing=True, id=task_id, max_instances=1, coalesce=True)
 
+    def date_job(self, task_id, time_stamp):  # 时间点任务
+        self.scheduler.add_job(self.auto_hello, 'date', run_date=datetime.datetime.fromtimestamp(time_stamp),
+                               replace_existing=True, id=task_id, max_instances=1, coalesce=True)
 
-def del_job(task_id):
-    DjangoJob.objects.filter(id=task_id).delete()
+    @staticmethod
+    def del_job(task_id):  # 删除任务
+        DjangoJob.objects.filter(id=task_id).delete()