Claude Code History Files Finder

Security scan passed
Scanned at: 2025-11-26T00:38:49.440767
Tool: gitleaks + pattern-based validation
Content hash: 592122abb9a569998dfe7130eb891a5038eab3af0e8d46e0008c9d45640b4dad

#!/usr/bin/env python3
"""
Analyze Claude Code session files to find relevant sessions and statistics.

This script helps locate sessions containing specific keywords, analyze
session activity, and generate reports about session content.
"""

import json
import os
import sys
from pathlib import Path
from typing import Dict, List, Any, Optional
from datetime import datetime
from collections import defaultdict


class SessionAnalyzer:
    """Analyze Claude Code session history files."""

    def __init__(self, projects_dir: Optional[Path] = None):
        """
        Initialize analyzer.

        Args:
            projects_dir: Path to Claude projects directory
                         (default: ~/.claude/projects)
        """
        if projects_dir:
            self.projects_dir = Path(projects_dir)
        else:
            self.projects_dir = Path.home() / ".claude" / "projects"

    def find_project_sessions(self, project_path: str) -> List[Path]:
        """
        Find all session files for a specific project.

        Args:
            project_path: Project path (e.g., ~/Workspace/js/myproject)

        Returns:
            List of session file paths
        """
        # Convert project path to Claude's directory naming
        # Example: ~/Workspace/js/myproject -> -Users-<username>-Workspace-js-myproject
        normalized = project_path.replace("/", "-")
        project_dir = self.projects_dir / normalized

        if not project_dir.exists():
            return []

        # Find all session JSONL files (exclude agent files)
        sessions = []
        for file in project_dir.glob("*.jsonl"):
            if not file.name.startswith("agent-"):
                sessions.append(file)

        return sorted(sessions, key=lambda p: p.stat().st_mtime, reverse=True)

    def search_sessions(
        self, sessions: List[Path], keywords: List[str], case_sensitive: bool = False
    ) -> Dict[Path, Dict[str, Any]]:
        """
        Search sessions for keywords.

        Args:
            sessions: List of session file paths
            keywords: Keywords to search for
            case_sensitive: Whether to perform case-sensitive search

        Returns:
            Dict mapping session paths to match information
        """
        matches = {}

        for session_file in sessions:
            keyword_counts = defaultdict(int)
            total_mentions = 0

            try:
                with open(session_file, "r") as f:
                    for line in f:
                        try:
                            data = json.loads(line.strip())

                            # Extract text content from message
                            text_content = self._extract_text_content(data)

                            # Search for keywords
                            search_text = (
                                text_content if case_sensitive else text_content.lower()
                            )
                            for keyword in keywords:
                                search_keyword = (
                                    keyword if case_sensitive else keyword.lower()
                                )
                                count = search_text.count(search_keyword)
                                if count > 0:
                                    keyword_counts[keyword] += count
                                    total_mentions += count

                        except json.JSONDecodeError:
                            continue

                if total_mentions > 0:
                    matches[session_file] = {
                        "total_mentions": total_mentions,
                        "keyword_counts": dict(keyword_counts),
                        "modified_time": session_file.stat().st_mtime,
                        "size": session_file.stat().st_size,
                    }

            except Exception as e:
                print(
                    f"Warning: Error processing {session_file}: {e}", file=sys.stderr
                )
                continue

        return matches

    def get_session_stats(self, session_file: Path) -> Dict[str, Any]:
        """
        Get detailed statistics for a session file.

        Args:
            session_file: Path to session JSONL file

        Returns:
            Dictionary of session statistics
        """
        stats = {
            "total_lines": 0,
            "user_messages": 0,
            "assistant_messages": 0,
            "tool_uses": defaultdict(int),
            "write_calls": 0,
            "edit_calls": 0,
            "read_calls": 0,
            "bash_calls": 0,
            "file_operations": [],
        }

        try:
            with open(session_file, "r") as f:
                for line in f:
                    stats["total_lines"] += 1

                    try:
                        data = json.loads(line.strip())

                        # Count message types
                        role = data.get("role") or data.get("message", {}).get("role")
                        if role == "user":
                            stats["user_messages"] += 1
                        elif role == "assistant":
                            stats["assistant_messages"] += 1

                        # Analyze tool uses
                        content = data.get("content") or data.get("message", {}).get(
                            "content", []
                        )
                        for item in content:
                            if not isinstance(item, dict):
                                continue

                            if item.get("type") == "tool_use":
                                tool_name = item.get("name", "unknown")
                                stats["tool_uses"][tool_name] += 1

                                # Track file operations
                                if tool_name == "Write":
                                    stats["write_calls"] += 1
                                    file_path = item.get("input", {}).get(
                                        "file_path", ""
                                    )
                                    if file_path:
                                        stats["file_operations"].append(
                                            ("write", file_path)
                                        )
                                elif tool_name == "Edit":
                                    stats["edit_calls"] += 1
                                    file_path = item.get("input", {}).get(
                                        "file_path", ""
                                    )
                                    if file_path:
                                        stats["file_operations"].append(
                                            ("edit", file_path)
                                        )
                                elif tool_name == "Read":
                                    stats["read_calls"] += 1
                                elif tool_name == "Bash":
                                    stats["bash_calls"] += 1

                    except json.JSONDecodeError:
                        continue

        except Exception as e:
            print(f"Error analyzing {session_file}: {e}", file=sys.stderr)

        # Convert defaultdict to regular dict
        stats["tool_uses"] = dict(stats["tool_uses"])

        return stats

    def _extract_text_content(self, data: Dict[str, Any]) -> str:
        """Extract all text content from a message."""
        text_parts = []

        # Get content from either location
        content = data.get("content") or data.get("message", {}).get("content", [])

        if isinstance(content, str):
            text_parts.append(content)
        elif isinstance(content, list):
            for item in content:
                if isinstance(item, dict):
                    if item.get("type") == "text":
                        text_parts.append(item.get("text", ""))
                    # Also check tool inputs for file paths etc
                    elif item.get("type") == "tool_use":
                        tool_input = item.get("input", {})
                        if isinstance(tool_input, dict):
                            # Add file paths from tool inputs
                            if "file_path" in tool_input:
                                text_parts.append(tool_input["file_path"])
                            # Add content from Write calls
                            if "content" in tool_input:
                                text_parts.append(tool_input["content"])

        return " ".join(text_parts)


def main():
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(
        description="Analyze Claude Code session history files"
    )

    subparsers = parser.add_subparsers(dest="command", help="Command to run")

    # List sessions command
    list_parser = subparsers.add_parser("list", help="List all sessions for a project")
    list_parser.add_argument("project_path", help="Project path")
    list_parser.add_argument(
        "--limit", type=int, default=10, help="Max sessions to show (default: 10)"
    )

    # Search command
    search_parser = subparsers.add_parser("search", help="Search sessions for keywords")
    search_parser.add_argument("project_path", help="Project path")
    search_parser.add_argument(
        "keywords", nargs="+", help="Keywords to search for"
    )
    search_parser.add_argument(
        "--case-sensitive", action="store_true", help="Case-sensitive search"
    )

    # Stats command
    stats_parser = subparsers.add_parser("stats", help="Get session statistics")
    stats_parser.add_argument("session_file", type=Path, help="Session file path")
    stats_parser.add_argument(
        "--show-files", action="store_true", help="Show file operations"
    )

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(1)

    analyzer = SessionAnalyzer()

    if args.command == "list":
        sessions = analyzer.find_project_sessions(args.project_path)
        if not sessions:
            print(f"No sessions found for project: {args.project_path}")
            sys.exit(1)

        print(f"Found {len(sessions)} session(s) for {args.project_path}\n")
        print(f"Showing {min(args.limit, len(sessions))} most recent:\n")

        for i, session in enumerate(sessions[: args.limit], 1):
            mtime = datetime.fromtimestamp(session.stat().st_mtime)
            size_kb = session.stat().st_size / 1024
            print(f"{i}. {session.name}")
            print(f"   Modified: {mtime.strftime('%Y-%m-%d %H:%M:%S')}")
            print(f"   Size: {size_kb:.1f} KB")
            print(f"   Path: {session}")
            print()

    elif args.command == "search":
        sessions = analyzer.find_project_sessions(args.project_path)
        if not sessions:
            print(f"No sessions found for project: {args.project_path}")
            sys.exit(1)

        print(f"Searching {len(sessions)} session(s) for: {', '.join(args.keywords)}\n")

        matches = analyzer.search_sessions(
            sessions, args.keywords, args.case_sensitive
        )

        if not matches:
            print("No matches found.")
            sys.exit(0)

        # Sort by total mentions
        sorted_matches = sorted(
            matches.items(), key=lambda x: x[1]["total_mentions"], reverse=True
        )

        print(f"Found {len(matches)} session(s) with matches:\n")

        for session, info in sorted_matches:
            mtime = datetime.fromtimestamp(info["modified_time"])
            print(f"📄 {session.name}")
            print(f"   Date: {mtime.strftime('%Y-%m-%d %H:%M')}")
            print(f"   Total mentions: {info['total_mentions']}")
            print(f"   Keywords: {', '.join(f'{k}({v})' for k, v in info['keyword_counts'].items())}")
            print(f"   Path: {session}")
            print()

    elif args.command == "stats":
        if not args.session_file.exists():
            print(f"Error: Session file not found: {args.session_file}")
            sys.exit(1)

        print(f"Analyzing session: {args.session_file}\n")

        stats = analyzer.get_session_stats(args.session_file)

        print("=" * 60)
        print("Session Statistics")
        print("=" * 60)
        print(f"\nMessages:")
        print(f"  Total lines: {stats['total_lines']:,}")
        print(f"  User messages: {stats['user_messages']}")
        print(f"  Assistant messages: {stats['assistant_messages']}")

        print(f"\nTool Usage:")
        print(f"  Write calls: {stats['write_calls']}")
        print(f"  Edit calls: {stats['edit_calls']}")
        print(f"  Read calls: {stats['read_calls']}")
        print(f"  Bash calls: {stats['bash_calls']}")

        if stats["tool_uses"]:
            print(f"\n  All tools:")
            for tool, count in sorted(
                stats["tool_uses"].items(), key=lambda x: x[1], reverse=True
            ):
                print(f"    {tool}: {count}")

        if args.show_files and stats["file_operations"]:
            print(f"\nFile Operations ({len(stats['file_operations'])}):")
            # Group by file
            files = defaultdict(list)
            for op, path in stats["file_operations"]:
                files[path].append(op)

            # Limit to 20 files to prevent terminal flooding on large sessions
            for file_path, ops in list(files.items())[:20]:
                filename = Path(file_path).name
                op_summary = ", ".join(
                    f"{op}({ops.count(op)})" for op in set(ops)
                )
                print(f"  {filename}")
                print(f"    Operations: {op_summary}")
                print(f"    Path: {file_path}")

        print()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Recover content from Claude Code history session files.

This script extracts Write tool calls, Edit operations, and text content
from Claude Code's JSONL session history files.
"""

import json
import sys
import os
from pathlib import Path
from typing import Dict, List, Any, Optional
from datetime import datetime


class SessionContentRecovery:
    """Extract and recover content from Claude Code session files."""

    def __init__(self, session_file: Path, output_dir: Optional[Path] = None):
        self.session_file = Path(session_file)
        self.output_dir = output_dir or Path.cwd() / "recovered_content"
        self.output_dir.mkdir(exist_ok=True)

        # Statistics
        self.stats = {
            "total_lines": 0,
            "write_calls": 0,
            "edit_calls": 0,
            "text_mentions": 0,
            "files_recovered": 0,
        }

    def extract_write_calls(self) -> List[Dict[str, Any]]:
        """Extract all Write tool calls from session."""
        write_calls = []

        with open(self.session_file, "r") as f:
            for line_num, line in enumerate(f, 1):
                self.stats["total_lines"] += 1

                try:
                    data = json.loads(line.strip())

                    # Check both direct role and nested message.role
                    role = data.get("role") or data.get("message", {}).get("role")
                    if role != "assistant":
                        continue

                    # Get content from either location
                    content = data.get("content") or data.get("message", {}).get(
                        "content", []
                    )

                    for item in content:
                        if not isinstance(item, dict):
                            continue

                        # Look for Write tool calls
                        if item.get("type") == "tool_use" and item.get("name") == "Write":
                            write_input = item.get("input", {})
                            write_calls.append(
                                {
                                    "line": line_num,
                                    "file_path": write_input.get("file_path", ""),
                                    "content": write_input.get("content", ""),
                                    "timestamp": data.get("timestamp", ""),
                                }
                            )
                            self.stats["write_calls"] += 1

                except json.JSONDecodeError:
                    continue
                except Exception as e:
                    print(f"Warning: Error processing line {line_num}: {e}", file=sys.stderr)
                    continue

        return write_calls

    def extract_edit_calls(self) -> List[Dict[str, Any]]:
        """Extract all Edit tool calls from session."""
        edit_calls = []

        with open(self.session_file, "r") as f:
            for line_num, line in enumerate(f, 1):
                try:
                    data = json.loads(line.strip())

                    role = data.get("role") or data.get("message", {}).get("role")
                    if role != "assistant":
                        continue

                    content = data.get("content") or data.get("message", {}).get(
                        "content", []
                    )

                    for item in content:
                        if not isinstance(item, dict):
                            continue

                        if item.get("type") == "tool_use" and item.get("name") == "Edit":
                            edit_input = item.get("input", {})
                            edit_calls.append(
                                {
                                    "line": line_num,
                                    "file_path": edit_input.get("file_path", ""),
                                    "old_string": edit_input.get("old_string", ""),
                                    "new_string": edit_input.get("new_string", ""),
                                    "timestamp": data.get("timestamp", ""),
                                }
                            )
                            self.stats["edit_calls"] += 1

                except Exception:
                    continue

        return edit_calls

    def save_recovered_files(
        self, write_calls: List[Dict[str, Any]], keywords: Optional[List[str]] = None
    ) -> List[Dict[str, Any]]:
        """
        Save recovered files to disk, preserving original directory structure.

        Args:
            write_calls: List of Write tool calls
            keywords: Optional keywords to filter files (matches any keyword in file path)

        Returns:
            List of saved file metadata
        """
        saved = []

        # Filter by keywords if provided
        if keywords:
            write_calls = [
                call
                for call in write_calls
                if any(kw.lower() in call["file_path"].lower() for kw in keywords)
            ]

        # Deduplicate: keep latest version of each file
        files_by_path = {}
        for call in write_calls:
            file_path = call["file_path"]
            if not file_path:
                continue

            # Keep latest version (assuming chronological order in session)
            files_by_path[file_path] = call

        # Save files
        for file_path, call in files_by_path.items():
            try:
                if not file_path:
                    continue

                # Preserve original directory structure
                # Convert absolute path to relative path within output directory
                original_path = Path(file_path)

                # Handle absolute paths: extract meaningful relative path
                # e.g., /Users/username/project/src/file.py -> src/file.py
                # e.g., /home/user/workspace/project/lib/module.py -> lib/module.py
                path_parts = original_path.parts
                if len(path_parts) > 1 and path_parts[0] == "/":
                    # For absolute paths, try to find a project-like directory
                    # Skip leading /, Users/username, home/username patterns
                    start_idx = 1  # Skip leading "/"
                    if len(path_parts) > 2 and path_parts[1].lower() in ("users", "home", "user"):
                        start_idx = 3  # Skip /Users/username or /home/user
                    relative_parts = path_parts[start_idx:]
                else:
                    relative_parts = path_parts

                # Construct output path preserving structure
                if relative_parts:
                    output_file = self.output_dir.joinpath(*relative_parts)
                else:
                    # Fallback to filename only if path is too shallow
                    output_file = self.output_dir / original_path.name

                # Create parent directories
                output_file.parent.mkdir(parents=True, exist_ok=True)

                with open(output_file, "w") as f:
                    f.write(call["content"])

                saved.append(
                    {
                        "file": output_file.name,
                        "original_path": file_path,
                        "size": len(call["content"]),
                        "lines": call["content"].count("\n") + 1,
                        "timestamp": call.get("timestamp", "unknown"),
                        "output_path": str(output_file),
                    }
                )

                self.stats["files_recovered"] += 1

            except Exception as e:
                print(f"Warning: Failed to save {file_path}: {e}", file=sys.stderr)
                continue

        return saved

    def generate_report(self, saved_files: List[Dict[str, Any]]) -> str:
        """Generate recovery report."""
        report_lines = [
            "=" * 60,
            "Claude Code Session Content Recovery Report",
            "=" * 60,
            "",
            f"Session file: {self.session_file}",
            f"Output directory: {self.output_dir}",
            "",
            "Statistics:",
            f"  Total lines processed: {self.stats['total_lines']:,}",
            f"  Write tool calls found: {self.stats['write_calls']}",
            f"  Edit tool calls found: {self.stats['edit_calls']}",
            f"  Files recovered: {self.stats['files_recovered']}",
            "",
        ]

        if saved_files:
            report_lines.extend(
                [
                    "Recovered Files:",
                    "",
                ]
            )

            for item in saved_files:
                report_lines.extend(
                    [
                        f"✅ {item['file']}",
                        f"   Original: {item['original_path']}",
                        f"   Size: {item['size']:,} characters",
                        f"   Lines: {item['lines']:,}",
                        f"   Saved to: {item['output_path']}",
                        "",
                    ]
                )
        else:
            report_lines.append("No files recovered (no matches or no Write calls found)")
            report_lines.append("")

        report_lines.extend(["=" * 60, ""])

        return "\n".join(report_lines)


def main():
    """Main entry point."""
    import argparse

    parser = argparse.ArgumentParser(
        description="Recover content from Claude Code session history files"
    )
    parser.add_argument(
        "session_file",
        type=Path,
        help="Path to Claude Code session JSONL file",
    )
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        help="Output directory (default: ./recovered_content)",
    )
    parser.add_argument(
        "-k",
        "--keywords",
        nargs="+",
        help="Filter files by keywords (matches any keyword in file path)",
    )
    parser.add_argument(
        "--show-edits",
        action="store_true",
        help="Also show Edit operations (not saved, just listed)",
    )

    args = parser.parse_args()

    # Validate session file exists
    if not args.session_file.exists():
        print(f"Error: Session file not found: {args.session_file}", file=sys.stderr)
        sys.exit(1)

    # Create recovery instance
    recovery = SessionContentRecovery(args.session_file, args.output)

    print(f"🔍 Analyzing session: {args.session_file}")
    print(f"📂 Output directory: {recovery.output_dir}\n")

    # Extract Write calls
    print("1️⃣ Extracting Write tool calls...")
    write_calls = recovery.extract_write_calls()
    print(f"   Found {len(write_calls)} Write calls\n")

    # Save files
    print("2️⃣ Saving recovered files...")
    if args.keywords:
        print(f"   Filtering by keywords: {', '.join(args.keywords)}")
    saved = recovery.save_recovered_files(write_calls, args.keywords)
    print(f"   Saved {len(saved)} files\n")

    # Optionally show edits
    if args.show_edits:
        print("3️⃣ Extracting Edit tool calls...")
        edit_calls = recovery.extract_edit_calls()
        print(f"   Found {len(edit_calls)} Edit calls")
        if edit_calls:
            print("\n   Recent edits:")
            for edit in edit_calls[-5:]:  # Show last 5
                print(f"   - {Path(edit['file_path']).name} (line {edit['line']})")
        print()

    # Generate and print report
    report = recovery.generate_report(saved)
    print(report)

    # Save report
    report_file = recovery.output_dir / "recovery_report.txt"
    with open(report_file, "w") as f:
        f.write(report)
    print(f"📄 Report saved to: {report_file}\n")


if __name__ == "__main__":
    main()