You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

不使用Pandas实现CSV中重叠时间段的合并操作

合并CSV中的重叠时间段(仅用Python核心库)

我有一份包含人员时间段数据的CSV文件,样本如下:

NameStartEnd
John12:0013:00
John12:1013:00
John12:2013:20
Tom12:0013:10
John13:5014:00
Jerry14:0014:30
Alice15:0016:00
Jerry11:0015:00

需要按人员合并重叠或连续的时间段,合并前后的效果对比:

合并前

  • John:[12:00,13:00],[12:10,13:00],[12:20,13:20],[13:50,14:00]
  • Jerry:[14:00,14:30],[11:00,15:00]
  • Tom:[12:00,13:10]
  • Alice:[15:00,16:00]

合并后

  • John:[12:00,13:20],[13:50,14:00]
  • Jerry:[11:00,15:00]
  • Tom:[12:00,13:10]
  • Alice:[15:00,16:00]

用Pandas能轻松实现,但要求只能用Python核心库。我尝试用csv模块写了部分代码,但不知道怎么判断时间段重叠并合并:

import csv

dict = {}
with open('person.csv', mode='r') as csv_file:
    csv_reader = csv.DictReader(csv_file)
    for row in csv_reader:
        name = row["Name"]
        start_time = row["Start"]
        end_time = row["End"]
        dict[name] = [start_time,end_time]

完整解决方案代码

你的代码问题在于每次覆盖了用户的时间段,没有收集所有条目,也没处理重叠逻辑。下面是完整的实现:

import csv

def merge_time_intervals(intervals):
    # 先把时间段按开始时间排序,HH:MM格式的时间字符串可直接比较
    sorted_intervals = sorted(intervals, key=lambda x: x[0])
    if not sorted_intervals:
        return []
    
    merged = [sorted_intervals[0]]
    for current_start, current_end in sorted_intervals[1:]:
        last_start, last_end = merged[-1]
        # 判断当前时间段是否和最后一个合并后的时间段重叠或连续
        if current_start <= last_end:
            # 合并:更新结束时间为两者的最大值
            new_end = max(last_end, current_end)
            merged[-1] = (last_start, new_end)
        else:
            # 不重叠,直接添加新时间段
            merged.append((current_start, current_end))
    return merged

# 读取CSV并收集每个用户的所有时间段
user_intervals = {}
with open('person.csv', mode='r') as csv_file:
    csv_reader = csv.DictReader(csv_file)
    for row in csv_reader:
        name = row["Name"].strip()
        start = row["Start"].strip()
        end = row["End"].strip()
        # 用列表存储每个用户的所有时间段,避免覆盖
        if name not in user_intervals:
            user_intervals[name] = []
        user_intervals[name].append((start, end))

# 对每个用户的时间段执行合并
merged_user_intervals = {}
for name, intervals in user_intervals.items():
    merged_user_intervals[name] = merge_time_intervals(intervals)

# 打印合并结果
print("合并后的结果:")
for name, intervals in merged_user_intervals.items():
    formatted_intervals = ",".join([f"[{s},{e}]" for s,e in intervals])
    print(f"- {name}:{formatted_intervals}")

# 可选:将合并结果写入新的CSV文件
with open('merged_person.csv', mode='w', newline='') as csv_file:
    fieldnames = ["Name", "Start", "End"]
    writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
    writer.writeheader()
    for name, intervals in merged_user_intervals.items():
        for start, end in intervals:
            writer.writerow({"Name": name, "Start": start, "End": end})

代码说明

  1. merge_time_intervals函数:专门处理单个用户的时间段合并
    • 先对时间段按开始时间排序,保证处理顺序正确
    • 遍历每个时间段,和已合并的最后一条对比:重叠/连续就更新结束时间,不重叠就新增条目
  2. CSV读取逻辑:用字典存储每个用户的所有时间段,避免之前的覆盖问题
  3. 结果输出:既可以打印到控制台,也可以写入新的CSV文件保存

内容的提问来源于stack exchange,提问作者Bk thomas

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.22 18:45:33