def generate_stats()

in identity-resolution/notebooks/identity-graph/nepytune/usecase/purchase_path.py [0:0]

• 32 lines of code
• 8 McCabe index (conditional complexity)

def generate_stats(all_activities, **kwargs):
    """Generate statistics per user (persistentId) activities."""
    result = dict()

    user_sessions = generate_session_from_event(all_activities, **kwargs)

    def grouper(session):
        return session.persistentId or session.transientId

    for persistent_id, session_list in (itertools.groupby(sorted(user_sessions, key=grouper), key=grouper)):
        session_list = list(session_list)
        session_durations = get_time_by_device(session_list)
        user_events_by_timestamp = sorted(
            itertools.chain.from_iterable([session.events for session in session_list]),
            key=lambda event: event.ts
        )

        if persistent_id not in result:
            result[persistent_id] = {
                "transient_ids": {},
                "devices_count": 0,
                "first_device": first_device_in_session(user_events_by_timestamp),
                "time_to_purchase": time_to_purchase(user_events_by_timestamp),
            }

        for transient_id, duration in session_durations.items():
            user_sessions = sorted(
                [session for session in session_list if session.transientId == transient_id],
                key=lambda session: session.events[0].ts
            )
            result[persistent_id]["transient_ids"][transient_id] = {
                "sessions_duration": duration,
                "sessions_count": len(user_sessions),
                "purchase_session": user_sessions[-1],
                "sessions": user_sessions
            }
            result[persistent_id]["devices_count"] += 1
    return result