Time Lens

#!/usr/bin/env python3
"""
Count Claude Code user prompts for a specific project, with per-day breakdown.

Data sources:
  Primary:   ~/.claude/history.jsonl  — one entry per submitted prompt, has "project" (abs path)
             and "timestamp" (ms epoch).
  Fallback:  ~/.claude/projects/<encoded>/*.jsonl  — session files, use for sessions that
             predate history.jsonl. Filter type="user", exclude tool_result content.

Usage:
  python3 claude_messages.py --project-path /abs/path/to/repo
  python3 claude_messages.py --filter project-name   [substring match on dir name]
  python3 claude_messages.py --project-path /path --projects-dir ~/.claude/projects

Output JSON:
  {
    "project_path": "...",
    "total_user_messages": N,
    "daily": {"YYYY-MM-DD": count, ...},
    "sources": {"history": N, "session_files": N}
  }
"""

import os
import json
import argparse
import glob
from datetime import datetime, timezone
from collections import defaultdict


def path_to_encoded(abs_path):
    """Convert absolute path to Claude's encoded project dir name: /a/b → -a-b"""
    return abs_path.replace("/", "-")


def find_project_dirs(projects_dir, project_path=None, name_filter=None):
    """Return list of matching Claude project dirs."""
    projects_dir = os.path.expanduser(projects_dir)
    if not os.path.isdir(projects_dir):
        return []

    if project_path:
        encoded = path_to_encoded(project_path)
        candidate = os.path.join(projects_dir, encoded)
        return [candidate] if os.path.isdir(candidate) else []

    return [
        e.path for e in sorted(os.scandir(projects_dir), key=lambda e: e.name)
        if e.is_dir() and name_filter and name_filter.lower() in e.name.lower()
    ]


def is_real_prompt(obj):
    """Return True if this session entry is an actual user prompt (not a tool result)."""
    content = obj.get("message", {}).get("content", "")
    if isinstance(content, str):
        return True
    if isinstance(content, list) and content:
        # list[text] = prompt with pasted content; list[tool_result] = tool output
        return content[0].get("type") != "tool_result"
    return False


def collect_from_history(project_path):
    """
    Collect prompts from ~/.claude/history.jsonl filtered by project path.
    Returns: {date: count}, set of rounded timestamps (for dedup with session files).
    """
    daily = defaultdict(int)
    seen_ts = set()
    history_path = os.path.expanduser("~/.claude/history.jsonl")

    if not os.path.exists(history_path) or not project_path:
        return daily, seen_ts

    with open(history_path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue

            if obj.get("project") != project_path:
                continue

            ts_ms = obj.get("timestamp")
            if ts_ms is None:
                continue

            ts_float = ts_ms / 1000.0
            date = datetime.fromtimestamp(ts_float, tz=timezone.utc).strftime("%Y-%m-%d")
            seen_ts.add(round(ts_float))
            daily[date] += 1

    return daily, seen_ts


def collect_from_session_files(project_dirs, project_path, seen_ts):
    """
    Collect actual user prompts from session .jsonl files.
    Skips entries already accounted for in history (dedup by rounded timestamp).
    Returns {date: count}.
    """
    daily = defaultdict(int)

    for proj_dir in project_dirs:
        for jsonl_file in glob.glob(os.path.join(proj_dir, "*.jsonl")):
            try:
                with open(jsonl_file, "r", encoding="utf-8") as f:
                    for line in f:
                        line = line.strip()
                        if not line:
                            continue
                        try:
                            obj = json.loads(line)
                        except json.JSONDecodeError:
                            continue

                        if obj.get("type") != "user":
                            continue
                        if project_path and obj.get("cwd") != project_path:
                            continue
                        if not is_real_prompt(obj):
                            continue

                        ts_str = obj.get("timestamp", "")
                        try:
                            dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
                            ts_rounded = round(dt.timestamp())
                            date = dt.strftime("%Y-%m-%d")
                        except Exception:
                            continue

                        if ts_rounded not in seen_ts:
                            seen_ts.add(ts_rounded)
                            daily[date] += 1
            except (IOError, PermissionError):
                pass

    return daily


def main():
    parser = argparse.ArgumentParser(description="Count Claude Code prompts per project by day")
    parser.add_argument("--project-path", help="Absolute path to the project repo (exact match)")
    parser.add_argument("--filter", help="Substring match on Claude project dir name")
    parser.add_argument("--projects-dir", default="~/.claude/projects")
    args = parser.parse_args()

    if not args.project_path and not args.filter:
        print(json.dumps({"error": "Provide --project-path or --filter"}))
        return

    project_dirs = find_project_dirs(args.projects_dir, args.project_path, args.filter)

    if not project_dirs:
        result = {
            "project_path": args.project_path or args.filter,
            "total_user_messages": 0,
            "daily": {},
            "timestamps": [],
            "note": "No matching Claude project directory found"
        }
        # Folder move detection: scan history.jsonl for same basename at different paths
        # (history.jsonl stores exact absolute paths — no lossy encoding)
        if args.project_path:
            target_name = os.path.basename(args.project_path)
            alternates = set()
            history_path = os.path.expanduser("~/.claude/history.jsonl")
            if os.path.exists(history_path):
                with open(history_path, "r", encoding="utf-8") as f:
                    for line in f:
                        try:
                            obj = json.loads(line.strip())
                            proj = obj.get("project", "")
                            if (proj and proj != args.project_path and
                                    os.path.basename(proj) == target_name):
                                alternates.add(proj)
                        except (json.JSONDecodeError, KeyError):
                            pass
            # Also check project dir names (session cwd fields)
            for entry in os.scandir(os.path.expanduser(args.projects_dir)):
                if not entry.is_dir():
                    continue
                for jf in glob.glob(os.path.join(entry.path, "*.jsonl")):
                    try:
                        with open(jf) as f:
                            obj = json.loads(f.readline().strip())
                            cwd = obj.get("cwd", "")
                            if (cwd and cwd != args.project_path and
                                    os.path.basename(cwd) == target_name):
                                alternates.add(cwd)
                    except Exception:
                        pass
                    break  # only need first line of first file per project
            if alternates:
                result["alternate_paths"] = sorted(alternates)
                result["note"] = (
                    f"No Claude history at {args.project_path}, "
                    f"but found history for '{target_name}' at "
                    f"{len(alternates)} other path(s). Project may have been moved."
                )
        print(json.dumps(result, indent=2))
        return

    # Primary: history.jsonl (one entry per prompt, most reliable)
    daily_history, seen_ts = collect_from_history(args.project_path)

    # Fallback: session files for prompts not captured in history
    daily_sessions = collect_from_session_files(project_dirs, args.project_path, seen_ts)

    # Merge
    all_dates = set(daily_history) | set(daily_sessions)
    daily_merged = {
        date: daily_history.get(date, 0) + daily_sessions.get(date, 0)
        for date in sorted(all_dates)
    }

    print(json.dumps({
        "project_path": args.project_path or args.filter,
        "project_dirs": project_dirs,
        "total_user_messages": sum(daily_merged.values()),
        "daily": daily_merged,
        "sources": {
            "history": sum(daily_history.values()),
            "session_files": sum(daily_sessions.values()),
        },
        # Raw sorted timestamps (Unix epoch floats) — use for merged session detection
        "timestamps": sorted(seen_ts),
    }, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Count Codex CLI user prompts for a specific project, with per-day breakdown.

Data source: ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl
  - session_meta entry: payload.cwd = project directory
  - event_msg entry: payload.type == "user_message" = actual user prompt

Usage:
  python3 codex_messages.py --project-path /abs/path/to/repo
  python3 codex_messages.py --filter project-name

Output JSON:
  {
    "project_path": "...",
    "total_user_messages": N,
    "daily": {"YYYY-MM-DD": count, ...},
    "timestamps": [epoch_float, ...],
    "sessions_found": N
  }
"""

import os
import json
import argparse
import glob
from datetime import datetime, timezone
from collections import defaultdict


def scan_sessions(sessions_dir, project_path=None, name_filter=None):
    """
    Scan Codex session files. Filter by cwd == project_path (exact) or
    basename match for name_filter (substring).
    Returns (daily counts, sorted timestamp list, session count).
    """
    sessions_dir = os.path.expanduser(sessions_dir)
    if not os.path.isdir(sessions_dir):
        return {}, [], 0

    daily = defaultdict(int)
    timestamps = []
    sessions_found = 0

    for jsonl_path in sorted(glob.glob(
        os.path.join(sessions_dir, "**", "*.jsonl"), recursive=True
    )):
        cwd = None
        session_prompts = []

        for line in open(jsonl_path, "r", encoding="utf-8"):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue

            # Extract cwd from session_meta (always first entry)
            if obj.get("type") == "session_meta":
                cwd = obj.get("payload", {}).get("cwd", "")
                continue

            # Check cwd match
            if cwd is None:
                continue
            if project_path and cwd != project_path:
                break  # wrong project, skip rest of file
            if name_filter and name_filter.lower() not in os.path.basename(cwd).lower():
                break

            # Count user prompts
            if (obj.get("type") == "event_msg" and
                    obj.get("payload", {}).get("type") == "user_message"):
                ts_str = obj.get("timestamp", "")
                try:
                    dt = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
                    ts_epoch = dt.timestamp()
                    date = dt.strftime("%Y-%m-%d")
                    session_prompts.append((date, ts_epoch))
                except Exception:
                    session_prompts.append((None, None))

        if session_prompts:
            sessions_found += 1
            for date, ts in session_prompts:
                if date:
                    daily[date] += 1
                if ts:
                    timestamps.append(ts)

    return dict(sorted(daily.items())), sorted(timestamps), sessions_found


def find_alternate_paths(sessions_dir, project_path):
    """
    When no sessions match project_path, look for sessions with the same
    basename (last path component) at different parent paths.
    Suggests the project may have been moved.
    """
    sessions_dir = os.path.expanduser(sessions_dir)
    if not os.path.isdir(sessions_dir):
        return []

    target_name = os.path.basename(project_path)
    alternates = set()

    for jsonl_path in glob.glob(
        os.path.join(sessions_dir, "**", "*.jsonl"), recursive=True
    ):
        for line in open(jsonl_path, "r", encoding="utf-8"):
            try:
                obj = json.loads(line.strip())
            except json.JSONDecodeError:
                continue
            if obj.get("type") == "session_meta":
                cwd = obj.get("payload", {}).get("cwd", "")
                if cwd and cwd != project_path and os.path.basename(cwd) == target_name:
                    alternates.add(cwd)
                break  # only read session_meta

    return sorted(alternates)


def main():
    parser = argparse.ArgumentParser(description="Count Codex user prompts per project by day")
    parser.add_argument("--project-path", help="Absolute path to the project repo (exact match)")
    parser.add_argument("--filter", help="Substring match on project basename")
    parser.add_argument("--sessions-dir", default="~/.codex/sessions")
    args = parser.parse_args()

    if not args.project_path and not args.filter:
        print(json.dumps({"error": "Provide --project-path or --filter"}))
        return

    daily, timestamps, sessions_found = scan_sessions(
        args.sessions_dir, args.project_path, args.filter
    )

    result = {
        "project_path": args.project_path or args.filter,
        "total_user_messages": sum(daily.values()),
        "daily": daily,
        "timestamps": timestamps,
        "sessions_found": sessions_found,
    }

    # Folder move detection
    if sum(daily.values()) == 0 and args.project_path:
        alternates = find_alternate_paths(args.sessions_dir, args.project_path)
        if alternates:
            result["alternate_paths"] = alternates
            result["note"] = (
                f"No Codex sessions found at {args.project_path}, "
                f"but found sessions for '{os.path.basename(args.project_path)}' "
                f"at {len(alternates)} other path(s). Project may have been moved."
            )

    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Count Cursor IDE user prompts for a specific project, with per-day breakdown.

Data sources (paths auto-detected per platform):
  Primary:   <Cursor User dir>/globalStorage/state.vscdb
             — cursorDiskKV table: composerData:{sessionId} (workspace URI) and
               bubbleId:{sessionId}:{messageId} (per-message timestamps)
  Fallback:  <Cursor User dir>/workspaceStorage/*/state.vscdb
             — ItemTable: composer.composerData (session-level timestamps only)
             — ItemTable: legacy chatdata keys (per-message timestamps)

  Cursor User dir:
    macOS:   ~/Library/Application Support/Cursor/User
    Windows: %APPDATA%/Cursor/User
    Linux:   ~/.config/Cursor/User

Usage:
  python3 cursor_messages.py --project-path /abs/path/to/repo
  python3 cursor_messages.py --filter project-name

Output JSON:
  {
    "project_path": "...",
    "total_user_messages": N,
    "daily": {"YYYY-MM-DD": count, ...},
    "timestamps": [epoch_float, ...],
    "sessions_found": N,
    "sources": {"global_storage": N, "workspace_storage": N}
  }
"""

import os
import sys
import json
import argparse
import sqlite3
from datetime import datetime, timezone
from collections import defaultdict
from urllib.parse import unquote, urlparse


def get_cursor_data_dir():
    """Return the Cursor User data directory for the current platform."""
    if sys.platform == "darwin":
        return os.path.expanduser("~/Library/Application Support/Cursor/User")
    elif sys.platform == "win32":
        return os.path.join(os.environ.get("APPDATA", ""), "Cursor", "User")
    else:  # Linux
        return os.path.expanduser("~/.config/Cursor/User")


def uri_to_path(uri):
    """Convert a file:// URI to an absolute path."""
    if not uri:
        return ""
    parsed = urlparse(uri)
    if parsed.scheme == "file":
        return unquote(parsed.path)
    return uri


def paths_match(path1, path2):
    """Check if two paths refer to the same directory."""
    return os.path.normpath(path1) == os.path.normpath(path2)


def scan_workspace_mappings(cursor_dir):
    """Scan workspaceStorage dirs → dict of workspace_id → project_path."""
    ws_dir = os.path.join(cursor_dir, "workspaceStorage")
    if not os.path.isdir(ws_dir):
        return {}

    mappings = {}
    for entry in os.scandir(ws_dir):
        if not entry.is_dir():
            continue
        ws_json = os.path.join(entry.path, "workspace.json")
        if not os.path.exists(ws_json):
            continue
        try:
            with open(ws_json, "r", encoding="utf-8") as f:
                data = json.load(f)
            folder = data.get("folder", "")
            if folder:
                mappings[entry.name] = uri_to_path(folder)
        except (json.JSONDecodeError, IOError):
            pass

    return mappings


def _extract_bubble_timestamp(bubble):
    """
    Extract a UTC epoch float from a Cursor bubble/message object.
    Tries fields in priority order (mirrors cursor-history's fallback chain).
    Returns epoch float or None.
    """
    # 1. createdAt ISO string (new format, >= Sept 2025)
    created_at = bubble.get("createdAt")
    if created_at and isinstance(created_at, str):
        try:
            dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
            return dt.timestamp()
        except (ValueError, TypeError):
            pass

    # 2. timingInfo fields (Unix ms)
    timing = bubble.get("timingInfo")
    if isinstance(timing, dict):
        for field in ("clientStartTime", "clientRpcSendTime",
                      "clientSettleTime", "clientEndTime"):
            val = timing.get(field)
            if val and isinstance(val, (int, float)) and val > 1_000_000_000_000:
                return val / 1000.0

    # 3. Plain timestamp field (legacy, ms epoch)
    ts = bubble.get("timestamp")
    if ts and isinstance(ts, (int, float)) and ts > 1_000_000_000_000:
        return ts / 1000.0

    return None


def collect_from_global_storage(cursor_dir, project_path=None, name_filter=None):
    """
    Collect user prompt timestamps from global storage's cursorDiskKV table.
    Returns (daily_counts, timestamps, sessions_found).
    """
    global_db = os.path.join(cursor_dir, "globalStorage", "state.vscdb")
    if not os.path.exists(global_db):
        return {}, [], 0

    daily = defaultdict(int)
    timestamps = []
    sessions_found = 0

    try:
        conn = sqlite3.connect(f"file:{global_db}?mode=ro", uri=True)
        cur = conn.cursor()

        # Check if cursorDiskKV table exists
        cur.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='cursorDiskKV'"
        )
        if not cur.fetchone():
            conn.close()
            return {}, [], 0

        # Get all composer sessions with their workspace URIs
        cur.execute(
            "SELECT key, value FROM cursorDiskKV WHERE key LIKE 'composerData:%'"
        )

        matching_session_ids = []
        for key, value in cur.fetchall():
            try:
                session_id = key.split(":", 1)[1]
                data = json.loads(value) if isinstance(value, str) else {}
                ws_uri = data.get("workspaceUri", "")
                ws_path = uri_to_path(ws_uri)

                if project_path:
                    if not ws_path or not paths_match(ws_path, project_path):
                        continue
                elif name_filter:
                    basename = os.path.basename(os.path.normpath(ws_path)) if ws_path else ""
                    if not basename or name_filter.lower() not in basename.lower():
                        continue

                matching_session_ids.append(session_id)
            except (json.JSONDecodeError, IndexError, AttributeError):
                continue

        # For each matching session, get user-message bubble data
        for session_id in matching_session_ids:
            session_ts = []
            cur.execute(
                "SELECT value FROM cursorDiskKV WHERE key LIKE ?",
                (f"bubbleId:{session_id}:%",),
            )

            for (value,) in cur.fetchall():
                try:
                    bubble = json.loads(value) if isinstance(value, str) else {}
                except json.JSONDecodeError:
                    continue

                # Only count user messages (type=1)
                if bubble.get("type") != 1:
                    continue

                ts_epoch = _extract_bubble_timestamp(bubble)
                if ts_epoch is not None:
                    session_ts.append(ts_epoch)

            if session_ts:
                sessions_found += 1
                for ts in session_ts:
                    date = datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%d")
                    daily[date] += 1
                    timestamps.append(ts)

        conn.close()
    except (sqlite3.Error, IOError):
        pass

    return dict(sorted(daily.items())), sorted(timestamps), sessions_found


def collect_from_workspace_storage(cursor_dir, project_path=None,
                                   name_filter=None, seen_ts=None):
    """
    Fallback: collect timestamps from per-workspace state.vscdb databases.
    Reads composer.composerData and legacy chatdata keys from ItemTable.
    Deduplicates against already-seen timestamps from global storage.
    Returns (daily_counts, timestamps, sessions_found).
    """
    if seen_ts is None:
        seen_ts = set()

    ws_mappings = scan_workspace_mappings(cursor_dir)
    if not ws_mappings:
        return {}, [], 0

    daily = defaultdict(int)
    timestamps = []
    sessions_found = 0

    for ws_id, ws_path in ws_mappings.items():
        if project_path and not paths_match(ws_path, project_path):
            continue
        if name_filter:
            basename = os.path.basename(os.path.normpath(ws_path))
            if name_filter.lower() not in basename.lower():
                continue

        db_path = os.path.join(cursor_dir, "workspaceStorage", ws_id, "state.vscdb")
        if not os.path.exists(db_path):
            continue

        try:
            conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
            cur = conn.cursor()

            # --- Composer data (session-level timestamps) ---
            cur.execute(
                "SELECT value FROM ItemTable WHERE key = 'composer.composerData'"
            )
            row = cur.fetchone()
            if row and row[0]:
                try:
                    data = json.loads(row[0])
                    for composer in data.get("allComposers", []):
                        created = composer.get("createdAt")
                        updated = composer.get("lastUpdatedAt")
                        for ts_ms in (created, updated):
                            if (ts_ms and isinstance(ts_ms, (int, float))
                                    and ts_ms > 1_000_000_000_000):
                                ts_epoch = ts_ms / 1000.0
                                ts_rounded = round(ts_epoch)
                                if ts_rounded not in seen_ts:
                                    seen_ts.add(ts_rounded)
                                    date = datetime.fromtimestamp(
                                        ts_epoch, tz=timezone.utc
                                    ).strftime("%Y-%m-%d")
                                    daily[date] += 1
                                    timestamps.append(ts_epoch)
                        if created or updated:
                            sessions_found += 1
                except json.JSONDecodeError:
                    pass

            # --- Legacy chat data keys (per-message timestamps) ---
            for key in (
                "workbench.panel.aichat.view.aichat.chatdata",
                "workbench.panel.chat.view.chat.chatdata",
            ):
                cur.execute("SELECT value FROM ItemTable WHERE key = ?", (key,))
                row = cur.fetchone()
                if not (row and row[0]):
                    continue
                try:
                    data = json.loads(row[0])
                    for session in data.get("chatSessions", data.get("tabs", [])):
                        has_msgs = False
                        for msg in session.get("messages", session.get("bubbles", [])):
                            if msg.get("role") != "user":
                                continue
                            ts_ms = msg.get("timestamp")
                            if (ts_ms and isinstance(ts_ms, (int, float))
                                    and ts_ms > 1_000_000_000_000):
                                ts_epoch = ts_ms / 1000.0
                                ts_rounded = round(ts_epoch)
                                if ts_rounded not in seen_ts:
                                    seen_ts.add(ts_rounded)
                                    date = datetime.fromtimestamp(
                                        ts_epoch, tz=timezone.utc
                                    ).strftime("%Y-%m-%d")
                                    daily[date] += 1
                                    timestamps.append(ts_epoch)
                                    has_msgs = True
                        if has_msgs:
                            sessions_found += 1
                except json.JSONDecodeError:
                    pass

            conn.close()
        except (sqlite3.Error, IOError):
            pass

    return dict(sorted(daily.items())), sorted(timestamps), sessions_found


def find_alternate_paths(cursor_dir, project_path):
    """
    When no sessions match project_path, look for workspaces with the same
    basename at different parent paths.
    """
    ws_mappings = scan_workspace_mappings(cursor_dir)
    target_name = os.path.basename(os.path.normpath(project_path))
    alternates = set()

    for _ws_id, ws_path in ws_mappings.items():
        if not ws_path:
            continue
        if (not paths_match(ws_path, project_path)
                and os.path.basename(os.path.normpath(ws_path)) == target_name):
            alternates.add(ws_path)

    # Also check global storage composerData for workspaceUri matches
    global_db = os.path.join(cursor_dir, "globalStorage", "state.vscdb")
    if os.path.exists(global_db):
        try:
            conn = sqlite3.connect(f"file:{global_db}?mode=ro", uri=True)
            cur = conn.cursor()
            cur.execute(
                "SELECT name FROM sqlite_master "
                "WHERE type='table' AND name='cursorDiskKV'"
            )
            if cur.fetchone():
                cur.execute(
                    "SELECT value FROM cursorDiskKV WHERE key LIKE 'composerData:%'"
                )
                for (value,) in cur.fetchall():
                    try:
                        data = json.loads(value) if isinstance(value, str) else {}
                        ws_uri = data.get("workspaceUri", "")
                        ws_path = uri_to_path(ws_uri)
                        if (ws_path and not paths_match(ws_path, project_path)
                                and os.path.basename(os.path.normpath(ws_path))
                                == target_name):
                            alternates.add(ws_path)
                    except (json.JSONDecodeError, AttributeError):
                        pass
            conn.close()
        except (sqlite3.Error, IOError):
            pass

    return sorted(alternates)


def main():
    parser = argparse.ArgumentParser(
        description="Count Cursor IDE prompts per project by day"
    )
    parser.add_argument(
        "--project-path",
        help="Absolute path to the project repo (exact match)",
    )
    parser.add_argument(
        "--filter", help="Substring match on project basename"
    )
    args = parser.parse_args()

    if not args.project_path and not args.filter:
        print(json.dumps({"error": "Provide --project-path or --filter"}))
        return

    cursor_dir = get_cursor_data_dir()

    # Primary: global storage (per-message timestamps from cursorDiskKV)
    daily_global, ts_global, sessions_global = collect_from_global_storage(
        cursor_dir, args.project_path, args.filter
    )

    # Fallback: workspace storage (session-level + legacy per-message timestamps)
    seen_ts = {round(t) for t in ts_global}
    daily_ws, ts_ws, sessions_ws = collect_from_workspace_storage(
        cursor_dir, args.project_path, args.filter, seen_ts
    )

    # Merge
    all_dates = set(daily_global) | set(daily_ws)
    daily_merged = {
        date: daily_global.get(date, 0) + daily_ws.get(date, 0)
        for date in sorted(all_dates)
    }
    all_timestamps = sorted(set(ts_global + ts_ws))

    result = {
        "project_path": args.project_path or args.filter,
        "total_user_messages": sum(daily_merged.values()),
        "daily": daily_merged,
        "timestamps": all_timestamps,
        "sessions_found": sessions_global + sessions_ws,
        "sources": {
            "global_storage": sum(daily_global.values()),
            "workspace_storage": sum(daily_ws.values()),
        },
    }

    # Folder move detection
    if sum(daily_merged.values()) == 0 and args.project_path:
        alternates = find_alternate_paths(cursor_dir, args.project_path)
        if alternates:
            result["alternate_paths"] = alternates
            result["note"] = (
                f"No Cursor sessions found at {args.project_path}, "
                f"but found workspaces for "
                f"'{os.path.basename(args.project_path)}' "
                f"at {len(alternates)} other path(s). "
                f"Project may have been moved."
            )

    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Parse git log to detect work sessions and estimate total hours.

Usage:
  python3 git_sessions.py <repo_path> [--since YYYY-MM-DD] [--until YYYY-MM-DD]
  python3 git_sessions.py /path/to/repo --since 2026-01-01 --until 2026-02-28

Output: JSON with sessions, daily hours, and totals.
"""

import subprocess
import json
import sys
import argparse
from datetime import datetime, timedelta
from collections import defaultdict

SESSION_GAP_HOURS = 1.5   # commits > this apart = new session
SESSION_BUFFER = 0.5       # hours added per session (startup/context-switching)
MIN_SESSION_HOURS = 0.5    # minimum session duration


def get_commits(repo_path, since=None, until=None):
    cmd = ["git", "-C", repo_path, "log", "--format=%H\t%ai\t%s", "--no-merges"]
    if since:
        cmd += [f"--since={since}"]
    if until:
        cmd += [f"--until={until}"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"Error: {result.stderr}", file=sys.stderr)
        return []
    commits = []
    for line in result.stdout.strip().split("\n"):
        if not line:
            continue
        parts = line.split("\t", 2)
        if len(parts) >= 2:
            sha, ts_str = parts[0], parts[1]
            msg = parts[2] if len(parts) > 2 else ""
            try:
                # Parse ISO 8601 with timezone offset
                ts = datetime.fromisoformat(ts_str)
                commits.append({"sha": sha, "ts": ts, "msg": msg})
            except ValueError:
                pass
    return sorted(commits, key=lambda c: c["ts"])


def detect_sessions(commits):
    """Group commits into sessions (gap > SESSION_GAP_HOURS = new session)."""
    if not commits:
        return []
    sessions = []
    current = [commits[0]]
    for commit in commits[1:]:
        gap = (commit["ts"] - current[-1]["ts"]).total_seconds() / 3600
        if gap > SESSION_GAP_HOURS:
            sessions.append(current)
            current = [commit]
        else:
            current.append(commit)
    sessions.append(current)
    return sessions


def session_hours(session):
    """Duration of a session including buffer, minimum enforced."""
    if len(session) == 1:
        raw = 0.0
    else:
        raw = (session[-1]["ts"] - session[0]["ts"]).total_seconds() / 3600
    return max(raw + SESSION_BUFFER, MIN_SESSION_HOURS)


def main():
    parser = argparse.ArgumentParser(description="Estimate work hours from git history")
    parser.add_argument("repo", help="Path to git repository")
    parser.add_argument("--since", help="Start date YYYY-MM-DD")
    parser.add_argument("--until", help="End date YYYY-MM-DD")
    args = parser.parse_args()

    commits = get_commits(args.repo, args.since, args.until)
    if not commits:
        print(json.dumps({"error": "No commits found"}))
        return

    sessions = detect_sessions(commits)

    # Build output
    session_data = []
    daily_hours = defaultdict(float)
    total_hours = 0.0

    for s in sessions:
        start = s[0]["ts"]
        end = s[-1]["ts"]
        h = session_hours(s)
        date_str = start.strftime("%Y-%m-%d")
        daily_hours[date_str] += h
        total_hours += h
        session_data.append({
            "date": date_str,
            "start": start.strftime("%H:%M"),
            "end": end.strftime("%H:%M"),
            "start_h": round(start.hour + start.minute / 60, 3),
            "end_h": round((end.hour + end.minute / 60) + SESSION_BUFFER, 3),
            "duration_h": round(h, 2),
            "commits": len(s),
        })

    print(json.dumps({
        "repo": args.repo,
        "total_commits": len(commits),
        "total_sessions": len(sessions),
        "total_hours": round(total_hours, 2),
        "date_range": {
            "first": commits[0]["ts"].strftime("%Y-%m-%d"),
            "last": commits[-1]["ts"].strftime("%Y-%m-%d"),
        },
        "sessions": session_data,
        "daily_hours": dict(sorted(daily_hours.items())),
    }, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Fetch WakaTime coding time from the WakaTime API.

Reads API key from ~/.wakatime.cfg [settings] api_key.

Usage:
  python3 wakatime_fetch.py --start 2026-01-15 --end 2026-02-02
  python3 wakatime_fetch.py --start 2026-01-15 --end 2026-02-02 --project my-project

Output: JSON with daily summaries and per-project breakdown.
"""

import configparser
import os
import json
import sys
import argparse
import urllib.request
import urllib.parse
import base64
from datetime import datetime, timedelta


def read_api_key():
    cfg_path = os.path.expanduser("~/.wakatime.cfg")
    if not os.path.exists(cfg_path):
        return None
    config = configparser.ConfigParser()
    config.read(cfg_path)
    return config.get("settings", "api_key", fallback=None)


def api_request(endpoint, api_key, params=None):
    base_url = "https://api.wakatime.com/api/v1"
    url = f"{base_url}{endpoint}"
    if params:
        url += "?" + urllib.parse.urlencode(params)

    # WakaTime uses Basic Auth with base64-encoded API key
    encoded_key = base64.b64encode(api_key.encode()).decode()
    req = urllib.request.Request(url, headers={
        "Authorization": f"Basic {encoded_key}",
        "User-Agent": "project-time-tracker-skill/1.0",
    })
    try:
        with urllib.request.urlopen(req, timeout=15) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        return {"error": f"HTTP {e.code}: {e.reason}"}
    except Exception as e:
        return {"error": str(e)}


def fetch_summaries(api_key, start, end, project=None):
    params = {"start": start, "end": end}
    if project:
        params["project"] = project
    data = api_request("/users/current/summaries", api_key, params)
    if "error" in data:
        return data

    # Parse daily summaries
    daily = []
    project_totals = {}
    total_seconds = 0

    for day in data.get("data", []):
        date = day.get("range", {}).get("date", "")
        day_seconds = day.get("grand_total", {}).get("total_seconds", 0)
        total_seconds += day_seconds

        if day_seconds > 0:
            daily.append({
                "date": date,
                "hours": round(day_seconds / 3600, 2),
                "text": day.get("grand_total", {}).get("text", ""),
            })

        # Aggregate per-project
        for proj in day.get("projects", []):
            pname = proj.get("name", "unknown")
            psecs = proj.get("total_seconds", 0)
            project_totals[pname] = project_totals.get(pname, 0) + psecs

    return {
        "start": start,
        "end": end,
        "total_hours": round(total_seconds / 3600, 2),
        "daily": daily,
        "projects": [
            {"project": k, "hours": round(v / 3600, 2)}
            for k, v in sorted(project_totals.items(), key=lambda x: -x[1])
        ],
    }


def fetch_durations(api_key, start, end, project=None):
    """
    Fetch coding duration intervals from the /durations API.
    Each entry has `time` (epoch start) and `duration` (seconds).
    Returns list of [start_epoch, end_epoch] intervals for union computation.

    Note: /durations requires one request per day, so this iterates over the date range.
    """
    intervals = []
    current = datetime.strptime(start, "%Y-%m-%d")
    end_dt = datetime.strptime(end, "%Y-%m-%d")

    while current <= end_dt:
        date_str = current.strftime("%Y-%m-%d")
        params = {"date": date_str}
        if project:
            params["project"] = project
        data = api_request("/users/current/durations", api_key, params)

        if "error" not in data:
            for entry in data.get("data", []):
                t = entry.get("time", 0)
                d = entry.get("duration", 0)
                if t and d > 0:
                    intervals.append([t, t + d])

        current += timedelta(days=1)

    # Merge adjacent/overlapping intervals (WakaTime may have per-file splits)
    if not intervals:
        return intervals
    intervals.sort()
    merged = [intervals[0]]
    for s, e in intervals[1:]:
        if s <= merged[-1][1] + 60:  # 60s tolerance for file switches
            merged[-1][1] = max(merged[-1][1], e)
        else:
            merged.append([s, e])

    return merged


def main():
    parser = argparse.ArgumentParser(description="Fetch WakaTime coding stats")
    parser.add_argument("--start", required=True, help="Start date YYYY-MM-DD")
    parser.add_argument("--end", required=True, help="End date YYYY-MM-DD")
    parser.add_argument("--project", help="Filter by project name")
    args = parser.parse_args()

    api_key = read_api_key()
    if not api_key:
        print(json.dumps({"error": "No WakaTime API key found in ~/.wakatime.cfg"}))
        sys.exit(1)

    result = fetch_summaries(api_key, args.start, args.end, args.project)

    # Also fetch duration intervals for union computation
    # Only do this for active days to avoid hammering the API for empty days
    active_dates = [d["date"] for d in result.get("daily", [])] if "error" not in result else []
    all_intervals = []
    for date_str in active_dates:
        params = {"date": date_str}
        if args.project:
            params["project"] = args.project
        data = api_request("/users/current/durations", api_key, params)
        if "error" not in data:
            for entry in data.get("data", []):
                t = entry.get("time", 0)
                d = entry.get("duration", 0)
                if t and d > 0:
                    all_intervals.append([t, t + d])

    # Merge adjacent per-file intervals
    if all_intervals:
        all_intervals.sort()
        merged_intervals = [all_intervals[0]]
        for s, e in all_intervals[1:]:
            if s <= merged_intervals[-1][1] + 60:
                merged_intervals[-1][1] = max(merged_intervals[-1][1], e)
            else:
                merged_intervals.append([s, e])
        result["intervals"] = merged_intervals
    else:
        result["intervals"] = []

    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()