Skip to content

Instantly share code, notes, and snippets.

@esc5221

esc5221/extract_conversation.py

Created September 7, 2025 14:20
Show Gist options

  • Select an option

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

Select an option

Learn more about clone URLs
Save esc5221/47d91a711210b4a0c892f909b07c2949 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
extract_conversation.py
#!/usr/bin/env python3
"""
최근 N시간 Claude 대화를 raw 형태로 추출
python3 extract_conversation.py --hours 24 --output chat.txt
"""
import json
import os
from pathlib import Path
from datetime import datetime, timedelta, timezone
import argparse
from collections import defaultdict
def truncate_text(text, max_length=800):
"""긴 텍스트를 앞/뒤만 보이게 자르기 (앞이 더 길게)"""
if len(text) <= max_length:
return text
# 앞 70%, 뒤 30% 비율로 자르기
front_length = int(max_length * 0.7)
back_length = int(max_length * 0.3)
front_part = text[:front_length].rstrip()
back_part = text[-back_length:].lstrip()
return f"{front_part}\n\n... [truncated] ...\n\n{back_part}"
def extract_text_content(content, hide_file_tools=False):
"""content에서 텍스트 부분만 추출"""
if isinstance(content, str):
return content.strip()
if isinstance(content, list):
text_parts = []
for item in content:
if isinstance(item, dict):
if item.get('type') == 'text':
text_parts.append(item.get('text', ''))
elif item.get('type') == 'tool_use':
# tool_use를 간단히 요약
tool_name = item.get('name', 'unknown')
input_data = item.get('input', {})
# 파일 관련 툴 숨기기 옵션
if hide_file_tools and tool_name in ['Read', 'Edit', 'Write', 'TodoWrite']:
continue
# 툴별로 주요 파라미터 추출
if tool_name == 'Bash':
command = input_data.get('command', '')[:120]
text_parts.append(f"[{tool_name}: {command}]")
elif tool_name == 'Grep':
pattern = input_data.get('pattern', '')[:80]
text_parts.append(f"[{tool_name}: {pattern}]")
elif tool_name == 'Read':
file_path = input_data.get('file_path', '')[:100]
text_parts.append(f"[{tool_name}: {file_path}]")
elif tool_name == 'Edit':
file_path = input_data.get('file_path', '')[:100]
text_parts.append(f"[{tool_name}: {file_path}]")
elif tool_name == 'Write':
file_path = input_data.get('file_path', '')[:100]
text_parts.append(f"[{tool_name}: {file_path}]")
elif tool_name == 'Glob':
pattern = input_data.get('pattern', '')[:80]
text_parts.append(f"[{tool_name}: {pattern}]")
elif tool_name == 'TodoWrite':
text_parts.append(f"[{tool_name}]")
else:
text_parts.append(f"[{tool_name}]")
return ' '.join(text_parts).strip()
return str(content).strip()
def get_session_conversations(projects_dir, hours=8):
"""최근 N시간의 대화를 세션별로 추출"""
current_time = datetime.now(timezone.utc)
cutoff_time = current_time - timedelta(hours=hours)
sessions = defaultdict(list)
# 모든 프로젝트 디렉토리 순회
for project_dir in Path(projects_dir).iterdir():
if not project_dir.is_dir():
continue
# 프로젝트 내 모든 JSONL 파일 처리
for jsonl_file in project_dir.glob("*.jsonl"):
try:
with open(jsonl_file, 'r', encoding='utf-8') as f:
for line in f:
try:
data = json.loads(line.strip())
# 시간 필터링
timestamp_str = data.get('timestamp', '')
if timestamp_str:
timestamp = datetime.fromisoformat(timestamp_str.replace('Z', '+00:00'))
if timestamp < cutoff_time:
continue
# 사용자/어시스턴트 메시지만 수집
if data.get('type') in ['user', 'assistant'] and 'message' in data:
message_data = {
'timestamp': timestamp_str,
'type': data['type'],
'message': data['message'],
'sessionId': data.get('sessionId', 'unknown'),
'cwd': data.get('cwd', ''),
'project': project_dir.name
}
sessions[data.get('sessionId', 'unknown')].append(message_data)
except (json.JSONDecodeError, ValueError):
continue
except Exception as e:
print(f"파일 읽기 오류: {jsonl_file} - {e}")
continue
return sessions
def format_conversation(sessions, max_text_length=800, hide_file_tools=False):
"""세션별 대화를 raw 형태로 포맷"""
output_lines = []
# 세션을 시간 역순으로 정렬 (최신이 위로)
sorted_sessions = []
for session_id, messages in sessions.items():
if messages:
# 세션 내 메시지를 시간순 정렬
messages.sort(key=lambda x: x.get('timestamp', ''))
first_msg = messages[0]
last_msg = messages[-1]
sorted_sessions.append({
'session_id': session_id,
'messages': messages,
'start_time': first_msg.get('timestamp', ''),
'end_time': last_msg.get('timestamp', ''),
'cwd': first_msg.get('cwd', ''),
'project': first_msg.get('project', '')
})
# 시간 정순 정렬 (오래된 것이 위로)
sorted_sessions.sort(key=lambda x: x['start_time'], reverse=False)
for session in sorted_sessions:
# 세션 헤더
start_time = session['start_time']
end_time = session['end_time']
cwd = session['cwd']
try:
start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00'))
end_dt = datetime.fromisoformat(end_time.replace('Z', '+00:00'))
start_str = start_dt.strftime('%H:%M')
end_str = end_dt.strftime('%H:%M')
time_range = f"{start_str}-{end_str}"
except:
time_range = "unknown"
output_lines.append(f"=== 세션: {cwd} ({time_range}) ===")
output_lines.append("")
# 대화 내용
prev_timestamp = None
for msg in session['messages']:
msg_type = msg['type']
message = msg['message']
content = extract_text_content(message.get('content', ''), hide_file_tools)
if not content.strip():
continue
# 10분 이상 간격 체크
current_timestamp_str = msg.get('timestamp', '')
if prev_timestamp and current_timestamp_str:
try:
prev_dt = datetime.fromisoformat(prev_timestamp.replace('Z', '+00:00'))
current_dt = datetime.fromisoformat(current_timestamp_str.replace('Z', '+00:00'))
time_diff = (current_dt - prev_dt).total_seconds() / 60 # 분 단위
if time_diff >= 10:
output_lines.append("---")
output_lines.append("")
except:
pass
# 텍스트 길이 제한
content = truncate_text(content, max_text_length)
if msg_type == 'user':
output_lines.append(f"user: {content}")
elif msg_type == 'assistant':
output_lines.append(f"assistant: {content}")
output_lines.append("")
prev_timestamp = current_timestamp_str
output_lines.append("") # 세션 간 구분
return '\n'.join(output_lines)
def main():
parser = argparse.ArgumentParser(description='Claude 대화를 raw 형태로 추출')
parser.add_argument('--hours',
type=int,
default=8,
help='추출할 시간 범위 (기본: 8시간)')
parser.add_argument('--projects-dir',
default=str(Path.home() / '.claude' / 'projects'),
help='Claude projects 디렉토리 경로')
parser.add_argument('--output',
help='출력 파일 경로 (미지정시 stdout)')
parser.add_argument('--max-length',
type=int,
default=800,
help='텍스트 최대 길이 (기본: 800자)')
parser.add_argument('--hide-file-tools',
action='store_true',
help='Read, Edit, Write, TodoWrite 툴 표시 숨기기')
args = parser.parse_args()
print(f"최근 {args.hours}시간 대화 추출 중...")
# 대화 추출
sessions = get_session_conversations(args.projects_dir, args.hours)
if not sessions:
print("추출된 대화가 없습니다.")
return
# 포맷팅
conversation_text = format_conversation(sessions, args.max_length, args.hide_file_tools)
# 출력
if args.output:
with open(args.output, 'w', encoding='utf-8') as f:
f.write(conversation_text)
print(f"대화 내용을 {args.output}에 저장했습니다.")
else:
print(conversation_text)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment