基于频率、工作日与调度规则计算报告下一次运行时间的技术问题

前置映射配置 #

先把配置表的规则转换为程序可直接调用的映射关系，避免每次查询数据库：

• 频率映射：

  FREQ_RULE = {
      1: {"type": "day", "step": 3}, # 3天周期
      2: {"type": "day", "step": 7}, # 周周期
      3: {"type": "day", "step": 14}, # 双周周期
      4: {"type": "month", "step": 1}, # 月周期
      5: {"type": "month", "step": 3} # 季度周期
  }
  

• 工作日映射（适配ISO星期标准，周一为1，周五为5）：

  WEEKDAY_MAP = {
      1: 1, # 周一
      2: 2, # 周二
      3: 3, # 周三
      4: 4, # 周四
      5: 5  # 周五
  }
  

核心计算逻辑 #

from datetime import date
from dateutil.relativedelta import relativedelta # 处理月增量的边界问题

def calc_next_run_time(schedule: dict) -> date:
    # schedule入参包含字段：start_date(日期类型)、freq_id(频率ID)、weekday_id(工作日ID)
    freq_rule = FREQ_RULE[schedule["freq_id"]]
    target_weekday = WEEKDAY_MAP[schedule["weekday_id"]]
    
    # 确定初始基准日期
    current_date = date.today()
    base_date = schedule["start_date"]
    if current_date > base_date:
        base_date = current_date - relativedelta(days=1) # 避免漏算当天的触发机会
    
    # 循环生成候选日期直到找到符合要求的
    candidate_date = base_date
    while True:
        # 按频率规则生成下一个候选日期
        if freq_rule["type"] == "day":
            candidate_date += relativedelta(days=freq_rule["step"])
        else:
            candidate_date += relativedelta(months=freq_rule["step"])
        
        # 校验星期是否匹配
        if candidate_date.isoweekday() == target_weekday:
            return candidate_date

边界场景处理说明 #

• 月/季度增量自动处理月末边界：示例中用relativedelta处理月增量的月末兼容问题，若不想引入第三方日期库，可自行实现：当月度增量后日期的月份和预期增量月份不符时，自动回退到上月最后一天即可，不会出现跨月异常
• 若按频率增量生成的候选日期不匹配工作日，会自动继续累加下一个周期，直到找到匹配的工作日
• 若初始的StartDate本身符合频率和工作日要求、且还未到StartDate的时间，会直接返回StartDate作为首次运行时间

内容的提问来源于stack exchange，提问作者John