定时接口

Controller/TestApi.py

@@ -56,6 +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
 
 ACCESS_KEY = "AKIA2E67UIMD3CYTIWPA"
 SECRET_KEY = "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"
@@ -145,6 +146,10 @@ 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)
         else:
             return response.json(414)
 
@@ -995,3 +1000,17 @@ class testView(View):
                 500,
                 "error_line:{}, error_msg:{}".format(e.__traceback__.tb_lineno, repr(e)),
             )
+
+    @staticmethod
+    def add_apscheduler_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)
+        return response.json(0)
+
+    @staticmethod
+    def update_apscheduler_job(request_dict, response):
+        task_id = request_dict.get('task_id', None)
+        del_job(task_id)
+        return response.json(0)

Model/views.py

@@ -1,19 +1,9 @@
 from django.shortcuts import render
 
 # Create your views here.
-import time
 from apscheduler.schedulers.background import BackgroundScheduler
 from django_apscheduler.jobstores import DjangoJobStore
-from Ansjer.config import LOGGER
-
-
-def auto_hello():
-    now_time = time.time()
-    print('hello world:[{}]'.format(now_time))
-
 
 scheduler = BackgroundScheduler()
 scheduler.add_jobstore(DjangoJobStore(), 'default')
-
-scheduler.add_job(auto_hello, 'cron', day='*', hour=15, minute=30, replace_existing=True, id='test_job')
 scheduler.start()