Manage Claude Code session transcripts across project dirs (list/move/copy)

cc-projects.py

#!/usr/bin/env python3
"""Manage Claude Code session transcripts across project dirs.

Sessions live at ~/.claude/projects/<encoded-cwd>/<uuid>.jsonl where encoded-cwd
is the absolute cwd with non-alphanumeric chars (except '-') replaced by '-'.

Subcommands:
  list                List absolute cwd of every known project dir.
  move                Move sessions from a source cwd into a destination cwd.
  copy                Copy sessions from a source cwd into a destination cwd.
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Literal

__all__ = ["encode", "main"]

PROJECTS_DIR = Path.home() / ".claude" / "projects"
_ENCODE_RE = re.compile(r"[^A-Za-z0-9-]")


def encode(cwd: str) -> str:
    # Claude Code replaces any non-alphanumeric (except '-') with '-'.
    return _ENCODE_RE.sub("-", cwd)


def _peek_session(path: Path) -> tuple[str | None, str | None]:
    """Returns (cwd, summary) by scanning a single jsonl file. Stops early."""
    cwd: str | None = None
    summary: str | None = None
    try:
        fh = path.open("r", encoding="utf-8")
    except OSError:
        return None, None
    with fh:
        for line in fh:
            s = line.strip()
            if not s:
                continue
            try:
                obj: Any = json.loads(s)
            except json.JSONDecodeError:
                continue
            if not isinstance(obj, dict):
                continue
            if cwd is None:
                v = obj.get("cwd")
                if isinstance(v, str):
                    cwd = v
            if summary is None and obj.get("type") == "summary":
                v = obj.get("summary")
                if isinstance(v, str):
                    summary = v
            if cwd is not None and summary is not None:
                break
    return cwd, summary


def rewrite(src: Path, dst: Path, source_cwds: set[str], dest_cwd: str) -> int:
    """Stream src -> dst, rewriting cwd field. Returns count of rewritten lines."""
    rewrites = 0
    with src.open("r", encoding="utf-8") as fin, dst.open("w", encoding="utf-8") as fout:
        for line in fin:
            stripped = line.strip()
            if not stripped:
                fout.write(line)
                continue
            try:
                obj: Any = json.loads(stripped)
            except json.JSONDecodeError:
                fout.write(line)
                continue
            if isinstance(obj, dict) and obj.get("cwd") in source_cwds:
                obj["cwd"] = dest_cwd
                fout.write(json.dumps(obj, ensure_ascii=False) + "\n")
                rewrites += 1
            else:
                fout.write(line)
    return rewrites


@dataclass(frozen=True)
class SessionInfo:
    uuid: str
    file: Path
    last_active: float   # epoch seconds (mtime)
    first_active: float  # epoch seconds (ctime)
    size_bytes: int
    summary: str | None


@dataclass(frozen=True)
class ProjectInfo:
    path: str           # cwd
    project_path: Path  # filesystem dir under ~/.claude/projects
    sessions: list[SessionInfo]

    @property
    def last_active(self) -> float:
        return max(s.last_active for s in self.sessions)


def _gather_projects() -> tuple[list[ProjectInfo], list[Path]]:
    """Returns (known projects, project dirs we couldn't identify)."""
    projects: list[ProjectInfo] = []
    unknown: list[Path] = []
    for pdir in sorted(PROJECTS_DIR.iterdir()):
        if not pdir.is_dir():
            continue
        jsonls = sorted(pdir.glob("*.jsonl"))
        if not jsonls:
            unknown.append(pdir)
            continue
        sessions: list[SessionInfo] = []
        project_cwd: str | None = None
        for f in jsonls:
            cwd, summary = _peek_session(f)
            if project_cwd is None and cwd is not None:
                project_cwd = cwd
            st = f.stat()
            sessions.append(SessionInfo(
                uuid=f.stem,
                file=f,
                last_active=st.st_mtime,
                first_active=st.st_ctime,
                size_bytes=st.st_size,
                summary=summary,
            ))
        if project_cwd is None:
            unknown.append(pdir)
            continue
        sessions.sort(key=lambda s: s.last_active, reverse=True)
        projects.append(ProjectInfo(path=project_cwd, project_path=pdir, sessions=sessions))
    return projects, unknown


def cmd_list(fmt: Literal["table", "json"]) -> int:
    if not PROJECTS_DIR.is_dir():
        print(f"error: projects dir not found: {PROJECTS_DIR}", file=sys.stderr)
        return 1
    projects, unknown = _gather_projects()
    projects.sort(key=lambda p: p.last_active, reverse=True)

    if fmt == "json":
        def iso(ts: float) -> str:
            return datetime.fromtimestamp(ts).astimezone().isoformat()
        out = [
            {
                "path": p.path,
                "project_path": str(p.project_path),
                "session_count": len(p.sessions),
                "last_active": iso(p.last_active),
                "sessions": [
                    {
                        "uuid": s.uuid,
                        "file": str(s.file),
                        "summary": s.summary,
                        "last_active": iso(s.last_active),
                        "first_active": iso(s.first_active),
                        "size_bytes": s.size_bytes,
                    }
                    for s in p.sessions
                ],
            }
            for p in projects
        ]
        print(json.dumps(out, indent=2))
    else:
        rows: list[tuple[str, str, str]] = [
            (
                p.path,
                str(len(p.sessions)),
                datetime.fromtimestamp(p.last_active).strftime("%Y-%m-%d %H:%M:%S"),
            )
            for p in projects
        ]
        headers = ("PATH", "SESSIONS", "LAST ACTIVE")
        widths = [
            max(len(headers[i]), *(len(r[i]) for r in rows)) if rows else len(headers[i])
            for i in range(3)
        ]
        # PATH + LAST ACTIVE left-aligned, SESSIONS right-aligned.
        def fmt_row(r: tuple[str, str, str]) -> str:
            return "  ".join([r[0].ljust(widths[0]), r[1].rjust(widths[1]), r[2].ljust(widths[2])])
        print(fmt_row(headers))
        for r in rows:
            print(fmt_row(r))

    for pdir in unknown:
        print(f"warn: skipped {pdir} (no identifiable session)", file=sys.stderr)
    return 0


def cmd_transfer(source_cwd: str, dest_cwd: str, *, copy: bool, dry_run: bool) -> int:
    if source_cwd == dest_cwd:
        print(f"error: source and destination are the same: {dest_cwd!r}", file=sys.stderr)
        return 2

    src_dir = PROJECTS_DIR / encode(source_cwd)
    dest_dir = PROJECTS_DIR / encode(dest_cwd)

    if not src_dir.is_dir():
        print(f"error: source dir not found: {src_dir}", file=sys.stderr)
        return 2

    existing: dict[str, Path] = {}
    if dest_dir.is_dir():
        for f in dest_dir.glob("*.jsonl"):
            existing[f.name] = f

    plan: list[tuple[Path, Path]] = []
    collisions: list[str] = []
    for f in sorted(src_dir.glob("*.jsonl")):
        if f.name in existing:
            collisions.append(f"{f.name}: {f} vs {existing[f.name]}")
        else:
            plan.append((f, dest_dir / f.name))

    if collisions:
        print("error: UUID collisions (aborting):", file=sys.stderr)
        for c in collisions:
            print(f"  {c}", file=sys.stderr)
        return 3

    if not plan:
        print("nothing to do")
        return 0

    op_word = "copy" if copy else "move"
    print(f"dest: {dest_dir}")
    print(f"source: {src_dir}, files: {len(plan)}")

    if dry_run:
        print(f"[dry-run] would {op_word} {len(plan)} file(s)")
        return 0

    dest_dir.mkdir(parents=True, exist_ok=True)
    source_set = {source_cwd}
    total_rewrites = 0
    for src, dst in plan:
        total_rewrites += rewrite(src, dst, source_set, dest_cwd)
        if not copy:
            src.unlink()

    removed_src_dir = False
    if not copy and not any(src_dir.iterdir()):
        src_dir.rmdir()
        removed_src_dir = True

    msg = f"{op_word}d {len(plan)} file(s), rewrote {total_rewrites} cwd line(s)"
    if not copy:
        msg += f", removed source dir: {removed_src_dir}"
    print(msg)
    return 0


def _build_parser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    sub = ap.add_subparsers(dest="cmd", required=True, metavar="{list,move,copy}")

    list_p = sub.add_parser("list", help="List known project cwds")
    list_p.add_argument(
        "--format", choices=["table", "json"], default="table",
        help="Output format (default: table)",
    )

    transfer_args = argparse.ArgumentParser(add_help=False)
    transfer_args.add_argument("source", metavar="SOURCE", help="Source cwd (absolute)")
    transfer_args.add_argument("destination", metavar="DESTINATION", help="Destination cwd (absolute)")
    transfer_args.add_argument("--dry-run", action="store_true")

    sub.add_parser("move", parents=[transfer_args], help="Move sessions: SOURCE DESTINATION")
    sub.add_parser("copy", parents=[transfer_args], help="Copy sessions: SOURCE DESTINATION")

    return ap


def main() -> int:
    args = _build_parser().parse_args()
    if args.cmd == "list":
        return cmd_list(args.format)
    return cmd_transfer(
        args.source, args.destination,
        copy=args.cmd == "copy", dry_run=args.dry_run,
    )


if __name__ == "__main__":
    sys.exit(main())