main.py

from __future__ import annotations

from copy import deepcopy
from typing import Any, Dict, Iterable, List


Session = Dict[str, Any]
Event = Dict[str, Any]


def _deep_merge_earliest(target: Dict[str, Any], src: Dict[str, Any]) -> Dict[str, Any]:
    """Deep merge src into target keeping earliest non-dict values."""
    for key, value in src.items():
        if key not in target:
            target[key] = deepcopy(value)
            continue

        existing = target[key]
        if isinstance(existing, dict) and isinstance(value, dict):
            _deep_merge_earliest(existing, value)
        else:
            # Keep earliest value already in target.
            continue
    return target


def _merge_session_meta(events_meta: Iterable[Dict[str, Any]]) -> Dict[str, Any]:
    merged: Dict[str, Any] = {}
    for meta in events_meta:
        if meta is None:
            continue
        _deep_merge_earliest(merged, meta)
    return merged


def _types_with_counts(types_in_order: List[str]) -> List[Dict[str, Any]]:
    """Return first-seen order with counts to preserve frequency without duplicates."""
    counts: Dict[str, int] = {}
    ordered_unique: List[str] = []
    for t in types_in_order:
        if t not in counts:
            ordered_unique.append(t)
            counts[t] = 0
        counts[t] += 1
    return [{"type": t, "count": counts[t]} for t in ordered_unique]


def merge_user_events(events: List[Event]) -> List[Session]:
    """Merge events into user sessions.

    Sessions are grouped per user_id and split when adjacent events are >600s apart.
    """
    # Work on deep copies to avoid in-place modifications.
    events_copy = [deepcopy(e) for e in events]

    # Group by user_id
    by_user: Dict[str, List[Event]] = {}
    for event in events_copy:
        user_id = event.get("user_id")
        by_user.setdefault(user_id, []).append(event)

    sessions: List[Session] = []

    for user_id, user_events in by_user.items():
        # Sort by timestamp
        user_events_sorted = sorted(user_events, key=lambda e: e.get("ts"))

        current_events: List[Event] = []
        last_ts = None

        for event in user_events_sorted:
            ts = event.get("ts")
            if last_ts is None or ts - last_ts <= 600:
                current_events.append(event)
            else:
                sessions.append(_build_session(user_id, current_events))
                current_events = [event]
            last_ts = ts

        if current_events:
            sessions.append(_build_session(user_id, current_events))

    # Sort across all users by start_ts
    sessions.sort(key=lambda s: s["start_ts"])
    return sessions


def _build_session(user_id: str, events: List[Event]) -> Session:
    events_sorted = sorted(events, key=lambda e: e.get("ts"))
    start_ts = events_sorted[0].get("ts")
    end_ts = events_sorted[-1].get("ts")
    types = _types_with_counts([e.get("type") for e in events_sorted])
    meta = _merge_session_meta([e.get("meta", {}) for e in events_sorted])

    return {
        "user_id": user_id,
        "start_ts": start_ts,
        "end_ts": end_ts,
        "types": types,
        "meta": meta,
    }