Python中如何按user_id分组而非排序实现活动数据转用户会话

核心实现逻辑 #

• 排序优化：先按user_id + first_seen_at升序排序，保证同一个用户的活动按发生时间先后排列，避免会话切割错误
• 分组存储：用字典结构存储最终的user_sessions，键为用户ID，值为对应用户的所有会话列表
• 会话切割规则：遍历每个活动时，若当前用户不存在则新建用户和初始会话；若用户存在，取用户最新的会话，比对当前活动的first_seen_at与最新会话的ended_at的时间差，超过300秒（5分钟）则新建会话，否则合并到当前会话
• 字段计算规则：
  • started_at：会话第一个活动的first_seen_at
  • ended_at：会话最后一个活动的answered_at
  • activity_ids：同一会话所有活动的ID集合
  • duration_seconds：ended_at减去started_at的总秒数（保留小数）

完整实现代码 #

import json
import datetime

# 读取JSON数据
with open('/Users/kenyacastellanos/Downloads/data.json') as json_data_file:
    data = json.load(json_data_file)

# 按用户ID+活动首次出现时间升序排序，保证同用户活动按时间先后排列
data['activities'].sort(key = lambda x: (x['user_id'], x['first_seen_at']))

# 初始化最终输出的用户会话结构
user_sessions = {}

# 遍历所有活动
for activity in data['activities']:
    uid = activity['user_id']
    # 转换时间格式
    curr_first = datetime.datetime.fromisoformat(activity['first_seen_at'])
    curr_answer = datetime.datetime.fromisoformat(activity['answered_at'])
    act_id = activity['id']

    # 该用户首次出现，新建用户和第一个会话
    if uid not in user_sessions:
        user_sessions[uid] = [
            {
                "started_at": activity['first_seen_at'],
                "ended_at": activity['answered_at'],
                "activity_ids": [act_id],
                "duration_seconds": (curr_answer - curr_first).total_seconds()
            }
        ]
        continue
    
    # 取该用户最新的会话
    latest_session = user_sessions[uid][-1]
    latest_end = datetime.datetime.fromisoformat(latest_session['ended_at'])
    # 计算当前活动和上一个会话的间隔
    gap = (curr_first - latest_end).total_seconds()

    # 间隔超过5分钟，新建会话
    if gap > 300:
        new_session = {
            "started_at": activity['first_seen_at'],
            "ended_at": activity['answered_at'],
            "activity_ids": [act_id],
            "duration_seconds": (curr_answer - curr_first).total_seconds()
        }
        user_sessions[uid].append(new_session)
    else:
        # 合并到当前会话，更新会话的结束时间和时长、活动ID列表
        latest_session['activity_ids'].append(act_id)
        latest_session['ended_at'] = activity['answered_at']
        new_duration = (curr_answer - datetime.datetime.fromisoformat(latest_session['started_at'])).total_seconds()
        latest_session['duration_seconds'] = new_duration

# 输出结果
final_result = {"user_sessions": user_sessions}
print(json.dumps(final_result, indent=2, ensure_ascii=False))

# 如需保存到文件可取消以下注释
# with open('user_sessions_output.json', 'w', encoding='utf-8') as f:
#     json.dump(final_result, f, indent=2, ensure_ascii=False)

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