cursor_chat_history_fix.py

#!/usr/bin/env python3
"""
Cursor Chat History Migrator
===========================
Restores missing chat history after moving projects or upgrading hard drives.
Cursor identifies workspaces by their absolute path; this script maps old 
paths to new ones and copies the 'state.vscdb' database.
"""
import os
import json
import shutil
from pathlib import Path
import sys
from datetime import datetime
import argparse
import platform

# Load .env file manually to avoid external dependencies
def load_env():
    env_path = Path(__file__).parent / ".env"
    if env_path.exists():
        with open(env_path, 'r') as f:
            for line in f:
                if line.strip() and not line.startswith("#"):
                    key, value = line.strip().split("=", 1)
                    os.environ[key] = value

load_env()

# Defaults (Can be overridden with arguments or .env)
DEFAULT_OLD_PREFIX = os.getenv("CURSOR_OLD_PREFIX", "file:///old/path/prefix")
DEFAULT_NEW_PREFIX = os.getenv("CURSOR_NEW_PREFIX", "file:///new/path/prefix")

def get_default_storage_dir():
    """Detects Cursor workspaceStorage path based on OS."""
    system = platform.system()
    home = Path.home()
    
    if system == "Windows":
        return Path(os.environ["APPDATA"]) / "Cursor" / "User" / "workspaceStorage"
    elif system == "Darwin":  # macOS
        return home / "Library" / "Application Support" / "Cursor" / "User" / "workspaceStorage"
    else:  # Linux
        return home / ".config" / "Cursor" / "User" / "workspaceStorage"

def get_workspace_info(folder_path):
    """Reads the workspace.json file to get the original project URI."""
    json_path = folder_path / "workspace.json"
    if not json_path.exists():
        return None
    try:
        with open(json_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
            return data.get("folder") or data.get("workspace")
    except Exception:
        return None

def get_db_size(folder_path):
    """Returns formatted size of the state.vscdb file."""
    db_path = folder_path / "state.vscdb"
    if db_path.exists():
        size_mb = db_path.stat().st_size / (1024 * 1024)
        return f"{size_mb:.2f}MB"
    return "No DB"

def check_cursor_running():
    """Checks if Cursor is currently running to prevent database corruption."""
    # Simple check for Linux/Mac
    if platform.system() != "Windows":
        is_running = os.popen("pgrep -i cursor").read().strip()
        if is_running:
            print("!! WARNING: Cursor is currently running.")
            print("!! Close Cursor before applying changes to avoid data loss.")
            confirm = input("Continue anyway? (y/N): ")
            if confirm.lower() != 'y':
                sys.exit(0)
    # Windows check could be added here with tasklist

def list_workspaces(storage_dir):
    """Lists all detected workspaces with their details."""
    print(f"\n{'ID':<34} | {'Size':<8} | {'Last Modified':<19} | {'Path'}")
    print("-" * 100)
    
    entries = []
    if not storage_dir.exists():
        print(f"Error: Directory {storage_dir} not found.")
        return

    for folder_name in os.listdir(storage_dir):
        folder_path = storage_dir / folder_name
        if not folder_path.is_dir():
            continue
        
        path_uri = get_workspace_info(folder_path)
        size = get_db_size(folder_path)
        mtime = datetime.fromtimestamp(folder_path.stat().st_mtime).strftime('%Y-%m-%d %H:%M:%S')
        entries.append((folder_name, size, mtime, path_uri or "Unknown"))
    
    # Sort by mtime (newest first)
    entries.sort(key=lambda x: x[2], reverse=True)
    
    for eid, size, mtime, path in entries:
        print(f"{eid:<34} | {size:<8} | {mtime:<19} | {path}")

def migrate_workspace(src, dst):
    """Copies the database from source to destination with backup."""
    backup_path = dst / "state.vscdb.backup_pre_migration"
    dst_db = dst / "state.vscdb"
    src_db = src / "state.vscdb"
    
    # Backup existing new DB if it exists
    if dst_db.exists() and not backup_path.exists():
        shutil.copy2(dst_db, backup_path)
    
    if src_db.exists():
        shutil.copy2(src_db, dst_db)
        print(f"  -> Successfully migrated database ({get_db_size(src)})")
        return True
    else:
        print("  -> Source DB missing!")
        return False

def main():
    parser = argparse.ArgumentParser(description="Restore Cursor chat history after moving projects.")
    
    # Action groups
    group = parser.add_mutually_exclusive_group()
    group.add_argument("--apply", action="store_true", help="Perform the migration (defaults to dry-run)")
    group.add_argument("--list", action="store_true", help="List all detected workspaces")
    group.add_argument("--manual", metavar="OLD_ID:NEW_ID", help="Manually migrate from OLD_ID to NEW_ID")
    
    # Configuration arguments
    parser.add_argument("--old-prefix", default=DEFAULT_OLD_PREFIX, help=f"Old path prefix to match (default: {DEFAULT_OLD_PREFIX})")
    parser.add_argument("--new-prefix", default=DEFAULT_NEW_PREFIX, help=f"New path prefix to replace with (default: {DEFAULT_NEW_PREFIX})")
    parser.add_argument("--storage-path", type=Path, help="Override Cursor workspace storage path")

    args = parser.parse_args()
    
    storage_dir = args.storage_path or get_default_storage_dir()

    if args.list:
        list_workspaces(storage_dir)
        return

    if args.manual:
        try:
            old_id, new_id = args.manual.split(":")
            src = storage_dir / old_id
            dst = storage_dir / new_id
            
            if not src.exists():
                print(f"Error: Source ID '{old_id}' not found.")
                return
            if not dst.exists():
                print(f"Error: Destination ID '{new_id}' not found. Open the project in Cursor first.")
                return
                
            check_cursor_running()
            print(f"Manually migrating {old_id} -> {new_id}...")
            migrate_workspace(src, dst)
            return
        except ValueError:
            print("Error: Invalid format. Use --manual OLD_ID:NEW_ID")
            return

    # Auto-Migration Logic
    dry_run = not args.apply
    if not dry_run:
        check_cursor_running()

    print(f"Scanning {storage_dir}...")
    
    workspaces = {} 
    if not storage_dir.exists():
        print(f"Error: Storage directory {storage_dir} not found.")
        return

    for folder_name in os.listdir(storage_dir):
        folder_path = storage_dir / folder_name
        if not folder_path.is_dir():
            continue
        path_uri = get_workspace_info(folder_path)
        if path_uri:
            if path_uri not in workspaces:
                workspaces[path_uri] = []
            workspaces[path_uri].append(folder_name)

    print(f"Found {len(workspaces)} unique workspace paths.")

    matches = []
    pending_open = []
    
    old_workspaces_uris = [uri for uri in workspaces.keys() if uri and uri.startswith(args.old_prefix)]
    
    print(f"Found {len(old_workspaces_uris)} historical workspace paths matching '{args.old_prefix}'.")

    for old_uri in old_workspaces_uris:
        # Find best old ID (largest DB)
        old_ids = workspaces[old_uri]
        old_id = max(old_ids, key=lambda fid: (storage_dir / fid / "state.vscdb").stat().st_size if (storage_dir / fid / "state.vscdb").exists() else 0)

        # Determine new URI
        new_uri = old_uri.replace(args.old_prefix, args.new_prefix, 1)
        
        if new_uri in workspaces:
            # Find best new ID (most recently used)
            new_ids = workspaces[new_uri]
            new_id = max(new_ids, key=lambda fid: (storage_dir / fid).stat().st_mtime)
            
            if old_id != new_id:
                matches.append((old_id, new_id, old_uri, new_uri))
        else:
            pending_open.append((old_id, old_uri, new_uri))

    print(f"\nAnalysis Results:")
    print(f"----------------")
    print(f"Ready to Migrate: {len(matches)}")
    print(f"Pending (Open project in Cursor first): {len(pending_open)}")
    
    if len(matches) > 0:
        print("\nMatches found (Old -> New):")
        for old_id, new_id, old_uri, _ in matches:
            old_size = get_db_size(storage_dir / old_id)
            new_size = get_db_size(storage_dir / new_id)
            print(f"[{old_id[:8]} ({old_size}) -> {new_id[:8]} ({new_size})] {os.path.basename(old_uri)}")

    if dry_run:
        print("\n[DRY RUN] No changes made. Run with --apply to execute.")
    else:
        print("\n[MIGRATION] Applying changes...")
        for old_id, new_id, old_uri, _ in matches:
            src_folder = storage_dir / old_id
            dst_folder = storage_dir / new_id
            print(f"Migrating {os.path.basename(old_uri)}...")
            migrate_workspace(src_folder, dst_folder)

if __name__ == "__main__":
    main()