基于IN/OUT时间的最长工作会话计算逻辑优化求助

需求说明 #

现有数据结构：每个用户对应一个以日期为键的字典，日期对应的值是包含event_time_in（签到时间）和event_time_out（签退时间）的字典列表。需要实现以下功能：

1. 按照规则计算2月内员工的最长工作会话（规则：签退后2小时未签到则会话结束，短于2小时的休息时间计入当前会话）
2. 将结果（user_id、最长会话时长）写入CSV文件
3. 仅使用Python标准库，禁止使用Pandas

数据结构示例 #

from datetime import date, datetime

data = {
    "user1": {
        date(2023, 2, 1): [
            {'event_time_in': datetime(2023, 2, 1, 8, 1, 28)},
            {'event_time_out': datetime(2023, 2, 1, 10, 38, 4)},
            {'event_time_in': datetime(2023, 2, 1, 14, 54, 42)},
            {'event_time_out': datetime(2023, 2, 1, 16, 20, 22)},
            {'event_time_in': datetime(2023, 2, 1, 17, 15, 41)},
            {'event_time_out': datetime(2023, 2, 1, 20, 20, 26)}
        ],
        date(2023, 2, 2): [
            {'event_time_in': datetime(2023, 2, 2, 9, 39, 16)},
            {'event_time_out': datetime(2023, 2, 2, 15, 4, 40)}
        ],
        date(2023, 2, 3): [
            {'event_time_in': datetime(2023, 2, 3, 8, 51, 23)},
            {'event_time_out': datetime(2023, 2, 3, 14, 57, 16)},
            {'event_time_in': datetime(2023, 2, 3, 15, 41, 36)},
            {'event_time_out': datetime(2023, 2, 3, 15, 47, 40)},
            {'event_time_in': datetime(2023, 2, 3, 16, 37, 28)},
            {'event_time_out': datetime(2023, 2, 3, 17, 41, 50)}
        ]
    }
}

会话规则示例 #

• 签到8点，签退12点，签到13点，签退18点 → 合并为一个会话（时长：18-8=10小时）
• 签到8点，签退12点，签到15点，签退18点 → 分为两个会话（时长：12-8=4小时，18-15=3小时）

现有实现问题 #

原代码逻辑存在多处错误：

1. 错误地将event_time_out赋值给current_session_start，混淆了会话起始与结束的概念
2. 间隔判断逻辑错误，应该判断下一次签到时间与上一次签退时间的间隔，而非签退时间与会话起始时间的间隔
3. 未对每天的事件按时间排序，若事件顺序混乱会导致计算完全错误
4. 未正确处理会话的合并逻辑，无法将连续的短间隔休息的会话合并

优化后的核心计算逻辑 #

步骤说明 #

1. 遍历每个用户，仅处理2月的日期数据
2. 对每天的所有事件按时间戳排序，确保签到、签退顺序正确
3. 将排序后的事件配对为（签到时间，签退时间）的元组列表（忽略异常的单独签到/签退）
4. 遍历配对后的时间列表，根据间隔规则合并会话：
   • 初始化第一个会话的起始和结束时间
   • 若下一个签到时间与当前会话结束时间的间隔≤2小时，合并会话（更新会话结束时间为下一个签退时间）
   • 若间隔>2小时，结束当前会话，开始新会话
5. 计算每个会话的时长，记录用户的最长会话时长

完整实现代码 #

from datetime import date, datetime, timedelta
import csv

def process_user_sessions(user_events):
    user_longest_session = timedelta(0)
    
    for day, events in user_events.items():
        # 仅处理2月的数据
        if day.month != 2:
            continue
        
        # 1. 提取所有事件的时间并排序
        event_times = []
        for event in events:
            if 'event_time_in' in event:
                event_times.append(('in', event['event_time_in']))
            elif 'event_time_out' in event:
                event_times.append(('out', event['event_time_out']))
        
        # 按时间排序事件
        event_times.sort(key=lambda x: x[1])
        
        # 2. 配对in和out，生成有效的(in_time, out_time)元组
        paired_times = []
        current_in = None
        for event_type, event_time in event_times:
            if event_type == 'in' and not current_in:
                current_in = event_time
            elif event_type == 'out' and current_in:
                paired_times.append((current_in, event_time))
                current_in = None
        
        # 3. 合并会话
        if not paired_times:
            continue
        
        sessions = []
        # 初始化第一个会话
        session_start, session_end = paired_times[0]
        for in_time, out_time in paired_times[1:]:
            # 计算当前签退与下一次签到的间隔
            gap = in_time - session_end
            if gap <= timedelta(hours=2):
                # 合并会话，更新结束时间
                session_end = out_time
            else:
                # 保存当前会话，开始新会话
                sessions.append(session_end - session_start)
                session_start, session_end = in_time, out_time
        # 保存最后一个会话
        sessions.append(session_end - session_start)
        
        # 更新用户最长会话
        day_longest = max(sessions) if sessions else timedelta(0)
        if day_longest > user_longest_session:
            user_longest_session = day_longest
    
    return user_longest_session

def write_results_to_csv(user_results, filename='employee_longest_sessions.csv'):
    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        fieldnames = ['user_id', 'session_length']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        
        writer.writeheader()
        for user_id, session_length in user_results.items():
            # 将timedelta转换为易读的字符串格式，如"3:45:20"
            hours, remainder = divmod(session_length.total_seconds(), 3600)
            minutes, seconds = divmod(remainder, 60)
            length_str = f"{int(hours)}:{int(minutes):02d}:{int(seconds):02d}"
            writer.writerow({'user_id': user_id, 'session_length': length_str})

# 主逻辑
if __name__ == "__main__":
    user_results = {}
    for user_id, user_events in data.items():
        longest_session = process_user_sessions(user_events)
        user_results[user_id] = longest_session
    
    write_results_to_csv(user_results)
    print("结果已写入employee_longest_sessions.csv")

代码说明 #

• 事件排序：确保所有事件按时间顺序处理，避免因原始数据顺序混乱导致的配对错误
• 配对逻辑：仅保留有效的签到-签退对，忽略单独的签到或签退（可根据需求调整异常处理）
• 会话合并：严格按照「签退后2小时内签到则合并会话」的规则执行，逻辑清晰准确
• CSV写入：将时长转换为时:分:秒的格式，便于阅读

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