功能说明
这段代码解析标准五段式 Cron 表达式（分、时、日、月、周），从给定时间开始计算接下来若干次运行时间。它适合在定时任务管理页面展示“接下来三次执行”、在命令行工具中校验表达式，或为调度器编写轻量预览。
实现支持 *、具体数字、范围 1-5、列表 1,15,30 和步长 */15，并遵循 Cron 的日期语义：当“日”和“周”都被限制时，两者是“或”的关系；只限制其中一个时，另一个视为通配。
关键代码
parse_field 把每个字段展开成整数集合，检查上下界和步长，并在星期字段中把 7 映射为 0（星期日）。next_runs 从下一分钟开始逐分钟扫描，分别检查分、时、月，再按上述规则计算 day_matches。
为了避免无限循环，代码设置了两年的扫描上限；找不到足够结果时抛出 ValueError。main 从 2026-10-21 08:50 开始，计算工作日每 15 分钟的运行时间，以及下个月一号的零点。
使用注意
逐分钟扫描对稀疏表达式效率不高，例如“每年 1 月 1 日”需要走完大量分钟；生产环境应跳到下一个可能匹配的时间。当前不支持秒字段、@daily 等宏、月份/星期英文缩写、时区和夏令时。Cron 的“日/周或语义”经常被误解，修改逻辑前应补充测试。
不存在的日期（如 2 月 30 日）不会匹配，最终可能触发“两年内无结果”的错误。起始时间只保留到分钟，秒和微秒会被清零后再加一分钟，因此结果是严格晚于 start 的下一次运行时间。
验证方式
示例输出 2026-10-21T09:00、09:15、09:30，说明步长和工作日限制生效；0 0 1 * * 从 10 月 21 日计算得到 2026-11-01T00:00，并通过断言。可以再加入 0 0 29 2 * 验证闰年边界，或加入 7 验证星期日别名。
代码示例
from __future__ import annotations
from datetime import datetime, timedelta
def parse_field(
text: str,
low: int,
high: int,
*,
sunday_alias: bool = False,
) -> set[int]:
values: set[int] = set()
for part in text.split(","):
step = 1
if "/" in part:
part, raw_step = part.split("/", 1)
step = int(raw_step)
if part == "*":
start, end = low, high
elif "-" in part:
raw_start, raw_end = part.split("-", 1)
start, end = int(raw_start), int(raw_end)
else:
start = end = int(part)
if not (low <= start <= end <= high) or step <= 0:
raise ValueError(f"invalid cron field: {text}")
values.update(range(start, end + 1, step))
if sunday_alias:
values = {0 if value == 7 else value for value in values}
return values
def next_runs(expression: str, start: datetime, count: int = 3) -> list[datetime]:
fields = expression.split()
if len(fields) != 5:
raise ValueError("expected five cron fields")
minutes = parse_field(fields[0], 0, 59)
hours = parse_field(fields[1], 0, 23)
days = parse_field(fields[2], 1, 31)
months = parse_field(fields[3], 1, 12)
weekdays = parse_field(fields[4], 0, 7, sunday_alias=True)
day_restricted = fields[2] != "*"
weekday_restricted = fields[4] != "*"
current = start.replace(second=0, microsecond=0) + timedelta(minutes=1)
deadline = current + timedelta(days=366 * 2)
results: list[datetime] = []
while len(results) < count and current <= deadline:
cron_weekday = (current.weekday() + 1) % 7
if day_restricted and weekday_restricted:
day_matches = current.day in days or cron_weekday in weekdays
else:
day_matches = current.day in days and cron_weekday in weekdays
if (
current.minute in minutes
and current.hour in hours
and current.month in months
and day_matches
):
results.append(current)
current += timedelta(minutes=1)
if len(results) < count:
raise ValueError("no matching run time within two years")
return results
def main() -> None:
start = datetime(2026, 10, 21, 8, 50)
for run in next_runs("*/15 9-17 * * 1-5", start):
print(run.isoformat(timespec="minutes"))
first_of_month = next_runs("0 0 1 * *", start, 1)[0]
print(first_of_month.isoformat(timespec="minutes"))
assert first_of_month == datetime(2026, 11, 1, 0, 0)
if __name__ == "__main__":
main()