Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 48 additions & 32 deletions src/extract_claude_logs.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,22 @@
from typing import Dict, List, Optional, Tuple


# Constants
INDENT_NUMBER = 2
MAJOR_SEPARATOR_WIDTH = 60
MINOR_SEPARATOR_WIDTH = 40
SESSION_ID_MAX_LENGTH = 8
LINES_SHOWN_MESSAGE = 8
LINES_PER_PAGE_MESSAGE = 30
MAX_LINES_PER_MESSAGE_DISPLAY = 50
MAX_LINE_LENGTH_DISPLAY = 100
MIN_PREVIEW_TEXT_LENGTH = 3
PREVIEW_TEXT_TRUNCATE_LENGTH = 100
PREVIEW_ERROR_TRUNCATE_LENGTH = 30
LIST_SEPARATOR_WIDTH = 80
SEARCH_MAX_RESULTS_DEFAULT = 30


class ClaudeConversationExtractor:
"""Extract and convert Claude Code conversations from JSONL to markdown."""

Expand Down Expand Up @@ -122,7 +138,7 @@ def extract_conversation(self, jsonl_path: Path, detailed: bool = False) -> List
conversation.append(
{
"role": "tool_use",
"content": f"🔧 Tool: {tool_name}\nInput: {json.dumps(tool_input, indent=2)}",
"content": f"🔧 Tool: {tool_name}\nInput: {json.dumps(tool_input, indent=INDENT_NUMBER)}",
"timestamp": entry.get("timestamp", ""),
}
)
Expand Down Expand Up @@ -183,7 +199,7 @@ def _extract_text_content(self, content, detailed: bool = False) -> str:
tool_name = item.get("name", "unknown")
tool_input = item.get("input", {})
text_parts.append(f"\n🔧 Using tool: {tool_name}")
text_parts.append(f"Input: {json.dumps(tool_input, indent=2)}\n")
text_parts.append(f"Input: {json.dumps(tool_input, indent=INDENT_NUMBER)}\n")
return "\n".join(text_parts)
else:
return str(content)
Expand All @@ -208,9 +224,9 @@ def display_conversation(self, jsonl_path: Path, detailed: bool = False) -> None

# Clear screen and show header
print("\033[2J\033[H", end="") # Clear screen
print("=" * 60)
print("=" * MAJOR_SEPARATOR_WIDTH)
print(f"📄 Viewing: {jsonl_path.parent.name}")
print(f"Session: {session_id[:8]}...")
print(f"Session: {session_id[:SESSION_ID_MAX_LENGTH]}...")

# Get timestamp from first message
first_timestamp = messages[0].get("timestamp", "")
Expand All @@ -221,26 +237,26 @@ def display_conversation(self, jsonl_path: Path, detailed: bool = False) -> None
except Exception:
pass

print("=" * 60)
print("=" * MAJOR_SEPARATOR_WIDTH)
print("↑↓ to scroll • Q to quit • Enter to continue\n")

# Display messages with pagination
lines_shown = 8 # Header lines
lines_per_page = 30
lines_shown = LINES_SHOWN_MESSAGE # Header lines
lines_per_page = LINES_PER_PAGE_MESSAGE

for i, msg in enumerate(messages):
role = msg["role"]
content = msg["content"]

# Format role display
if role == "user" or role == "human":
print(f"\n{'─' * 40}")
print(f"\n{'─' * MINOR_SEPARATOR_WIDTH}")
print(f"👤 HUMAN:")
print(f"{'─' * 40}")
print(f"{'─' * MINOR_SEPARATOR_WIDTH}")
elif role == "assistant":
print(f"\n{'─' * 40}")
print(f"\n{'─' * MINOR_SEPARATOR_WIDTH}")
print(f"🤖 CLAUDE:")
print(f"{'─' * 40}")
print(f"{'─' * MINOR_SEPARATOR_WIDTH}")
elif role == "tool_use":
print(f"\n🔧 TOOL USE:")
elif role == "tool_result":
Expand All @@ -252,12 +268,12 @@ def display_conversation(self, jsonl_path: Path, detailed: bool = False) -> None

# Display content (limit very long messages)
lines = content.split('\n')
max_lines_per_msg = 50
max_lines_per_msg = MAX_LINES_PER_MESSAGE_DISPLAY

for line_idx, line in enumerate(lines[:max_lines_per_msg]):
# Wrap very long lines
if len(line) > 100:
line = line[:97] + "..."
if len(line) > MAX_LINE_LENGTH_DISPLAY:
line = line[:(MAX_LINE_LENGTH_DISPLAY - 3)] + "..."
print(line)
lines_shown += 1

Expand All @@ -275,9 +291,9 @@ def display_conversation(self, jsonl_path: Path, detailed: bool = False) -> None
print(f"... [{len(lines) - max_lines_per_msg} more lines truncated]")
lines_shown += 1

print("\n" + "=" * 60)
print("\n" + "=" * MAJOR_SEPARATOR_WIDTH)
print("📄 End of conversation")
print("=" * 60)
print("=" * MAJOR_SEPARATOR_WIDTH)
input("\nPress Enter to continue...")

except Exception as e:
Expand Down Expand Up @@ -306,7 +322,7 @@ def save_as_markdown(
date_str = datetime.now().strftime("%Y-%m-%d")
time_str = ""

filename = f"claude-conversation-{date_str}-{session_id[:8]}.md"
filename = f"claude-conversation-{date_str}-{session_id[:SESSION_ID_MAX_LENGTH]}.md"
output_path = self.output_dir / filename

with open(output_path, "w", encoding="utf-8") as f:
Expand Down Expand Up @@ -361,7 +377,7 @@ def save_as_json(
else:
date_str = datetime.now().strftime("%Y-%m-%d")

filename = f"claude-conversation-{date_str}-{session_id[:8]}.json"
filename = f"claude-conversation-{date_str}-{session_id[:SESSION_ID_MAX_LENGTH]}.json"
output_path = self.output_dir / filename

# Create JSON structure
Expand All @@ -373,7 +389,7 @@ def save_as_json(
}

with open(output_path, "w", encoding="utf-8") as f:
json.dump(output, f, indent=2, ensure_ascii=False)
json.dump(output, f, indent=INDENT_NUMBER, ensure_ascii=False)

return output_path

Expand All @@ -398,7 +414,7 @@ def save_as_html(
date_str = datetime.now().strftime("%Y-%m-%d")
time_str = ""

filename = f"claude-conversation-{date_str}-{session_id[:8]}.html"
filename = f"claude-conversation-{date_str}-{session_id[:SESSION_ID_MAX_LENGTH]}.html"
output_path = self.output_dir / filename

# HTML template with modern styling
Expand All @@ -407,7 +423,7 @@ def save_as_html(
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Claude Conversation - {session_id[:8]}</title>
<title>Claude Conversation - {session_id[:SESSION_ID_MAX_LENGTH]}</title>
<style>
body {{
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
Expand Down Expand Up @@ -593,8 +609,8 @@ def get_conversation_preview(self, session_path: Path) -> Tuple[str, int]:
text = parts[1].strip()

# If we have real user text, use it
if text and len(text) > 3: # Lower threshold to catch "hello"
first_user_msg = text[:100].replace('\n', ' ')
if text and len(text) > MIN_PREVIEW_TEXT_LENGTH: # Lower threshold to catch "hello"
first_user_msg = text[:PREVIEW_TEXT_TRUNCATE_LENGTH].replace('\n', ' ')
break

# Handle string content (less common but possible)
Expand All @@ -615,14 +631,14 @@ def get_conversation_preview(self, session_path: Path) -> Tuple[str, int]:

# Skip tool results and interruptions
if not content.startswith("tool_use_id") and "[Request interrupted" not in content:
if content and len(content) > 3: # Lower threshold to catch short messages
first_user_msg = content[:100].replace('\n', ' ')
if content and len(content) > MIN_PREVIEW_TEXT_LENGTH: # Lower threshold to catch short messages
first_user_msg = content[:PREVIEW_TEXT_TRUNCATE_LENGTH].replace('\n', ' ')
except json.JSONDecodeError:
continue

return first_user_msg or "No preview available", msg_count
except Exception as e:
return f"Error: {str(e)[:30]}", 0
return f"Error: {str(e)[:PREVIEW_ERROR_TRUNCATE_LENGTH]}", 0

def list_recent_sessions(self, limit: int = None) -> List[Path]:
"""List recent sessions with details."""
Expand All @@ -634,7 +650,7 @@ def list_recent_sessions(self, limit: int = None) -> List[Path]:
return []

print(f"\n📚 Found {len(sessions)} Claude sessions:\n")
print("=" * 80)
print("=" * LIST_SEPARATOR_WIDTH)

# Show all sessions if no limit specified
sessions_to_show = sessions[:limit] if limit else sessions
Expand All @@ -656,13 +672,13 @@ def list_recent_sessions(self, limit: int = None) -> List[Path]:

# Print formatted info
print(f"\n{i}. 📁 {project}")
print(f" 📄 Session: {session_id[:8]}...")
print(f" 📄 Session: {session_id[:SESSION_ID_MAX_LENGTH]}...")
print(f" 📅 Modified: {modified.strftime('%Y-%m-%d %H:%M')}")
print(f" 💬 Messages: {msg_count}")
print(f" 💾 Size: {size_kb:.1f} KB")
print(f" 📝 Preview: \"{preview}...\"")

print("\n" + "=" * 80)
print("\n" + "=" * LIST_SEPARATOR_WIDTH)
return sessions[:limit]

def extract_multiple(
Expand Down Expand Up @@ -844,7 +860,7 @@ def main():
date_to=date_to,
speaker_filter=speaker_filter,
case_sensitive=args.case_sensitive,
max_results=30,
max_results=SEARCH_MAX_RESULTS_DEFAULT,
)

if not results:
Expand All @@ -867,11 +883,11 @@ def main():
print(f"\n{len(file_paths_list)}. 📄 {file_path.parent.name} ({len(file_results)} matches)")
# Show first match preview
first = file_results[0]
print(f" {first.speaker}: {first.matched_content[:100]}...")
print(f" {first.speaker}: {first.matched_content[:PREVIEW_TEXT_TRUNCATE_LENGTH]}...")

# Offer to view conversations
if file_paths_list:
print("\n" + "=" * 60)
print("\n" + "=" * MAJOR_SEPARATOR_WIDTH)
try:
view_choice = input("\nView a conversation? Enter number (1-{}) or press Enter to skip: ".format(
len(file_paths_list))).strip()
Expand Down
22 changes: 15 additions & 7 deletions src/interactive_ui.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,14 @@
from search_conversations import ConversationSearcher


# Constants
SESSION_DISPLAY_LIMIT = 20
PROJECT_LENGTH = 30
MAJOR_SEPARATOR_WIDTH = 60
PROGRESS_BAR_WIDTH = 40
RECENT_SESSIONS_LIMIT = 5


class InteractiveUI:
"""Interactive terminal UI for easier conversation extraction"""

Expand Down Expand Up @@ -120,18 +128,18 @@ def show_sessions_menu(self) -> List[int]:
print(f"\n✅ Found {len(self.sessions)} conversations!\n")

# Display sessions
for i, session_path in enumerate(self.sessions[:20], 1): # Show max 20
for i, session_path in enumerate(self.sessions[:SESSION_DISPLAY_LIMIT], 1): # Show max SESSION_DISPLAY_LIMIT
project = session_path.parent.name
modified = datetime.fromtimestamp(session_path.stat().st_mtime)
size_kb = session_path.stat().st_size / 1024

date_str = modified.strftime("%Y-%m-%d %H:%M")
print(f" {i:2d}. [{date_str}] {project[:30]:<30} ({size_kb:.1f} KB)")
print(f" {i:2d}. [{date_str}] {project[:PROJECT_LENGTH]:<{PROJECT_LENGTH}} ({size_kb:.1f} KB)")

if len(self.sessions) > 20:
print(f"\n ... and {len(self.sessions) - 20} more conversations")
if len(self.sessions) > SESSION_DISPLAY_LIMIT:
print(f"\n ... and {len(self.sessions) - SESSION_DISPLAY_LIMIT} more conversations")

print("\n" + "=" * 60)
print("\n" + "=" * MAJOR_SEPARATOR_WIDTH)
print("\nOptions:")
print(" A. Extract ALL conversations")
print(" R. Extract 5 most RECENT")
Expand All @@ -147,7 +155,7 @@ def show_sessions_menu(self) -> List[int]:
elif choice == "A":
return list(range(len(self.sessions)))
elif choice == "R":
return list(range(min(5, len(self.sessions))))
return list(range(min(RECENT_SESSIONS_LIMIT, len(self.sessions))))
elif choice == "S":
selection = input("Enter conversation numbers (e.g., 1,3,5): ").strip()
try:
Expand All @@ -169,7 +177,7 @@ def show_sessions_menu(self) -> List[int]:

def show_progress(self, current: int, total: int, message: str = ""):
"""Display a simple progress bar"""
bar_width = 40
bar_width = PROGRESS_BAR_WIDTH
progress = current / total if total > 0 else 0
filled = int(bar_width * progress)
bar = "█" * filled + "░" * (bar_width - filled)
Expand Down
Loading