Python实现会议室最小数量分配算法部分测试用例报错排查

问题描述 #

给定一组会议时间块，需要将其分配到会议室中，同时保证使用的会议室数量最少。现有实现算法在大部分场景运行正常，但偶发返回错误结果，无法定位问题原因。

现有问题代码 #

from typing import List, Tuple
import heapq
from collections import defaultdict

def offer(begin:int, end:int, heap:List[int]) -> int:
    while heap and heap[0] <= begin:
        heapq.heappop(heap)
    heapq.heappush(heap, end)
    return len(heap)


def schedule(times:List[Tuple[int, int]]):
    heap = []
    groups = defaultdict(list)
    for begin, end in sorted(times):
        key = offer(begin, end, heap)
        groups[key].append( (begin, end))
    
    return groups

注：原代码遗漏from collections import defaultdict导入，此处补充以保证可运行

已验证通过的测试用例 #

1. 基础用例 #

times = [(12, 13), (12, 15), (17, 20)]
# 运行结果符合预期：1号会议室安排(12,13)、(17,20)，2号会议室安排(12,15)

2. 复杂场景用例 #

times = [(12,13), (13,15), (17,20), (13,14), (19 , 21), (18, 20), (12,13)] 
# 运行结果符合预期：共使用3间会议室，无时间重叠的会议分配

失败测试用例 #

该场景逻辑简单但执行失败：

times =  [(12, 16), (15,18), (17,20)]
# 预期结果：
# {
#    1: [(12, 16), (17,20)],
#    2: [(15,18)],
# }

实际返回结果存在错误，2号会议室被分配了两个时间重叠的会议：
defaultdict(list, {1: [(12, 16)], 2: [(15, 18), (17, 20)]})

根因分析 #

现有代码的核心错误是：将堆的当前长度直接作为会议室分组编号，这个值仅能表示当前所需的最少会议室总数，无法对应到实际空闲的会议室ID。
以失败用例的执行过程为例：

1. 第一个会议(12,16)进入：堆为空，推入16，返回长度1，分组1添加该会议，逻辑正常
2. 第二个会议(15,18)进入：堆顶16>15，推入18，返回长度2，分组2添加该会议，逻辑正常
3. 第三个会议(17,20)进入：堆顶16<=17，弹出16（对应分组1的会议室已经空闲），推入20后堆长度为2，直接返回2作为分组ID，导致会议被错分到分组2，覆盖了本该分到分组1的逻辑。

修复方案 #

堆中不再仅存储结束时间，改为存储(结束时间, 会议室ID)的元组，同时维护可用会议室ID池，回收已结束会议的会议室ID优先分配给新会议：

from typing import List, Tuple
import heapq
from collections import defaultdict

def schedule(times: List[Tuple[int, int]]):
    heap = []
    groups = defaultdict(list)
    available_ids = []
    max_id = 0
    for begin, end in sorted(times):
        # 回收所有已结束的会议室ID
        while heap and heap[0][0] <= begin:
            _, free_id = heapq.heappop(heap)
            heapq.heappush(available_ids, free_id)
        # 取可用ID，没有则生成新ID
        if available_ids:
            use_id = heapq.heappop(available_ids)
        else:
            max_id += 1
            use_id = max_id
        groups[use_id].append((begin, end))
        heapq.heappush(heap, (end, use_id))
    return groups

修复后运行失败测试用例可得到预期结果。

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