基于空闲时间间隔的Python任务调度代码结果不符问题排查

问题概述 #

我编写了一个Python程序，用于根据每周各天的空闲时间间隔调度任务，输入为空闲时间字典和带时长的任务字典，但运行结果与预期不符。

预期调度结果 #

Task 'Task 1' scheduled from 08:30 AM to 09:50 AM
Task 'Task 2' scheduled from 09:50 AM to 10:10 AM
Task 'Task 3 Part 1' scheduled from 10:10 AM to 10:30 AM
Task 'Task 3 Part 2' scheduled from 01:00 PM to 2:00 PM
Task 'Task 3 Part 3' scheduled from 03:30 PM to 4:10 PM
Task 'Task 3 Part 4' scheduled from 03:30 PM to 4:10 PM
Task 'Task 4' and so on

实际输出 #

Task 'Task 1' scheduled from 09:00 AM to 10:20 AM

程序代码 #

from datetime import datetime, timedelta
import re


def schedule_tasks(tasks, freetime):
    if len(tasks) == 0 or len(freetime) == 0:
        return

    current_day = datetime.now().strftime("%A")
    current_time = datetime.now().strftime("%I:%M %p")

    if current_day not in freetime:
        print("No free time available for the current day.")
        return

    intervals = freetime[current_day]
    intervals = [interval for interval in intervals if interval[1] > current_time]

    if len(intervals) == 0:
        print("No remaining free time for the current day.")
        return

    for task_name, task_time in tasks.items():
        pattern = r'(\d+)h (\d+)m'
        matches = re.match(pattern, task_time)

        try:
            hours = int(matches.group(1))
            minutes = int(matches.group(2))
        except:
            continue

        task_interval = timedelta(hours=hours, minutes=minutes)
        scheduled = False

        for interval in intervals:
            start_time = datetime.strptime(interval[0], "%I:%M %p")
            end_time = datetime.strptime(interval[1], "%I:%M %p")
            time_interval = end_time - start_time

            if task_interval <= time_interval:
                end_datetime = start_time + task_interval
                print(f"Task '{task_name}' scheduled from {start_time.strftime('%I:%M %p')} to {end_datetime.strftime('%I:%M %p')}")
                scheduled = True
                break

        if not scheduled:
            print(f"Not enough free time to schedule '{task_name}'.")

from datetime import datetime, timedelta

def schedule_tasks(tasks, freetime):
    if len(tasks) == 0 or len(freetime) == 0:
        return

    current_day = datetime.now().strftime("%A")
    current_time = datetime.now().strftime("%I:%M %p")

    if current_day not in freetime:
        print("No free time available for the current day.")
        return

    intervals = freetime[current_day]
    intervals = [interval for interval in intervals if interval[1] > current_time]

    if len(intervals) == 0:
        print("No remaining free time for the current day.")
        return

    for task_name, task_time in tasks.items():
        pattern = r'(\d+)h (\d+)m'
        matches = re.match(pattern, task_time)

        try:
            hours = int(matches.group(1))
            minutes = int(matches.group(2))
        except:
            continue

        task_interval = timedelta(hours=hours, minutes=minutes)
        scheduled = False

        for interval in intervals:
            start_time = datetime.strptime(interval[0], "%I:%M %p")
            end_time = datetime.strptime(interval[1], "%I:%M %p")
            time_interval = end_time - start_time

            if task_interval <= time_interval:
                end_datetime = start_time + task_interval
                print(f"Task '{task_name}' scheduled from {start_time.strftime('%I:%M %p')} to {end_datetime.strftime('%I:%M %p')}")
                scheduled = True
                break

        if not scheduled:
            print(f"Not enough free time to schedule '{task_name}'.")


freetime = {
    "Monday": [("08:30 AM", "12:00 PM"), ("02:00 PM", "06:00 PM")],
    "Tuesday": [("09:00 AM", "12:30 PM"), ("03:00 PM", "05:00 PM")],
    "Wednesday": [("08:00 AM", "11:30 AM"), ("01:00 PM", "04:00 PM")]
}

tasks = {
    "Task 1": "1h 20m",
    "Task 2": "20m",
    "Task 3 Part 1": "20m",
    "Task 3 Part 2": "1h",
    "Task 3 Part 3": "40m",
    "Task 3 Part 4": "40m",
    "Task 4": "30m"
}

schedule_tasks(tasks, freetime)

问题根源与修复方案 #

核心问题 #

1. 重复定义函数：代码中两次定义schedule_tasks，后定义的函数会覆盖前一个，属于冗余代码。
2. 未更新空闲时间间隔：每次调度任务后，未将已占用的时间从空闲间隔中移除，导致所有任务都尝试从原始空闲间隔的起始点调度。
3. 任务时长解析不完整：当前正则仅匹配xh ym格式，无法解析20m这类仅含分钟的任务时长，导致此类任务被跳过。
4. 未处理当前时间对间隔的影响：仅过滤结束时间晚于当前时间的间隔，但未调整间隔的起始时间（若当前时间处于间隔中间）。

修复后的代码 #

from datetime import datetime, timedelta
import re


def parse_duration(duration_str):
    """解析任务时长，支持xh ym、xm、xh格式"""
    hours = 0
    minutes = 0
    hour_match = re.search(r'(\d+)h', duration_str)
    minute_match = re.search(r'(\d+)m', duration_str)
    if hour_match:
        hours = int(hour_match.group(1))
    if minute_match:
        minutes = int(minute_match.group(1))
    return timedelta(hours=hours, minutes=minutes)


def schedule_tasks(tasks, freetime):
    if not tasks or not freetime:
        return

    # 测试时可固定日期，比如改为"Monday"
    current_day = datetime.now().strftime("%A")
    current_time_dt = datetime.now()

    if current_day not in freetime:
        print(f"当前日期{current_day}无可用空闲时间。")
        return

    # 处理空闲间隔：过滤并调整起始时间
    available_intervals = []
    for interval in freetime[current_day]:
        start_dt = datetime.strptime(interval[0], "%I:%M %p").replace(year=current_time_dt.year, month=current_time_dt.month, day=current_time_dt.day)
        end_dt = datetime.strptime(interval[1], "%I:%M %p").replace(year=current_time_dt.year, month=current_time_dt.month, day=current_time_dt.day)
        # 若当前时间在间隔内，调整间隔起始为当前时间
        if start_dt < current_time_dt < end_dt:
            available_intervals.append((current_time_dt, end_dt))
        elif end_dt > current_time_dt:
            available_intervals.append((start_dt, end_dt))

    if not available_intervals:
        print("当前日期剩余时间无可用空闲时段。")
        return

    for task_name, duration_str in tasks.items():
        task_duration = parse_duration(duration_str)
        scheduled = False

        # 遍历可用间隔，尝试调度
        for idx, (start_dt, end_dt) in enumerate(available_intervals):
            if (end_dt - start_dt) >= task_duration:
                task_end_dt = start_dt + task_duration
                print(f"Task '{task_name}' scheduled from {start_dt.strftime('%I:%M %p')} to {task_end_dt.strftime('%I:%M %p')}")
                # 更新空闲间隔：移除原间隔，添加剩余时段（如果有）
                del available_intervals[idx]
                if task_end_dt < end_dt:
                    available_intervals.insert(idx, (task_end_dt, end_dt))
                scheduled = True
                break

        if not scheduled:
            print(f"无法为'{task_name}'找到足够的空闲时间。")


freetime = {
    "Monday": [("08:30 AM", "12:00 PM"), ("02:00 PM", "06:00 PM")],
    "Tuesday": [("09:00 AM", "12:30 PM"), ("03:00 PM", "05:00 PM")],
    "Wednesday": [("08:00 AM", "11:30 AM"), ("01:00 PM", "04:00 PM")]
}

tasks = {
    "Task 1": "1h 20m",
    "Task 2": "20m",
    "Task 3 Part 1": "20m",
    "Task 3 Part 2": "1h",
    "Task 3 Part 3": "40m",
    "Task 3 Part 4": "40m",
    "Task 4": "30m"
}

schedule_tasks(tasks, freetime)

修复说明 #

1. 新增时长解析函数：支持多种时长格式，避免任务因解析失败被跳过。
2. 动态更新空闲间隔：调度任务后，移除已占用的时间并插入剩余空闲时段，确保后续任务能使用剩余时间。
3. 处理当前时间的影响：将空闲时间转换为带日期的datetime对象，若当前时间在间隔内，自动调整间隔起始点。
4. 可选固定测试日期：若需测试特定日期的调度结果，可手动设置current_day的值。

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