基于Pandas实现工时分配：规避周末与日工时上限

核心操作顺序 #

1. 数据预处理：统一日期格式，确保EarliestStart为datetime类型，过滤Hours为非正的无效数据
2. 构建全局可用日期池：基于所有ID的最早起始日期，生成覆盖总工时需求的日期序列，排除周末，标记每日初始可用工时（7小时）
3. 按ID依次分配工时：遍历每个ID，从其EarliestStart开始，逐个占用可用日期的剩余工时，直到该ID总工时分配完毕，实时更新全局日期的剩余可用工时
4. 结果整理：将分配记录转换为目标结构，补充星期数字段

代码实现 #

首先导入依赖库：

import pandas as pd
import numpy as np
from datetime import timedelta

1. 预处理初始数据

先确保日期格式正确并做基础校验：

# 示例初始数据
df_initial = pd.DataFrame({
    'ID': ['A', 'B', 'C'],
    'EarliestStart': pd.to_datetime(['2024-05-20', '2024-05-20', '2024-05-21']),
    'Hours': [15, 8, 5]
})

# 过滤无效数据，保留工时大于0的记录
df_initial = df_initial[df_initial['Hours'] > 0].reset_index(drop=True)

2. 生成全局可用工时表

估算足够覆盖需求的日期范围，生成工作日序列：

# 计算总工时需求，估算所需最大天数（额外加5天缓冲应对周末）
total_required_hours = df_initial['Hours'].sum()
max_days = int(np.ceil(total_required_hours / 7)) + 5
start_date = df_initial['EarliestStart'].min()
end_date = start_date + timedelta(days=max_days)

# 生成工作日序列（排除周末：weekday 0=周一，4=周五）
work_days = pd.date_range(start=start_date, end=end_date, freq='D')
work_days = work_days[work_days.weekday < 5]

# 创建全局可用工时表：记录每日剩余可分配工时
available_hours = pd.DataFrame({
    'Date': work_days,
    'RemainingHours': 7  # 每日最多7小时
})

3. 工时分配逻辑

定义函数逐个处理每个ID的工时分配：

def allocate_single_id(row, available_df):
    remaining = row['Hours']
    earliest = row['EarliestStart']
    id_val = row['ID']
    alloc_records = []
    
    # 筛选当前ID可使用的日期：不早于EarliestStart且还有剩余工时
    valid_dates = available_df[(available_df['Date'] >= earliest) & (available_df['RemainingHours'] > 0)]
    
    for _, date_row in valid_dates.iterrows():
        if remaining <= 0:
            break
        
        # 计算当日可分配的工时：取剩余需求和当日剩余工时的最小值
        assign_hours = min(remaining, date_row['RemainingHours'])
        alloc_records.append({
            'ID': id_val,
            'PlannedStart': date_row['Date'],
            'HoursSplitted': assign_hours
        })
        
        # 更新全局剩余工时
        available_df.loc[available_df['Date'] == date_row['Date'], 'RemainingHours'] -= assign_hours
        remaining -= assign_hours
    
    return pd.DataFrame(alloc_records)

# 按顺序处理每个ID（如需调整优先级，提前对df_initial排序即可）
all_allocations = []
for _, row in df_initial.iterrows():
    alloc_df = allocate_single_id(row, available_hours)
    all_allocations.append(alloc_df)

# 合并所有分配记录
df_result = pd.concat(all_allocations, ignore_index=True)

4. 补充星期数字段并整理结构

# 添加中文星期几（如需英文，去掉locale参数即可）
df_result['Weekday'] = df_result['PlannedStart'].dt.day_name(locale='zh_CN.UTF-8')

# 调整列顺序为目标格式
df_result = df_result[['ID', 'PlannedStart', 'Weekday', 'HoursSplitted']]

关键优化说明 #

• 全局日期池复用：提前生成所有可能的工作日，避免重复生成日期，提升效率
• 实时状态更新：每次分配后立即更新全局剩余工时，确保ID间的日期资源共享逻辑准确
• 灵活优先级：如果需要调整ID的分配优先级，只需在处理前对df_initial按优先级字段排序即可
• 性能兼容：对于小规模数据，遍历行的方式足够简洁；如果数据量极大，可以考虑将同批次日期的ID批量处理，减少循环次数

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