CeleryBeatObject.py 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133
  1. # @Author : Rocky
  2. # @File : CeleryBeatObject.py
  3. # @Time : 2024/3/21 16:35
  4. import json
  5. import zoneinfo
  6. from django_celery_beat.models import PeriodicTask, IntervalSchedule, ClockedSchedule, CrontabSchedule
  7. from Model.models import TimeZoneInfo
  8. from Service.CommonService import CommonService
  9. class CeleryBeatObj:
  10. # celery beat对象,封装定时任务函数
  11. # https://github.com/celery/django-celery-beat
  12. @staticmethod
  13. def creat_interval_task(every, period, name, task, args=None, kwargs=None):
  14. """
  15. 创建间隔任务
  16. @param every: 时间间隔
  17. @param period: 单位,seconds,minutes,hours,days(或IntervalSchedule.SECONDS...)
  18. @param name: 任务名称
  19. @param task: 任务函数
  20. @param args: 参数
  21. @param kwargs: 参数
  22. @return:
  23. """
  24. if args is None:
  25. args = []
  26. args = json.dumps(args)
  27. if kwargs is None:
  28. kwargs = {}
  29. kwargs = json.dumps(kwargs)
  30. schedule, _ = IntervalSchedule.objects.get_or_create(every=every, period=period)
  31. PeriodicTask.objects.create(interval=schedule, name=name, task=task, args=args, kwargs=kwargs)
  32. @staticmethod
  33. def creat_clocked_task(name, task, time_stamp=0, timezone_offset=0, time_string='', args=None, kwargs=None):
  34. """
  35. 创建定时任务
  36. @param name: 任务名称
  37. @param task: 任务函数
  38. @param time_stamp: 时间戳
  39. @param timezone_offset: 时区偏移量
  40. @param time_string: 时间字符串
  41. @param args: 参数
  42. @param kwargs: 参数
  43. @return:
  44. """
  45. if args is None:
  46. args = []
  47. args = json.dumps(args)
  48. if kwargs is None:
  49. kwargs = {}
  50. kwargs = json.dumps(kwargs)
  51. # 不传时间戳时,将时间字符串转为时间戳
  52. if time_stamp == 0:
  53. time_stamp = CommonService.convert_to_timestamp(timezone_offset, time_string)
  54. clocked_time = CommonService.get_date_from_timestamp(time_stamp, timezone_offset)
  55. schedule, _ = ClockedSchedule.objects.get_or_create(clocked_time=clocked_time)
  56. PeriodicTask.objects.create(clocked=schedule, one_off=True, name=name, task=task, args=args, kwargs=kwargs)
  57. @staticmethod
  58. def creat_crontab_task(
  59. timezone_offset, name, task,
  60. minute='*', hour='*', day_of_week='*', day_of_month='*', month_of_year='*', args=None, kwargs=None):
  61. """
  62. 创建周期任务
  63. @param timezone_offset: 时区偏移量
  64. @param name: 任务名称
  65. @param task: 任务函数
  66. @param minute: 分
  67. @param hour: 时
  68. @param day_of_week: 周,1-7对应周一到周日,也可写mon,tue,wed,thu,fri,sat,sun
  69. @param day_of_month: 月
  70. @param month_of_year: 年
  71. @param args: 参数
  72. @param kwargs: 参数
  73. @return:
  74. """
  75. if args is None:
  76. args = []
  77. args = json.dumps(args)
  78. if kwargs is None:
  79. kwargs = {}
  80. kwargs = json.dumps(kwargs)
  81. time_zone_info_qs = TimeZoneInfo.objects.filter(tz=timezone_offset).values('zone_info')
  82. if time_zone_info_qs.exists():
  83. zone_info = time_zone_info_qs[0]['zone_info']
  84. timezone = zoneinfo.ZoneInfo(zone_info)
  85. schedule, _ = CrontabSchedule.objects.get_or_create(
  86. minute=minute,
  87. hour=hour,
  88. day_of_week=day_of_week,
  89. day_of_month=day_of_month,
  90. month_of_year=month_of_year,
  91. timezone=timezone)
  92. PeriodicTask.objects.create(crontab=schedule, name=name, task=task, args=args, kwargs=kwargs)
  93. @staticmethod
  94. def disable_task(name):
  95. """
  96. 暂停任务
  97. @param name: 任务名称
  98. @return:
  99. """
  100. periodic_task = PeriodicTask.objects.get(name=name)
  101. periodic_task.enabled = False
  102. periodic_task.save()
  103. @staticmethod
  104. def enable_task(name):
  105. """
  106. 恢复任务
  107. @param name: 任务名称
  108. @return:
  109. """
  110. periodic_task = PeriodicTask.objects.get(name=name)
  111. periodic_task.enabled = True
  112. periodic_task.save()
  113. @staticmethod
  114. def del_task(name):
  115. """
  116. 删除任务
  117. @param name: 任务名称
  118. @return:
  119. """
  120. PeriodicTask.objects.get(name=name).delete()