生成复杂排班表时Kernel崩溃问题排查求助

核心崩溃原因 #

1. 排列生成导致内存爆炸
   代码中使用list(itertools.permutations(available_people))生成所有可用人员的排列组合，当可用人数超过10人时，排列数会呈阶乘级增长（比如15人就有1.3e12个排列），直接耗尽服务器内存。更致命的是，你要找main_presentation not in order的排列，但permutations生成的是包含所有可用人员的全排列，所以这个条件永远不成立，循环会遍历所有排列直到内存崩溃。

2. 日期可用性检查逻辑错误
   你的people_availability字典中同时存在单个日期和日期区间（比如(datetime(2023,9,8), datetime(2023,9,20))），但代码中date not in unavailable_dates只会检查单个日期是否在列表中，完全忽略了区间，导致可用性判断完全错误，进一步引发后续逻辑混乱。

3. 批量处理未解决本质问题
   批量拆分只是把大任务拆成小任务，但每个小任务里的日期仍在生成巨量排列，所以无法解决内存溢出问题。

修复方案与优化代码 #

关键修改点 #

• 移除阶乘级排列生成，改用随机抽样替代，直接从排除主 presenter 的可用人员中选highlight和午餐饮品提供者。
• 重构日期可用性检查函数，支持单个日期和区间的判断，并预先将不可用日期处理为快速查询的集合/区间列表。
• 简化任务分配逻辑，避免不必要的遍历。

修改后的完整代码 #

#!/usr/bin/env python
import csv
import random
from datetime import datetime, timedelta

def is_date_unavailable(check_date, unavailable_entries):
    """检查日期是否在不可用列表（支持单个日期或区间）"""
    for entry in unavailable_entries:
        if isinstance(entry, tuple) and len(entry) == 2:
            start, end = entry
            if start <= check_date <= end:
                return True
        elif isinstance(entry, datetime):
            if check_date == entry:
                return True
    return False

def generate_weekly_schedule(people_availability, start_date, end_date):
    schedule = []
    current_date = start_date
    while current_date <= end_date:
        # 筛选当日可用人员
        available_people = [
            person for person, unavailable in people_availability.items()
            if not is_date_unavailable(current_date, unavailable)
        ]
        if not available_people:
            current_date += timedelta(days=1)
            continue

        # 选主 presenter：要求未来7天都可用
        main_presenter = None
        for person in available_people:
            next_week_dates = [current_date + timedelta(days=i) for i in range(7)]
            if all(not is_date_unavailable(d, people_availability[person]) for d in next_week_dates):
                main_presenter = person
                break
        if not main_presenter:
            current_date += timedelta(days=1)
            continue

        # 从可用人员中排除主 presenter，用于分配其他任务
        non_main_people = [p for p in available_people if p != main_presenter]
        if len(non_main_people) < 3:  # 至少需要2个highlight+1个午餐+1个咖啡的候选
            current_date += timedelta(days=1)
            continue

        # 随机打乱后分配任务
        random.shuffle(non_main_people)
        highlight1, highlight2 = non_main_people[:2]
        lunch_candidates = non_main_people[2:]

        # 选午餐和咖啡提供者：要求下周同一天可用
        lunch_provider = None
        coffee_provider = None
        target_date = current_date + timedelta(days=7)
        for person in lunch_candidates:
            if not is_date_unavailable(target_date, people_availability[person]):
                if not lunch_provider:
                    lunch_provider = person
                elif not coffee_provider:
                    coffee_provider = person
                if lunch_provider and coffee_provider:
                    break

        # 补充：如果找不到符合条件的，退而求其次选任意可用人员
        if not lunch_provider:
            lunch_provider = lunch_candidates[0]
        if not coffee_provider:
            coffee_provider = lunch_candidates[1] if len(lunch_candidates)>=2 else lunch_candidates[0]

        schedule.append([
            current_date.strftime("%Y-%m-%d"),
            main_presenter,
            highlight1,
            highlight2,
            lunch_provider,
            coffee_provider
        ])
        current_date += timedelta(days=1)
    return schedule

def write_schedule_to_csv(schedule, csv_filename):
    with open(csv_filename, mode='w', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(["Date", "Main Presentation", "Highlight 1", "Highlight 2", "Lunch Provider", "Coffee Provider"])
        writer.writerows(schedule)

def batch_process_and_merge(people_availability, start_date, end_date, batch_size):
    batched_schedule = []
    current_date = start_date
    while current_date <= end_date:
        batch_end = min(current_date + timedelta(days=batch_size-1), end_date)
        batch_sched = generate_weekly_schedule(people_availability, current_date, batch_end)
        batched_schedule.extend(batch_sched)
        current_date = batch_end + timedelta(days=1)
    return batched_schedule

if __name__ == "__main__":
    people_availability = {
        "Person 1": [(datetime(2023, 9, 8), datetime(2023, 9, 20))],
        "Person 2": [(datetime(2023, 9, 8), datetime(2023, 10, 13))],
        "Person 3": [],
        "Person 4": [],
        "Person 5":[(datetime(2023, 10, 1), datetime(2023, 12, 15))],
        "Person 6": [],
        "Person 7": [],
        "Person 8": [datetime(2023, 11, 10), datetime(2023, 11, 17)],
        "Person 9": [],
        "Person 10":[],
        "Person 11": [],
        "Person 12": [(datetime(2023, 10, 27), datetime(2023, 12, 15))],
        "Person 13": [datetime(2023, 9, 15)],
        "Person 14": [(datetime(2023, 9, 8), datetime(2023, 12, 15))],
        "Person 15": [(datetime(2023, 9, 8), datetime(2023, 12, 15))],
        "Person 16": [(datetime(2023, 9, 8), datetime(2023, 12, 15))],
        "Person 17": [],
        "Person 18": [],
        "Person 19": [datetime(2023, 9, 22), datetime(2023, 11, 3)],
        "Person 20": [],
        "Person 21": [],
        "Person 22": [],
        "Person 23": [datetime(2023, 9, 8), datetime(2023, 9, 13)],
        "Person 24": [datetime(2023, 9, 8), datetime(2023, 9, 30)],               
    }

    start_date = datetime(2023, 9, 8)
    end_date = datetime(2023, 12, 15)
    batch_size = 7 

    batched_schedule = batch_process_and_merge(people_availability, start_date, end_date, batch_size)
    write_schedule_to_csv(batched_schedule, "weekly_schedule.csv")

额外优化建议 #

1. 预先将所有不可用日期转换为集合：如果人员的不可用区间不多，可以提前把区间展开成单个日期存入集合，让可用性检查从O(n)变成O(1)，进一步提升效率。
2. 添加日志输出：在关键步骤打印当前处理日期和分配结果，方便调试和监控进度。
3. 增加任务冲突检查：可以记录每个人最近的任务，确保同一个人不会连续承担相同任务（比如连续两周带午餐），符合你"尽量错开任务"的需求。

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