main.py

#!/usr/bin/env python3
"""
Claude Code Transcript Explorer - Main Entry Point

A comprehensive tool for searching, analyzing, and exploring Claude Code conversation transcripts.
"""

import sys
import os
import argparse
import logging
from pathlib import Path

# Add current directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from ui.app import create_interface
from core.data_loader import TranscriptLoader
from core.search_engine import SearchEngine, SearchIndex, SearchQuery, SearchMode

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def main():
    """Main entry point for the application."""
    parser = argparse.ArgumentParser(
        description='Claude Code Transcript Explorer - Search and analyze conversation transcripts'
    )
    
    parser.add_argument(
        '--mode',
        choices=['gui', 'cli', 'export'],
        default='gui',
        help='Run mode: gui (Gradio interface), cli (command line), or export (batch export)'
    )
    
    parser.add_argument(
        '--data-path',
        type=str,
        default='/home/ygg/.claude/projects',
        help='Path to transcript data directory'
    )
    
    parser.add_argument(
        '--port',
        type=int,
        default=7860,
        help='Port for Gradio server (GUI mode only)'
    )
    
    parser.add_argument(
        '--share',
        action='store_true',
        help='Create a public share link (GUI mode only)'
    )
    
    parser.add_argument(
        '--search',
        type=str,
        help='Search query (CLI mode only)'
    )
    
    parser.add_argument(
        '--export-format',
        choices=['json', 'csv', 'markdown'],
        default='json',
        help='Export format for results'
    )
    
    parser.add_argument(
        '--output',
        type=str,
        help='Output file path for exports'
    )
    
    parser.add_argument(
        '--cache-dir',
        type=str,
        default='/tmp/transcript_explorer_cache',
        help='Directory for caching search indices'
    )
    
    args = parser.parse_args()
    
    try:
        if args.mode == 'gui':
            run_gui(args)
        elif args.mode == 'cli':
            run_cli(args)
        elif args.mode == 'export':
            run_export(args)
    except KeyboardInterrupt:
        logger.info("Application terminated by user")
        sys.exit(0)
    except Exception as e:
        logger.error(f"Application error: {e}")
        sys.exit(1)


def run_gui(args):
    """Run the Gradio GUI interface.
    
    Args:
        args: Command line arguments
    """
    logger.info("Starting Gradio interface...")
    
    # Create and launch the interface
    app = create_interface()
    
    logger.info(f"Launching server on port {args.port}")
    app.launch(
        server_name="0.0.0.0",
        server_port=args.port,
        share=args.share,
        show_error=True
    )


def run_cli(args):
    """Run command-line search.
    
    Args:
        args: Command line arguments
    """
    if not args.search:
        print("Error: --search query required for CLI mode")
        sys.exit(1)
    
    logger.info("Running CLI search...")
    
    # Initialize components
    loader = TranscriptLoader(args.data_path)
    search_engine = SearchEngine(SearchIndex(args.cache_dir))
    
    # Check for cached index
    if not search_engine.index.load_index_from_cache():
        print("Loading transcript files...")
        messages, conversations = loader.load_all_files(show_progress=True)
        
        print("Building search index...")
        search_engine.index.build_index(messages, conversations)
    else:
        print("Using cached search index")
        # Still need to load messages and conversations for results
        messages, conversations = loader.load_all_files(show_progress=True)
        search_engine.index.messages = messages
        search_engine.index.conversations = conversations
    
    # Create search query
    query = SearchQuery(
        text=args.search,
        mode=SearchMode.FUZZY,
        limit=50
    )
    
    # Execute search
    print(f"\nSearching for: '{args.search}'...")
    results = search_engine.search(query)
    
    # Display results
    if not results:
        print("No results found.")
    else:
        print(f"\nFound {len(results)} results:\n")
        print("-" * 80)
        
        for i, result in enumerate(results[:10], 1):
            msg = result.message
            print(f"\n{i}. [{msg.role.value.upper()}] {msg.timestamp.strftime('%Y-%m-%d %H:%M')}")
            print(f"   Session: {msg.session_id[:8]}...")
            print(f"   Score: {result.score:.2f}")
            print(f"   Content: {result.get_context_snippet(200)}")
            
            if msg.tool_calls:
                tools = ", ".join([tc.tool_name for tc in msg.tool_calls])
                print(f"   Tools: {tools}")
            
            print("-" * 80)
        
        if len(results) > 10:
            print(f"\n... and {len(results) - 10} more results")
    
    # Export if requested
    if args.output:
        export_results(results, args.export_format, args.output)
        print(f"\nResults exported to: {args.output}")


def run_export(args):
    """Run batch export of all transcripts.
    
    Args:
        args: Command line arguments
    """
    logger.info("Running batch export...")
    
    # Initialize loader
    loader = TranscriptLoader(args.data_path)
    
    print("Loading transcript files...")
    messages, conversations = loader.load_all_files(show_progress=True)
    
    # Get statistics
    stats = loader.get_statistics()
    
    print(f"\nLoaded {stats['total_messages']} messages from {stats['total_conversations']} conversations")
    
    # Determine output path
    if args.output:
        output_path = Path(args.output)
    else:
        output_path = Path(f"transcript_export.{args.export_format}")
    
    # Export based on format
    if args.export_format == 'json':
        import json
        
        export_data = {
            'statistics': stats,
            'conversations': []
        }
        
        for conv_id, conv in conversations.items():
            conv_data = {
                'session_id': conv_id,
                'start_time': conv.start_time.isoformat(),
                'end_time': conv.end_time.isoformat(),
                'duration_minutes': conv.get_duration_minutes(),
                'total_messages': conv.total_messages,
                'messages': [msg.to_dict() for msg in conv.messages]
            }
            export_data['conversations'].append(conv_data)
        
        with open(output_path, 'w') as f:
            json.dump(export_data, f, indent=2, default=str)
    
    elif args.export_format == 'csv':
        import pandas as pd
        
        rows = []
        for msg in messages:
            rows.append({
                'timestamp': msg.timestamp.isoformat(),
                'session_id': msg.session_id,
                'role': msg.role.value,
                'type': msg.type.value,
                'content': msg.content[:1000],  # Truncate long content
                'tools': ', '.join([tc.tool_name for tc in msg.tool_calls]),
                'model': msg.model or '',
                'cwd': msg.cwd or ''
            })
        
        df = pd.DataFrame(rows)
        df.to_csv(output_path, index=False)
    
    elif args.export_format == 'markdown':
        with open(output_path, 'w') as f:
            f.write("# Claude Code Transcript Export\n\n")
            f.write(f"Generated: {datetime.now().isoformat()}\n\n")
            
            f.write("## Statistics\n\n")
            for key, value in stats.items():
                if isinstance(value, list):
                    value = len(value)
                f.write(f"- **{key}**: {value}\n")
            
            f.write("\n## Conversations\n\n")
            
            for conv_id, conv in list(conversations.items())[:10]:  # Limit to 10 for markdown
                f.write(f"### Session {conv_id[:8]}...\n\n")
                f.write(f"- Start: {conv.start_time}\n")
                f.write(f"- Duration: {conv.get_duration_minutes():.1f} minutes\n")
                f.write(f"- Messages: {conv.total_messages}\n\n")
                
                for msg in conv.messages[:5]:  # First 5 messages
                    f.write(f"**[{msg.role.value.upper()}]** {msg.timestamp.strftime('%H:%M:%S')}\n\n")
                    f.write(f"```\n{msg.content[:500]}\n```\n\n")
                
                if conv.total_messages > 5:
                    f.write(f"... and {conv.total_messages - 5} more messages\n\n")
                
                f.write("---\n\n")
    
    print(f"Export complete: {output_path}")
    print(f"File size: {output_path.stat().st_size / 1024 / 1024:.2f} MB")


def export_results(results, format, output_path):
    """Export search results to file.
    
    Args:
        results: Search results to export
        format: Export format
        output_path: Output file path
    """
    from datetime import datetime
    import json
    import pandas as pd
    
    if format == 'json':
        export_data = []
        for result in results:
            msg = result.message
            export_data.append({
                'timestamp': msg.timestamp.isoformat(),
                'session_id': msg.session_id,
                'role': msg.role.value,
                'content': msg.content,
                'score': result.score,
                'matched_fields': result.matched_fields,
                'tools': [tc.tool_name for tc in msg.tool_calls]
            })
        
        with open(output_path, 'w') as f:
            json.dump(export_data, f, indent=2, default=str)
    
    elif format == 'csv':
        rows = []
        for result in results:
            msg = result.message
            rows.append({
                'timestamp': msg.timestamp.isoformat(),
                'session_id': msg.session_id,
                'role': msg.role.value,
                'content': msg.content[:500],
                'score': result.score,
                'tools': ', '.join([tc.tool_name for tc in msg.tool_calls])
            })
        
        df = pd.DataFrame(rows)
        df.to_csv(output_path, index=False)
    
    elif format == 'markdown':
        with open(output_path, 'w') as f:
            f.write("# Search Results\n\n")
            f.write(f"Query executed: {datetime.now().isoformat()}\n\n")
            
            for i, result in enumerate(results, 1):
                msg = result.message
                f.write(f"## Result {i}\n\n")
                f.write(f"- **Score**: {result.score:.2f}\n")
                f.write(f"- **Timestamp**: {msg.timestamp}\n")
                f.write(f"- **Role**: {msg.role.value}\n")
                f.write(f"- **Session**: {msg.session_id[:8]}...\n\n")
                f.write(f"**Content:**\n```\n{msg.content}\n```\n\n")
                f.write("---\n\n")


if __name__ == "__main__":
    main()