Skip to content

Instantly share code, notes, and snippets.

@esc5221

esc5221/claude session visualize

Created November 23, 2025 14:22
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save esc5221/4439ccd5809b35059b54e9897081b29b to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save esc5221/4439ccd5809b35059b54e9897081b29b to your computer and use it in GitHub Desktop.
Download ZIP
Raw
claude session visualize
#!/usr/bin/env python3
"""
Claude Code 세션 분석 및 시각화 통합 CLI
Usage:
python3 analyze_visualize.py --hours 24 # 최근 24시간
python3 analyze_visualize.py --days 7 # 최근 7일
python3 analyze_visualize.py --days 1 # 최근 1일 (어제+오늘)
python3 analyze_visualize.py --date 2025-11-06 2025-11-07 # 특정 날짜 범위
"""
import argparse
import json
from pathlib import Path
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import re
from collections import defaultdict
import plotly.graph_objects as go
from plotly.subplots import make_subplots
# 프로젝트 이름 파싱 함수
def parse_project_name(dir_name):
"""
-Users-lullu-turing-gpai-gpai-monorepo-4 -> gpai/monorepo-4
-Users-lullu-projects-x37 -> projects/x37
-Users-lullu-study-retrovibe -> study/retrovibe
"""
cleaned = dir_name.replace('-Users-lullu-', '', 1)
parts = cleaned.split('-')
if len(parts) >= 2:
if 'turing' in parts[0]:
if 'gpai' in cleaned:
match = re.search(r'gpai-(monorepo(?:-\d+|-\w+)*)', cleaned)
if match:
return f"gpai/{match.group(1)}"
return f"gpai/{'-'.join(parts[2:])}" if len(parts) > 2 else cleaned
return f"turing/{'-'.join(parts[1:])}"
elif 'projects' in parts[0]:
return f"projects/{'-'.join(parts[1:])}" if len(parts) > 1 else 'projects'
elif 'study' in parts[0]:
return f"study/{'-'.join(parts[1:])}" if len(parts) > 1 else 'study'
elif 'claude' in parts[0]:
return f"claude/{'-'.join(parts[1:])}" if len(parts) > 1 else 'claude'
return cleaned
def get_project_group(project_name):
"""gpai/monorepo-4 -> gpai"""
if '/' in project_name:
return project_name.split('/')[0]
return 'other'
def analyze_sessions(target_dates, output_file='session_activities.json'):
"""Analyze session data"""
print('=' * 100)
print(f'📊 Project Session Analysis ({", ".join(target_dates)}, KST)')
print('=' * 100)
print()
projects_dir = Path.home() / '.claude/projects'
session_activities = []
for project_dir in projects_dir.iterdir():
if not project_dir.is_dir():
continue
for session_file in project_dir.glob('*.jsonl'):
if session_file.name.startswith('agent-'):
continue
try:
with open(session_file, 'r') as f:
events = [json.loads(line) for line in f if line.strip()]
all_events = []
user_messages = []
assistant_messages = []
tool_uses = []
for event in events:
if event.get('timestamp'):
try:
ts_utc = datetime.fromisoformat(event['timestamp'].replace('Z', '+00:00'))
ts_kst = ts_utc.astimezone(ZoneInfo('Asia/Seoul'))
event_data = {
'type': event.get('type'),
'timestamp': ts_kst,
'uuid': event.get('uuid')
}
all_events.append(event_data)
if event.get('type') == 'user':
user_messages.append(event_data)
elif event.get('type') == 'assistant':
assistant_messages.append(event_data)
if 'message' in event and isinstance(event['message'].get('content'), list):
for block in event['message']['content']:
if block.get('type') == 'tool_use':
tool_uses.append({
'tool_name': block.get('name'),
'timestamp': ts_kst
})
except:
pass
if not all_events or not user_messages:
continue
all_events.sort(key=lambda x: x['timestamp'])
user_messages.sort(key=lambda x: x['timestamp'])
first_date = all_events[0]['timestamp'].strftime('%Y-%m-%d')
last_date = all_events[-1]['timestamp'].strftime('%Y-%m-%d')
if first_date not in target_dates and last_date not in target_dates:
continue
# 활동 구간 추출
active_periods = []
current_start = all_events[0]['timestamp']
last_time = all_events[0]['timestamp']
for event in all_events[1:]:
gap = (event['timestamp'] - last_time).total_seconds() / 60
if gap > 30:
active_periods.append({
'start': current_start,
'end': last_time,
'duration_minutes': (last_time - current_start).total_seconds() / 60
})
current_start = event['timestamp']
last_time = event['timestamp']
active_periods.append({
'start': current_start,
'end': last_time,
'duration_minutes': (last_time - current_start).total_seconds() / 60
})
# 휴식 구간 추출
idle_periods = []
for i in range(len(user_messages) - 1):
current_user_msg = user_messages[i]['timestamp']
next_user_msg = user_messages[i + 1]['timestamp']
gap_minutes = (next_user_msg - current_user_msg).total_seconds() / 60
if gap_minutes >= 30:
idle_periods.append({
'start': current_user_msg,
'end': next_user_msg,
'duration_minutes': gap_minutes
})
# 대상 날짜 필터링
target_active_periods = []
for p in active_periods:
if p['start'].strftime('%Y-%m-%d') in target_dates or p['end'].strftime('%Y-%m-%d') in target_dates:
target_active_periods.append(p)
target_idle_periods = []
for p in idle_periods:
if p['start'].strftime('%Y-%m-%d') in target_dates or p['end'].strftime('%Y-%m-%d') in target_dates:
target_idle_periods.append(p)
if target_active_periods:
project_full_name = parse_project_name(project_dir.name)
project_group = get_project_group(project_full_name)
session_activities.append({
'project': project_full_name,
'project_group': project_group,
'project_dir': project_dir.name,
'session_id': session_file.stem,
'total_messages': len(all_events),
'user_messages': len(user_messages),
'assistant_messages': len(assistant_messages),
'tool_uses': len(tool_uses),
'tool_details': tool_uses,
'active_periods': target_active_periods,
'idle_periods': target_idle_periods,
'total_active_time': sum(p['duration_minutes'] for p in target_active_periods),
'total_idle_time': sum(p['duration_minutes'] for p in target_idle_periods)
})
except Exception as e:
pass
# Statistics output
group_stats = defaultdict(lambda: {'sessions': 0, 'active': 0, 'idle': 0})
for sess in session_activities:
group = sess['project_group']
group_stats[group]['sessions'] += 1
group_stats[group]['active'] += sess['total_active_time']
group_stats[group]['idle'] += sess['total_idle_time']
print('📦 Project Group Statistics:')
for group in sorted(group_stats.keys()):
stats = group_stats[group]
print(f" • {group:15s}: {stats['sessions']:2d} sessions, {stats['active']/60:5.1f}h active, {stats['idle']/60:5.1f}h idle")
print()
# JSON으로 저장
output_data = []
for sess in session_activities:
periods_serializable = []
for p in sess['active_periods']:
periods_serializable.append({
'start': p['start'].isoformat(),
'end': p['end'].isoformat(),
'duration_minutes': p['duration_minutes']
})
idle_serializable = []
for p in sess['idle_periods']:
idle_serializable.append({
'start': p['start'].isoformat(),
'end': p['end'].isoformat(),
'duration_minutes': p['duration_minutes']
})
tools_serializable = []
for tool in sess['tool_details']:
tools_serializable.append({
'tool_name': tool['tool_name'],
'timestamp': tool['timestamp'].isoformat()
})
output_data.append({
'project': sess['project'],
'project_group': sess['project_group'],
'project_dir': sess['project_dir'],
'session_id': sess['session_id'],
'total_messages': sess['total_messages'],
'user_messages': sess['user_messages'],
'assistant_messages': sess['assistant_messages'],
'tool_uses': sess['tool_uses'],
'tool_details': tools_serializable,
'active_periods': periods_serializable,
'idle_periods': idle_serializable,
'total_active_time': sess['total_active_time'],
'total_idle_time': sess['total_idle_time']
})
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(output_data, f, indent=2, ensure_ascii=False)
print(f'✅ Data saved: {output_file}')
print()
return session_activities
def visualize_sessions(session_activities, date_range_str, x_range):
"""Visualize sessions"""
# 프로젝트 그룹별 색상
group_colors = {
'gpai': 'rgb(75, 192, 192)',
'projects': 'rgb(255, 159, 64)',
'claude': 'rgb(153, 102, 255)',
'study': 'rgb(255, 99, 132)',
'other': 'rgb(201, 203, 207)'
}
# 통계 계산
total_user_messages = 0
total_active_minutes = 0
for sess in session_activities:
total_user_messages += sess['user_messages']
for period in sess['active_periods']:
total_active_minutes += period['duration_minutes']
total_active_hours = total_active_minutes / 60
# 동시성 데이터 계산
time_events = []
for sess in session_activities:
for period in sess['active_periods']:
# period는 이미 datetime 객체
start_dt = period['start'] if isinstance(period['start'], datetime) else datetime.fromisoformat(period['start'])
end_dt = period['end'] if isinstance(period['end'], datetime) else datetime.fromisoformat(period['end'])
padded_start = start_dt - timedelta(minutes=5)
padded_end = end_dt + timedelta(minutes=5)
time_events.append(('START', padded_start, sess['session_id']))
time_events.append(('END', padded_end, sess['session_id']))
time_events.sort(key=lambda x: x[1])
active_sessions = set()
concurrency_timeline = []
prev_count = 0
for event_type, timestamp, session_id in time_events:
if len(concurrency_timeline) > 0:
concurrency_timeline.append({'time': timestamp, 'count': prev_count})
if event_type == 'START':
active_sessions.add(session_id)
else:
active_sessions.discard(session_id)
concurrency_timeline.append({'time': timestamp, 'count': len(active_sessions)})
prev_count = len(active_sessions)
max_concurrency = max([item['count'] for item in concurrency_timeline]) if concurrency_timeline else 0
avg_concurrency = sum([item['count'] for item in concurrency_timeline]) / len(concurrency_timeline) if concurrency_timeline else 0
# 프로젝트별 세션 수집 및 레인 할당
MAX_LANES = 5
project_sessions = defaultdict(list)
for sess in session_activities:
project = sess['project']
all_times = []
for period in sess['active_periods']:
start_dt = period['start'] if isinstance(period['start'], datetime) else datetime.fromisoformat(period['start'])
end_dt = period['end'] if isinstance(period['end'], datetime) else datetime.fromisoformat(period['end'])
all_times.append(start_dt)
all_times.append(end_dt)
if all_times:
session_start = min(all_times)
session_end = max(all_times)
project_sessions[project].append({
'session_id': sess['session_id'],
'project_group': sess['project_group'],
'start': session_start,
'end': session_end,
'active_periods': sess['active_periods'],
'idle_periods': sess['idle_periods'],
'total_messages': sess['total_messages'],
'user_messages': sess['user_messages'],
'assistant_messages': sess['assistant_messages'],
'tool_uses': sess['tool_uses']
})
def assign_session_lanes(sessions, max_lanes=5):
sorted_sessions = sorted(sessions, key=lambda x: x['start'])
lane_end_times = [None] * max_lanes
for session in sorted_sessions:
assigned = False
for lane_idx in range(max_lanes):
if lane_end_times[lane_idx] is None or lane_end_times[lane_idx] < session['start']:
session['lane'] = lane_idx
lane_end_times[lane_idx] = session['end']
assigned = True
break
if not assigned:
earliest_lane = min(range(max_lanes), key=lambda i: lane_end_times[i] or datetime.min)
session['lane'] = earliest_lane
lane_end_times[earliest_lane] = session['end']
used_lanes = max([s['lane'] for s in sorted_sessions]) + 1 if sorted_sessions else 0
return sorted_sessions, used_lanes
project_lane_data = {}
for project, sessions in project_sessions.items():
assigned_sessions, num_lanes = assign_session_lanes(sessions, MAX_LANES)
project_lane_data[project] = {
'sessions': assigned_sessions,
'num_lanes': num_lanes,
'project_group': assigned_sessions[0]['project_group'] if assigned_sessions else 'other'
}
y_labels = []
projects_sorted = sorted(project_lane_data.keys(),
key=lambda p: (project_lane_data[p]['project_group'], p))
for project in projects_sorted:
num_lanes = project_lane_data[project]['num_lanes']
for lane_idx in range(num_lanes):
y_labels.append(f"{project} L{lane_idx + 1}")
# Create figure
fig = make_subplots(
rows=3, cols=1,
row_heights=[0.12, 0.18, 0.70],
vertical_spacing=0.05, # Increased spacing between plots
subplot_titles=('', '⚡ Concurrent Sessions', '📅 Project Session Timeline (Lane-based)'),
specs=[[{"type": "xy"}], [{"type": "xy"}], [{"type": "xy"}]]
)
# Statistics cards
card_y_pos = 0.98 # Moved up slightly
card_height = 0.065 # Reduced height
card_width = 0.12 # Reduced width
cards = [
{'x': 0.17, 'title': '📊 Total Work Time', 'value': f'{total_active_hours:.1f}h', 'subtitle': f'{total_active_minutes:.0f} minutes'},
{'x': 0.5, 'title': '💬 User Inputs', 'value': f'{total_user_messages}', 'subtitle': 'messages sent'},
{'x': 0.83, 'title': '⚙️ Avg Concurrency', 'value': f'x{avg_concurrency:.1f}', 'subtitle': f'max {max_concurrency} concurrent'}
]
for card in cards:
fig.add_shape(
type="rect", xref="paper", yref="paper",
x0=card['x'] - card_width, y0=card_y_pos - card_height,
x1=card['x'] + card_width, y1=card_y_pos,
fillcolor="rgba(240, 240, 245, 0.8)",
line=dict(color="rgba(100, 100, 120, 0.3)", width=2),
layer="below"
)
fig.add_annotation(xref="paper", yref="paper", x=card['x'], y=card_y_pos - 0.012,
text=f"<b>{card['title']}</b>", showarrow=False,
font=dict(size=12, color="rgb(60, 60, 80)"), align="center")
fig.add_annotation(xref="paper", yref="paper", x=card['x'], y=card_y_pos - 0.035,
text=f"<b style='font-size:18px'>{card['value']}</b>", showarrow=False,
font=dict(size=18, color="rgb(40, 40, 60)"), align="center")
fig.add_annotation(xref="paper", yref="paper", x=card['x'], y=card_y_pos - 0.055,
text=card['subtitle'], showarrow=False,
font=dict(size=9, color="rgb(120, 120, 140)"), align="center")
# Concurrency graph
conc_times_filtered = []
conc_counts_filtered = []
for i, item in enumerate(concurrency_timeline):
if item['count'] == 0:
if i > 0 and concurrency_timeline[i-1]['count'] > 0:
conc_times_filtered.append(item['time'])
conc_counts_filtered.append(0)
conc_times_filtered.append(item['time'])
conc_counts_filtered.append(None)
else:
if i > 0 and concurrency_timeline[i-1]['count'] == 0:
conc_times_filtered.append(item['time'])
conc_counts_filtered.append(0)
conc_times_filtered.append(item['time'])
conc_counts_filtered.append(item['count'])
fig.add_trace(
go.Scatter(
x=conc_times_filtered, y=conc_counts_filtered,
mode='lines', fill='tozeroy',
line=dict(color='rgb(100, 150, 255)', width=2),
fillcolor='rgba(100, 150, 255, 0.3)',
name='Concurrent Sessions',
hovertemplate='<b>%{x|%m/%d %H:%M:%S}</b><br>Concurrent: %{y} sessions<extra></extra>',
connectgaps=False
),
row=2, col=1
)
# Timeline
for project in projects_sorted:
data = project_lane_data[project]
color = group_colors.get(data['project_group'], group_colors['other'])
for session in data['sessions']:
lane_label = f"{project} L{session['lane'] + 1}"
# Idle periods
for period in session['idle_periods']:
start_dt = period['start'] if isinstance(period['start'], datetime) else datetime.fromisoformat(period['start'])
end_dt = period['end'] if isinstance(period['end'], datetime) else datetime.fromisoformat(period['end'])
duration_ms = (end_dt - start_dt).total_seconds() * 1000
fig.add_trace(go.Bar(
name='Idle', x=[duration_ms], y=[lane_label], base=start_dt, orientation='h',
marker=dict(color='rgba(200, 200, 200, 0.3)', line=dict(color='rgba(150, 150, 150, 0.5)', width=0.5)),
hovertemplate=f"<b>💤 Idle</b><br>{project}<br>L{session['lane'] + 1}<br>{session['session_id'][:8]}<br>{period['duration_minutes']:.0f} min<extra></extra>",
showlegend=False, xaxis='x3', yaxis='y3'
), row=3, col=1)
# Active periods
for period in session['active_periods']:
start_dt = period['start'] if isinstance(period['start'], datetime) else datetime.fromisoformat(period['start'])
end_dt = period['end'] if isinstance(period['end'], datetime) else datetime.fromisoformat(period['end'])
duration_ms = (end_dt - start_dt).total_seconds() * 1000
fig.add_trace(go.Bar(
name=project, x=[duration_ms], y=[lane_label], base=start_dt, orientation='h',
marker=dict(color=color, line=dict(color='white', width=0.5)),
hovertemplate=f"<b>⚡ {project}</b><br>L{session['lane'] + 1}<br>{session['session_id'][:8]}<br>{period['duration_minutes']:.0f} min<br>💬 {session['total_messages']} msgs<extra></extra>",
showlegend=False, xaxis='x3', yaxis='y3'
), row=3, col=1)
# 그룹 구분선
current_group = None
group_boundaries = []
for i, label in enumerate(y_labels):
project = label.rsplit(' L', 1)[0]
project_group = project_lane_data[project]['project_group']
if project_group != current_group:
if current_group is not None:
group_boundaries.append(i - 0.5)
current_group = project_group
for boundary in group_boundaries:
fig.add_hline(y=boundary, line_dash="dot", line_color="gray", line_width=1.5, opacity=0.6, row=3, col=1)
# Layout
fig.update_layout(
title={'text': f'📊 Claude Code Session Analysis v6 ({date_range_str})<br><sub>Lane-based Compressed View - Max 5 lanes per project</sub>',
'x': 0.5, 'xanchor': 'center', 'font': {'size': 20}},
height=max(1000, len(y_labels) * 25 + 600),
barmode='overlay', hovermode='closest', plot_bgcolor='white',
showlegend=False, margin=dict(l=280, r=50, t=150, b=80)
)
fig.update_xaxes(type='date', tickformat='%H:%M', dtick=3600000, gridcolor='lightgray', range=x_range, row=2, col=1)
fig.update_xaxes(title='Time (KST)', type='date', tickformat='%H:%M', dtick=3600000, gridcolor='lightgray', range=x_range, row=3, col=1)
# Add date change lines
from datetime import datetime as dt_class
start_date = dt_class.fromisoformat(x_range[0])
end_date = dt_class.fromisoformat(x_range[1])
current = start_date.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)
while current < end_date:
# Add vertical line at midnight for date changes
fig.add_vline(x=current, line_dash="solid", line_color="rgba(100, 100, 100, 0.3)", line_width=2, row=2, col=1)
fig.add_vline(x=current, line_dash="solid", line_color="rgba(100, 100, 100, 0.3)", line_width=2, row=3, col=1)
# Add date annotation
fig.add_annotation(
x=current, y=1.0, yref="paper", xref="x2",
text=f"<b>{current.strftime('%m/%d')}</b>",
showarrow=False,
font=dict(size=11, color="rgb(80, 80, 80)"),
yshift=10
)
current += timedelta(days=1)
fig.update_yaxes(title='Concurrent Sessions', gridcolor='lightgray', row=2, col=1)
fig.update_yaxes(title='Project Lanes', categoryorder='array', categoryarray=y_labels[::-1], tickfont=dict(size=9), tickmode='linear', row=3, col=1)
output_file = '/Users/lullu/claude-work/session_timeline_v6.html'
fig.write_html(output_file)
print('✅ v6 lane-based visualization completed!')
print(f'📁 File: {output_file}')
print()
print('📊 Summary Statistics:')
print(f' • Total sessions: {len(session_activities)}')
print(f' • Y-axis lanes: {len(y_labels)} (compression: {(1 - len(y_labels)/len(session_activities))*100:.1f}%)')
print(f' • Total work time: {total_active_hours:.1f}h')
print(f' • User inputs: {total_user_messages}')
print(f' • Avg concurrency: x{avg_concurrency:.1f} (max {max_concurrency})')
def main():
parser = argparse.ArgumentParser(description='Claude Code session analysis and visualization')
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--hours', type=int, help='Last N hours')
group.add_argument('--days', type=int, help='Last N days')
group.add_argument('--date', nargs=2, help='Date range (YYYY-MM-DD YYYY-MM-DD)')
args = parser.parse_args()
kst = ZoneInfo('Asia/Seoul')
now = datetime.now(kst)
if args.hours:
start_time = now - timedelta(hours=args.hours)
target_dates = list(set([
start_time.strftime('%Y-%m-%d'),
now.strftime('%Y-%m-%d')
]))
date_range_str = f'Last {args.hours} hours'
x_range = [start_time.isoformat(), (now + timedelta(hours=1)).isoformat()]
elif args.days:
start_date = now - timedelta(days=args.days - 1)
target_dates = [(start_date + timedelta(days=i)).strftime('%Y-%m-%d') for i in range(args.days)]
date_range_str = f'Last {args.days} days'
x_range = [start_date.replace(hour=0, minute=0, second=0).isoformat(),
(now + timedelta(days=1)).replace(hour=0, minute=0, second=0).isoformat()]
else: # --date
target_dates = args.date
date_range_str = f'{args.date[0]} ~ {args.date[1]}'
start = datetime.fromisoformat(args.date[0]).replace(tzinfo=kst)
end = datetime.fromisoformat(args.date[1]).replace(tzinfo=kst) + timedelta(days=1)
x_range = [start.isoformat(), end.isoformat()]
# Run analysis
session_activities = analyze_sessions(target_dates)
# Run visualization
visualize_sessions(session_activities, date_range_str, x_range)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment