不使用Topological Sort，用Python标准库计算项目最短工期

问题背景 #

我有一个名为activities.py的Python列表，包含1001个项目活动，每个活动字段说明：

• Cost: 活动成本
• Duration: 活动工期
• Name: 活动名称
• Predecessors: 前置活动列表（None表示无前置）

示例数据如下：

activities = [
    {'Cost': 10000, 'Duration': 5, 'Name': 'Activity 1', 'Predecessors': None},
    {'Cost': 8000, 'Duration': 3, 'Name': 'Activity 2', 'Predecessors': ['Activity 1']},
    {'Cost': 2000, 'Duration': 15, 'Name': 'Activity 3', 'Predecessors': ['Activity 2', 'Activity 1']},
    {'Cost': 7000, 'Duration': 16, 'Name': 'Activity 4', 'Predecessors': ['Activity 2']},
    {'Cost': 5000, 'Duration': 20, 'Name': 'Activity 5', 'Predecessors': ['Activity 4']},
    # ... 剩余996个活动
]

之前我用以下代码计算总工期，但这是所有活动串行的总时长，现在需要计算考虑活动并行的项目最短工期：

def get_total_project_duration(activities: list) -> int:
    total_time = sum(item['Duration'] for item in activities)
    return total_time

已知可以通过拓扑排序实现，要求仅使用Python标准库完成，此前尝试图结构因数据量过大失败。

解决方案代码 #

from collections import deque

def get_min_project_duration(activities: list) -> int:
    # 构建活动名称到活动的映射，快速查找
    activity_map = {act['Name']: act for act in activities}
    # 构建反向映射：前置活动 -> 后续活动列表
    successor_map = {act['Name']: [] for act in activities}
    # 记录每个活动的入度（未完成的前置活动数量）
    in_degree = {}
    # 记录每个活动的最早完成时间
    earliest_finish = {}

    # 初始化数据结构
    for act in activities:
        act_name = act['Name']
        predecessors = act['Predecessors']
        # 处理无前置的情况
        if predecessors is None or not predecessors:
            in_degree[act_name] = 0
            earliest_finish[act_name] = act['Duration']
        else:
            in_degree[act_name] = len(predecessors)
            earliest_finish[act_name] = 0  # 初始化为0，后续更新
            # 给每个前置活动添加当前活动为后续
            for pred_name in predecessors:
                successor_map[pred_name].append(act_name)

    # 初始化队列，放入所有入度为0的活动
    q = deque([name for name, degree in in_degree.items() if degree == 0])

    max_duration = 0

    while q:
        current_name = q.popleft()
        current_ef = earliest_finish[current_name]
        # 更新最大工期
        if current_ef > max_duration:
            max_duration = current_ef
        # 遍历当前活动的所有后续活动
        for succ_name in successor_map[current_name]:
            succ_act = activity_map[succ_name]
            # 计算后续活动的最早开始时间：取所有前置活动的最大EF
            new_es = current_ef
            if new_es > earliest_finish[succ_name] - succ_act['Duration']:
                earliest_finish[succ_name] = new_es + succ_act['Duration']
            # 入度减1
            in_degree[succ_name] -= 1
            # 入度为0时加入队列
            if in_degree[succ_name] == 0:
                q.append(succ_name)

    return max_duration

代码说明 #

1. 数据结构优化：用activity_map快速通过活动名称查找活动，避免遍历整个列表；successor_map记录每个活动的后续活动，减少重复查找。
2. 高效队列：使用collections.deque实现队列，popleft()操作时间复杂度为O(1)，比列表的pop(0)（O(n)）效率高，适合处理1000+数据量。
3. 拓扑排序逻辑：
   • 初始化时处理无前置活动的节点，设置其最早完成时间为自身工期。
   • 遍历过程中，每个活动出队后更新其后续活动的最早完成时间（确保后续活动在所有前置完成后才开始）。
   • 后续活动入度减至0时加入队列，保证所有前置依赖都已处理。
4. 结果计算：遍历过程中记录最大的最早完成时间，即为项目的最短工期（关键路径长度）。

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