Python实现从STDIN读取事件并翻转重叠事件overlap标志
处理百万级重叠事件的高效解决方案
嘿,恭喜你发出第一篇Stack Overflow提问!针对你这个百万级事件处理的作业难题,我来给你捋捋思路和具体实现——毕竟数据量这么大,效率肯定是第一位的对吧?
先明确前提
你提到事件格式是{ 'start_time':... },这里默认每个事件还包含end_time(不然没法判断重叠),且每个事件初始带有overlap布尔标志,我就基于这个前提来展开方案啦。
核心思路:排序+一次遍历(O(n log n)复杂度)
百万级数据绝对不能用暴力比对(O(n²)会直接超时),排序是这里的关键,能把重叠判断简化成线性遍历。
具体步骤
- 高效读取输入:百万行数据用逐行读取会很慢,得用对应语言的快速IO工具(比如Python里用
sys.stdin.read()一次性读入再拆分)。同时要给每个事件保留原始索引,因为排序后要还原回输入顺序输出。 - 排序事件:把所有事件按
start_time升序排序;如果start_time相同,按end_time升序排序。排序的时间复杂度是O(n log n),完全能处理百万级数据。 - 标记重叠事件:
- 初始化一个长度为N的数组
is_overlapping,默认全为False。 - 遍历排序后的事件,记录当前已遍历事件中最晚的
end_time和对应的原始索引。 - 对于当前事件:
- 如果当前事件的
start_time < 最晚end_time:说明当前事件和之前的事件重叠,把is_overlapping[当前事件原始索引]和is_overlapping[最晚end_time对应的索引]都设为True。 - 更新最晚
end_time:如果当前事件的end_time比记录的最晚值大,就更新最晚end_time和对应索引。
- 如果当前事件的
- 连续重叠的情况(比如A和B重叠、B和C重叠)会自动被覆盖,因为B被标记后,C和B重叠时也会被标记,A的标记也已经存在。
- 初始化一个长度为N的数组
- 翻转标志并输出:按原始顺序遍历事件,若
is_overlapping[索引]为True,就翻转该事件的overlap标志,最后按原始顺序输出所有事件。
Python代码示例(优化IO和内存)
import sys import ast def main(): # 一次性读取所有输入,减少IO开销 lines = sys.stdin.read().splitlines() n = int(lines[0]) events = [] for idx in range(n): # 解析Python字典格式的事件行 event = ast.literal_eval(lines[idx+1]) # 保留原始索引,用于后续还原顺序 event['original_idx'] = idx events.append(event) # 按start_time排序,start_time相同则按end_time排序 events.sort(key=lambda x: (x['start_time'], x['end_time'])) is_overlapping = [False] * n if n == 0: return # 初始化最晚结束时间和对应索引 last_end = events[0]['end_time'] last_original_idx = events[0]['original_idx'] for i in range(1, n): current = events[i] current_start = current['start_time'] current_end = current['end_time'] current_idx = current['original_idx'] if current_start < last_end: # 标记当前事件和之前的事件为重叠 is_overlapping[current_idx] = True is_overlapping[last_original_idx] = True # 更新最晚结束时间 if current_end > last_end: last_end = current_end last_original_idx = current_idx # 按原始顺序整理事件,翻转overlap标志 result = [None] * n for event in events: idx = event['original_idx'] if is_overlapping[idx]: event['overlap'] = not event.get('overlap', False) # 移除临时索引,恢复输出格式 del event['original_idx'] result[idx] = event # 批量输出,减少IO次数 output = '\n'.join(str(e) for e in result) print(output) if __name__ == "__main__": main()
注意事项
- 输入格式适配:如果你的事件是JSON格式,把
ast.literal_eval换成json.loads即可。 - 边界调整:如果作业里把首尾相连(start_time等于另一个事件的end_time)算重叠,把判断条件改成
current_start <= last_end就行。 - 内存占用:百万个事件在Python里大概占几百MB内存,普通机器完全能承受,无需分块处理。
内容的提问来源于stack exchange,提问作者benny
相关产品推荐
相关产品推荐

