接着上期代码内容，继续完善优化系统功能。
本次增加配置定时调度任务功能，学习任务系统定时任务管理添加的定时学习任务，需要通过配置调度任务，定时发布周期性的学习任务。
以及每天定时发送学生用户属性值，积分值等信息到学生用户知晓。以及其他需要定时调度的任务都可以配置到定时任务中，方便及时提醒自己。
第一步：熟悉Django框架的apscheduler模块
Django-apscheduler是一个基于apscheduler库的Python定时任务调度库，可用于轻松执行定时任务。
它支持三种不同的调度方式：固定时间间隔、固定时间点（日期）和crontab命令。
使用django-apscheduler，您可以轻松设置和管理定时任务，以在预定的时间执行特定的操作。
1，django-apscheduler安装
pip install django-apscheduler
2，修改django项目配置文件
./mysite/mysite/settings.py:
setting.py配置文件的INSTALLED_APPSl里面进行注册
INSTALLED_APPS = ( # ... ... "django_apscheduler", )
3，数据迁移
python manage.py migrate
迁移完成后数据库会创建如下两张表
django_apscheduler_djangojob
django_apscheduler_djangojobexecution
django_apscheduler_djangojob表，用于存储定时任务。
django_apscheduler_djangojobexecution表，用于存储每次的执行记录、执行的时长和执行结果等。
status: 执行状态duration: 执行了多久exception: 是否出现异常
4，简单使用定时器
在views.py中编写代码，当django启动时会自动创建定时任务，或者也可以写在urls.py文件中。
from apscheduler.schedulers.background import BackgroundScheduler def scheduler_test(): print('定时任务触发') # BlockingScheduler() 调度器在主线程中执行， 阻塞当前线程。 # scheduler = BlockingScheduler() # BackgroundScheduler() 调度器在后台执行，不会阻塞当前进程，一般情况使用此方法。 scheduler = BackgroundScheduler try: # 添加定时任务，第一个参数为需要定时执行的任务，'cron'定时任务类型，每天0点，30分执行一次，任务id为test。 scheduler.add_job(scheduler_test, 'cron', hour=0, minute=30, id='test', replace_existing=True, timezone='Asia/Shanghai') # 启动定时任务 scheduler.start() except Exception as e: print(e) # 停止定时任务 scheduler.shutdown()
注意：在创建定时任务时，为了确保任务的唯一性和避免重复创建问题，建议指定任务ID。如果没有指定ID，系统会随机生成一个。在多线程环境中，相同任务重复创建的问题可能会更加突出。通过指定ID，当触发add_job时，如果任务已经存在，系统会报错，否则会新创建一个定时任务。这样能够更加高效地管理和执行定时任务。
5，cron定时器参数解释
# 添加定时任务，第一个参数为需要定时执行的任务，'cron'定时任务类型，每天1点，任务id为test。 scheduler.add_job(scheduler_test, 'cron', hour=1, minute=0, id='test', replace_existing=True, timezone='Asia/Shanghai') # 每天9点-17点，每10分钟执行一次。 # scheduler.add_job(create_article_index, 'cron', hour='9-17', minute='0/10', id='test', replace_existing=True, timezone='Asia/Shanghai')
6，interval定时器参数解释
# 添加定时任务，第一个参数为需要定时执行的任务，'interval'定时任务类型，每隔900秒执行一次，任务id为test。 scheduler.add_job(scheduler_test, 'interval', seconds=900, id='test', replace_existing=True, timezone='Asia/Shanghai')
7，date定时器参数解释
# 添加定时任务，第一个参数为需要定时执行的任务，'interval'定时任务类型，2023年7月31日执行一次，任务id为test。 scheduler.add_job(scheduler_test, 'date', run_date=date(2023, 7, 31), id='test', replace_existing=True，timezone='Asia/Shanghai')
8，其他操作
# 删除定时任务 scheduler.remove_job(job_id) # 暂停定时任务 scheduler.pause_job(job_id) # 开启定时任务 scheduler.resume_job(job_id) # 修改定时任务 scheduler.modify_job(job_id) # 获取所有定时任务 scheduler.get_jobs()
第二步：apscheduler定时任务调度库整合到学习任务系统中
1，编写定时调度任务工具
./mysite/study_system/my_apscheduler.py:
# -*- coding: UTF-8 -*- ''' @模块名称: 定时调度任务工具 @注 释：创建 django_apscheduler 定时任务的固定代码 @作 者: PandaCode辉 @weixin公众号: PandaCode辉 @创建时间: 2023-10-10 ''' from apscheduler.schedulers.background import BackgroundScheduler from django_apscheduler.jobstores import DjangoJobStore, register_job from .utils import * # 1.实例化调度器 scheduler = BackgroundScheduler() # 2.调度器使用DjangoJobStore() scheduler.add_jobstore(DjangoJobStore(), "default") ''' 3.设置定时任务， 选择方式为: interval，时间间隔为10s date：希望在某个时间仅运行一次，# 例如在2023-04-14 20:12:00 仅执行一次 interval：要以固定的时间间隔运行作业时使用， 任务隔10分钟执行一次，还可以设置days、hours、seconds参数也可以设置日期范围，start_date-end_date cron：每天固定时间执行任务， 例如每天9点30分10秒; 执行一次, @register_job(scheduler, 'cron', hour='9', minute='30', second='10', id='task_time') ''' try: # 时间间隔可选: seconds 、minutes、hours、days、weeks、start_date、end_date # interval，时间间隔为10分钟 ，每隔10分钟执行一次任务 @register_job(scheduler, "interval", minutes=10, id='my_job_1', replace_existing=True) def my_job_1(): import time # UTC格式当前时区时间 t = time.localtime() now_day_time = time.strftime("%Y-%m-%d %H:%M:%S", t) print('当前日期时间:' + str(now_day_time)) # 这里写你要执行的任务 print(".....开始定时任务.....") # 定时发布学习系统的定时任务，同时邮件通知 study_sys_pub_day_task() print(".....完成定时任务.....") # cron：每天固定时间执行任务， 例如每天8点30分10秒; 执行一次, @register_job(scheduler, "cron", hour='08', minute='30', second='10', id='my_job_2', replace_existing=True) def my_job_2(): import time # UTC格式当前时区时间 t = time.localtime() now_day_time = time.strftime("%Y-%m-%d %H:%M:%S", t) print('当前日期时间:' + str(now_day_time)) # 这里写你要执行的任务 print(".....开始定时任务.....") from . import utils # 发邮件通知，查看最新学习系统的用户信息 study_sys_user_info_notice() print(".....完成定时任务.....") except Exception as e: print(e) # 遇到错误，停止定时器 scheduler.shutdown() # 4.开启定时任务 scheduler.start() ''' 触发器控制的是什么时候会执行任务 1）date 在特定的时间日期执行 from datetime import date# 在2020年11月11日00:00:00执行 sched.add_job(my_job, 'date', run_date=date(2020, 11, 11))# 在2020年11月1日16:30:05 sched.add_job(my_job, 'date', run_date=datetime(2020, 11, 11, 16, 30, 5)) sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05')# 立即执行 sched.add_job(my_job, 'date') sched.start() 2）interval 经过指定的时间间隔执行 from datetime import datetime# 每两小时执行一次 sched.add_job(job_function, 'interval', hours=2)# 在2010年10月10日09:30:00 到2014年6月15日的时间内，每两小时执行一次 sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00') 时间间隔可选: seconds 、minutes、hours、days、weeks、start_date、end_date 、timezone 3）cron 按指定的周期执行 # 在6、7、8、11、12月的第三个周五的00:00, 01:00, 02:00和03:00 执行 sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')# 在2014年5月30日前的周一到周五的5:30执行 sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30') '''
2，编写定时任务实现方法
./mysite/study_system/utils.py:
# -*- coding: utf-8 -*- import re import time from .models import StudyUser, StudyUserAttribute, StudyScheduledTask, StudyTask from datetime import datetime from .mailutil import send_qq_mail def time_diff(srctime, dsttime, timeunit): ''' @方法名称: 时间差值计算 @中文注释: 根据输入的源日期时间,目的日期时间和单位进行计算目标时间比源时间大多少,单位支持秒,分钟,小时和天 @入参: @param srctime str 源时间(YYYYMMDDHHMMSS) @param dstctime str 目标时间(YYYYMMDDHHMMSS) @param timeunit str 时间单位(seconds/minutes/hours/days) @出参: @param timediff int 计算出的差值 @返回状态: @return 0 异常 @return 1 成功 @作 者: PandaCode辉 @weixin公众号: PandaCode辉 @创建时间: 2020-04-28 @使用范例: time_diff("20110521110101", "20110521110909", "seconds"); ''' try: if len(srctime) == 8: srctime = srctime + "000000"; if len(dsttime) == 8: dsttime = dsttime + "000000"; srcdatetime = datetime.strptime(srctime, "%Y%m%d%H%M%S") dstdatetime = datetime.strptime(dsttime, "%Y%m%d%H%M%S") difdatetime = dstdatetime - srcdatetime # 计算时间差值，以秒为单位 dif = difdatetime.days * 24 * 60 * 60 + difdatetime.seconds # print("diff is : " + str(dif)) if timeunit == "seconds": None elif timeunit == "minutes": dif = dif / 60 elif timeunit == "hours": dif = dif / 3600; elif timeunit == "days": dif = difdatetime.days else: return [0, "FBD001", "不支持的时间单位[" + timeunit + "]"] if (dif < 0): dif = -dif return [1, None, None, [dif]] except Exception as e: print("计算日期时间差值异常:" + str(e)) return [0, "999999", "计算日期时间差值异常:" + str(e)] def study_sys_user_info_notice(): ''' @方法名称: 发邮件通知，查看最新学习系统的用户属性信息 @中文注释: 发邮件通知，查看最新学习系统的用户属性信息 @入参: @出参: @返回状态: @return 0 失败,数据库异常 @return 1 成功 @返回错误码 @返回错误信息 @作 者: PandaCode辉 @weixin公众号: PandaCode辉 @创建时间: 2023-08-28 ''' try: # 查询用户信息 # 角色:1: 系统管理员;2: 辅导员;3: 学生;4: 自导自学 role_list = (1, 3, 4) # role__in=role_list 等于字段 role in (1, 3) study_user_list = StudyUser.objects.filter(role__in=role_list) # 查出数据 if len(study_user_list) > 0: for study_user in study_user_list: print(study_user) user_id = str(study_user.user_id) user_name = study_user.username receiver_mail = study_user.email # 用户，关联的属性表的值 studyUserAttribute = StudyUserAttribute.objects.get(user_id=study_user) study_level = str(studyUserAttribute.study_level) intelligence = str(studyUserAttribute.intelligence) memory = str(studyUserAttribute.memory) diligence = str(studyUserAttribute.diligence) physical_fitness = str(studyUserAttribute.physical_fitness) total_points = str(studyUserAttribute.total_points) other_subjects = studyUserAttribute.other_subjects notes = studyUserAttribute.notes update_time = str(studyUserAttribute.update_time)[:19] # 发邮件通知 # 邮件主题 subject = '【学习系统：用户属性】' # 邮件内容 content = '' # 邮件内容换行符，'plain'纯文本格式邮件时候用\n或者\r字符串换行符 # HTML格式，使用<br>换行符 content += '【用户ID ：' + user_id + '】' + '\n' content += '【用户名 ：' + user_name + '】' + '\n' # 级别 if study_level == '1': content += '【级别 ：Lv1-小学】' + '\n' elif study_level == '2': content += '【级别 ：Lv2-初中】' + '\n' elif study_level == '3': content += '【级别 ：Lv3-高中】' + '\n' elif study_level == '4': content += '【级别 ：Lv4-本科】' + '\n' elif study_level == '5': content += '【级别 ：Lv5-硕士】' + '\n' elif study_level == '6': content += '【级别 ：Lv6-博士】' + '\n' elif study_level == '7': content += '【级别 ：Lv7-博士后】' + '\n' content += '【智力 ：' + intelligence + ' 点】' + '\n' content += '【记忆力 ：' + memory + ' 点】' + '\n' content += '【勤奋力 ：' + diligence + ' 点】' + '\n' content += '【体能 ：' + physical_fitness + ' 点】' + '\n' content += '【总积分 ：' + total_points + ' 点】' + '\n' content += '【其他学科属性 ：' + other_subjects + '】' + '\n' content += '【备注提醒 ：' + notes + '】' + '\n' content += '【更新时间 ：' + update_time + '】' + '\n' content += '=================================' # 发送邮件 rstmail = send_qq_mail(subject, content, receiver_mail) if rstmail[0] == 1: return [1, '000000', "发送邮件成功.", [None]] else: return [0, '999999', "发送邮件失败.", [None]] else: return [0, '999999', "查询用户信息数据失败。查无记录.", [None]] except Exception as e: return [0, '999999', "发邮件通知，查看最新学习系统的用户信息异常," + str(e), [None]] def study_sys_pub_day_task(): ''' @方法名称: 定时发布学习系统的定时任务，同时邮件通知 @中文注释: 定时发布学习系统的定时任务，同时邮件通知 @入参: @出参: @返回状态: @return 0 失败,数据库异常 @return 1 成功 @返回错误码 @返回错误信息 @作 者: PandaCode辉 @weixin公众号: PandaCode辉 @创建时间: 2023-08-28 ''' try: # 查询全部待发布定时日常任务列表 studyScheduledTaskList = StudyScheduledTask.objects.all() # 查询是否成功 if len(studyScheduledTaskList) > 0: # 循环发布定时任务 for scheduledTask in studyScheduledTaskList: # 定时任务参数信息 scheduled_id = str(scheduledTask.scheduled_id) schedule_type = scheduledTask.schedule_type task_title = scheduledTask.task_title task_description = scheduledTask.task_description deadline_days = str(scheduledTask.deadline_days) reward_points = str(scheduledTask.reward_points) user_id = scheduledTask.created_by task_type = str(scheduledTask.task_type) exp_daytime = str(scheduledTask.schedule_time) pub_daytime = str(scheduledTask.pub_daytime) phone_num = scheduledTask.phone_num # 今天 # UTC格式当前时区时间 t = time.localtime() now_daytime = str(time.strftime("%Y%m%d%H%M%S", t)) # print('当前日期时间:' + now_daytime) now_day = now_daytime[:8] print('当前日期:' + now_day) # 每天一次，判断当天是否已经发布任务？ if (pub_daytime == now_day): print('当天已经发布任务，不用再发.') continue # 根据循环周期判断 if schedule_type == 'day-1': exp_daytime = now_day + exp_daytime + '00' # print(exp_daytime) exp_daytime = re.sub(r'\D', "", exp_daytime) # print('预定日期时间:' + exp_daytime) # 计算日期差值 dis = time_diff(now_daytime, exp_daytime, "minutes") # print(dis[3][0]) # 计算时间差值 dis_min = int(dis[3][0]) # print(dis_min) # 再判断当前时间是否处于预定时间区间内？ # 比如 08：30，当前时间08：27或08：33查到都算预定时间区间，前后相差10分钟作为判断区间。 if dis_min <= 10: print('处于预定时间区间内，开始发布定时任务.') # 今天 # UTC格式当前时区时间 t = time.localtime() work_date = time.strftime("%Y-%m-%d %H:%M:%S", t) # print('当前日期时间:' + str(work_date)) actual_days = 0 task_status = 0 # 获取当前用户对象 cur_user = user_id # 创建对象并保存到数据库 study_task = StudyTask(task_title=task_title, task_type=task_type, task_description=task_description, reward_points=reward_points, deadline_days=deadline_days, task_status=task_status, actual_days=actual_days, created_by=cur_user, created_time=work_date, update_time=work_date) # 保存到数据库是否成功 study_task.save() # 发布成功后，再更新定时任务表日期字段 pub_daytime # 根据ID查询对象数据 scheduledTask2 = StudyScheduledTask.objects.filter(scheduled_id=scheduled_id) # 更新定时任务表，发布日期 scheduledTask2.update(pub_daytime=now_day) print('更新定时任务表，发布日期，成功') # 邮件通知,接收邮箱 receiver_mail = cur_user.email # 发邮件通知 # 邮件主题 subject = '【学习系统新任务：' + task_title + '】' # 邮件内容 content = '' # 邮件内容换行符，'plain'纯文本格式邮件时候用\n或者\r字符串换行符 # HTML格式，使用<br>换行符 content += '【任务ID ：' + str(study_task.task_id) + '】' + '\n' content += '【发布时间 ：' + work_date + '】' + '\n' content += '【任务名称 ：' + task_title + '】' + '\n' # 任务类型 if task_type == 1: # color : blue content += '【任务类型 ：1-系统任务】' + '\n' elif task_type == 2: content += '【任务类型 ：2-辅导员任务】' + '\n' elif task_type == 4: content += '【任务类型 ：4-自学任务】' + '\n' content += '【任务内容说明 ：' + task_description + '】' + '\n' content += '【计划完成天数 ：' + str(deadline_days) + ' 天】' + '\n' content += '【任务奖励 ：' + str(reward_points) + '点】' + '\n' content += '=================================' # 发送邮件 rstmail = send_qq_mail(subject, content, receiver_mail) print('发送邮件成功') # 再发送短信通知 else: print('不处于预定时间区间内，不用处理.') pass elif (schedule_type == 'timing'): # 定时一次，判断当前时间是否处于预定日期 + 时间区间内？ exp_daytime = re.sub(r'\D', "", exp_daytime) + '00' # print('预定日期时间:' + exp_daytime) # 计算日期差值 dis = time_diff(now_daytime, exp_daytime, "minutes") # print(dis[3][0]) # 计算时间差值 dis_min = int(dis[3][0]) # print(dis_min) # 再判断当前时间是否处于预定时间区间内？ # 比如 08：30，当前时间08：27或08：33查到都算预定时间区间，前后相差10分钟作为判断区间。 if dis_min <= 10: print('处于预定时间区间内，开始发布定时任务.') # 今天 # UTC格式当前时区时间 t = time.localtime() work_date = time.strftime("%Y-%m-%d %H:%M:%S", t) # print('当前日期时间:' + str(work_date)) actual_days = 0 task_status = 0 # 获取当前用户对象 cur_user = user_id # 创建对象并保存到数据库 study_task = StudyTask(task_title=task_title, task_type=task_type, task_description=task_description, reward_points=reward_points, deadline_days=deadline_days, task_status=task_status, actual_days=actual_days, created_by=cur_user, created_time=work_date, update_time=work_date) # 保存到数据库是否成功 study_task.save() receiver_mail = cur_user.email # 发邮件通知 # 邮件主题 subject = '【学习系统新任务：' + task_title + '】' # 邮件内容 content = '' # 邮件内容换行符，'plain'纯文本格式邮件时候用\n或者\r字符串换行符 # HTML格式，使用<br>换行符 content += '【任务ID ：' + str(study_task.task_id) + '】' + '\n' content += '【发布时间 ：' + work_date + '】' + '\n' content += '【任务名称 ：' + task_title + '】' + '\n' # 任务类型 if task_type == 1: # color : blue content += '【任务类型 ：1-系统任务】' + '\n' elif task_type == 2: content += '【任务类型 ：2-辅导员任务】' + '\n' elif task_type == 4: content += '【任务类型 ：4-自学任务】' + '\n' content += '【任务内容说明 ：' + task_description + '】' + '\n' content += '【计划完成天数 ：' + str(deadline_days) + ' 天】' + '\n' content += '【任务奖励 ：' + str(reward_points) + '点】' + '\n' content += '=================================' # 发送邮件 rstmail = send_qq_mail(subject, content, receiver_mail) # 再发送短信通知 # 发布成功后，再删除定时任务表数据，1次性任务 if rstmail[0] == 1: # 发布完就自动删除任务配置。 scheduledTask.delete() print('发布完就自动删除任务配置，完成。') else: print('不处于预定时间区间内，不用处理.') pass else: # 每周几一次，判断当天是否已经发布任务？再判断当前时间是否处于预定时间区间内？ week_int = datetime.now().weekday() + 1 # print('今天周:' + str(week_int)) week_str = 'week-' + str(week_int) # print('今天周:' + week_str) # 先判断当天是否预定周几 if week_str == schedule_type: # 然后再判断当前时间是否处于预定时间区间内？ exp_daytime = now_day + exp_daytime + '00' # print(exp_daytime) exp_daytime = re.sub(r'\D', "", exp_daytime) # print('预定日期时间:' + exp_daytime) # 计算日期差值 dis = time_diff(now_daytime, exp_daytime, "minutes") # print(dis[3][0]) # 计算时间差值 dis_min = int(dis[3][0]) # print(dis_min) # 再判断当前时间是否处于预定时间区间内？ # 比如 08：30，当前时间08：27或08：33查到都算预定时间区间，前后相差10分钟作为判断区间。 if dis_min <= 10: print('处于预定时间区间内，开始发布定时任务.') # 今天 # UTC格式当前时区时间 t = time.localtime() work_date = time.strftime("%Y-%m-%d %H:%M:%S", t) # print('当前日期时间:' + str(work_date)) actual_days = 0 task_status = 0 # 获取当前用户对象 cur_user = user_id # 创建对象并保存到数据库 study_task = StudyTask(task_title=task_title, task_type=task_type, task_description=task_description, reward_points=reward_points, deadline_days=deadline_days, task_status=task_status, actual_days=actual_days, created_by=cur_user, created_time=work_date, update_time=work_date) # 保存到数据库是否成功 study_task.save() # 发布成功后，再更新定时任务表日期字段 pub_daytime # 根据ID查询对象数据 scheduledTask2 = StudyScheduledTask.objects.filter(scheduled_id=scheduled_id) # 更新定时任务表，发布日期 scheduledTask2.update(pub_daytime=now_day) print('更新定时任务表，发布日期，成功') receiver_mail = cur_user.email # 发邮件通知 # 邮件主题 subject = '【学习系统新任务：' + task_title + '】' # 邮件内容 content = '' # 邮件内容换行符，'plain'纯文本格式邮件时候用\n或者\r字符串换行符 # HTML格式，使用<br>换行符 content += '【任务ID ：' + str(study_task.task_id) + '】' + '\n' content += '【发布时间 ：' + work_date + '】' + '\n' content += '【任务名称 ：' + task_title + '】' + '\n' # 任务类型 if task_type == 1: # color : blue content += '【任务类型 ：1-系统任务】' + '\n' elif task_type == 2: content += '【任务类型 ：2-辅导员任务】' + '\n' elif task_type == 4: content += '【任务类型 ：4-自学任务】' + '\n' content += '【任务内容说明 ：' + task_description + '】' + '\n' content += '【计划完成天数 ：' + str(deadline_days) + ' 天】' + '\n' content += '【任务奖励 ：' + str(reward_points) + '点】' + '\n' content += '=================================' # 发送邮件 rstmail = send_qq_mail(subject, content, receiver_mail) print('发送邮件成功') # 再发送短信通知 else: print('不处于预定时间区间内，不用处理.') pass else: print('当天不是预定周几，不用处理.') pass print('========================') # 休眠2秒 time.sleep(2) print('循环发布结束成功.') else: print('查询失败') return [1, '000000', "循环发布结束成功", [None]] except Exception as e: return [0, '999999', "定时发布学习系统的日常任务，同时邮件通知异常," + str(e), [None]]
第三步：运行测试效果
1，定时发邮件通知，查看最新用户属性信息
2，定时发布学习系统的定时学习任务，同时发邮件通知
-------------------------------------------------end -------------------------------------------------