opencode_stats.py

#!/usr/bin/env python3
import os
import json
import sqlite3
import argparse
import platform
from datetime import datetime
from collections import defaultdict

try:
    from rich.console import Console
    from rich.table import Table
    from rich.panel import Panel
    from rich.layout import Layout
    from rich import box
except ImportError:
    print("Please install 'rich' via 'pip install rich'")
    exit(1)

SCHEMA = {
    "total_sessions": "int",
    "total_messages": "int",
    "total_parts": "int",
    "total_cost": "float",
    "total_api_cost": "float",
    "total_tokens": "dict (total, input, output, reasoning, cache_read, cache_write)",
    "model_usage": "dict (model_key -> {input: int, output: int, reasoning: int, cache_read: int, cache_write: int, cost: float, api_cost: float, messages: int})",
    "project_usage": "dict (project_id -> {messages: int, cost: float, sessions: int, name: str, directory: str})",
    "tool_usage": "dict (tool_name -> int)",
    "session_durations": "list of dict ({title: str, duration: float, date: str})",
    "active_days": "dict (date_str -> int)"
}

# API pricing per 1M tokens (USD) - based on official provider pricing as of March 2026
# Format: {input: price, output: price, cached_input: price, cached_output: price}
# Reasoning tokens are billed as output tokens by all providers
API_PRICING = {
    # Google Gemini
    "google/gemini-3-flash-preview": {"input": 0.50, "output": 3.00, "cached_input": 0.05},
    "google/gemini-3.1-pro-preview": {"input": 2.00, "output": 12.00, "cached_input": 0.20},
    "google/gemini-2.5-pro": {"input": 1.25, "output": 10.00, "cached_input": 0.125},
    "google/gemini-2.5-flash": {"input": 0.30, "output": 2.50, "cached_input": 0.03},

    # MiniMax
    "opencode/minimax-m2.5-free": {"input": 0.30, "output": 1.20, "cached_input": 0.03},
    "opencode-go/minimax-m2.7": {"input": 0.30, "output": 1.20, "cached_input": 0.06},
    "minimax/MiniMax-M2.7-highspeed": {"input": 0.60, "output": 2.40, "cached_input": 0.06},

    # Xiaomi MiMo
    "opencode/mimo-v2-pro-free": {
        "input": {"tier1": 1.00, "tier2": 2.00},
        "output": {"tier1": 3.00, "tier2": 6.00},
        "cached_input": {"tier1": 0.20, "tier2": 0.40}
    },
    "opencode/mimo-v2-omni-free": {"input": 0.40, "output": 2.00, "cached_input": 0.08},

    # Kimi (Moonshot)
    "opencode-go/kimi-k2.5": {"input": 0.50, "output": 2.40, "cached_input": 0.12},

    # Fireworks AI - Kimi models
    "fireworks-ai/accounts/fireworks/routers/kimi-k2p5-turbo": {"input": 0.99, "output": 4.94, "cached_input": 0.16},

    # GLM (Zhipu)
    "opencode-go/glm-5": {"input": 1.00, "output": 3.20, "cached_input": 0.20},
}


def get_agent_guide():
    return """AGENT GUIDE:
This module analyzes OpenCode session history from SQLite database.
To use programmatically:
    import opencode_stats
    stats = opencode_stats.analyze(db_path="path/to/opencode.db", silent=True)

The 'analyze' function returns a dictionary with the following structure:
%s
""" % json.dumps(SCHEMA, indent=4)


def get_default_db_path():
    system = platform.system()
    home = os.path.expanduser("~")

    if system == "Windows":
        return os.path.join(home, ".local", "share", "opencode", "opencode.db")
    else:
        return os.path.join(home, ".local", "share", "opencode", "opencode.db")


def format_duration(seconds):
    if seconds < 60:
        return f"{seconds:.0f}s"
    elif seconds < 3600:
        return f"{seconds / 60:.1f}m"
    else:
        return f"{seconds / 3600:.1f}h"


def format_number(n):
    if n >= 1_000_000:
        return f"{n / 1_000_000:.2f}M"
    elif n >= 1_000:
        return f"{n / 1_000:.1f}K"
    return str(n)


def calculate_api_cost(model_key, input_tokens, output_tokens, reasoning_tokens=0, cache_read=0):
    """Calculate hypothetical API cost based on standard provider pricing."""
    pricing = API_PRICING.get(model_key)
    if not pricing:
        return 0.0

    input_price: float
    output_price: float
    cached_price: float

    if isinstance(pricing["input"], dict) and "tier1" in pricing["input"]:
        input_price = pricing["input"]["tier2"] if input_tokens > 200000 else pricing["input"]["tier1"]
    else:
        input_price = float(pricing["input"])

    if isinstance(pricing["output"], dict) and "tier1" in pricing["output"]:
        output_price = pricing["output"]["tier2"] if output_tokens > 200000 else pricing["output"]["tier1"]
    else:
        output_price = float(pricing["output"])

    cached_input_price = pricing.get("cached_input", pricing["input"])
    if isinstance(cached_input_price, dict) and "tier1" in cached_input_price:
        cached_price = cached_input_price["tier2"] if cache_read > 200000 else cached_input_price["tier1"]
    else:
        cached_price = float(cached_input_price)  # type: ignore

    cost = (input_tokens / 1_000_000 * input_price) + \
           (cache_read / 1_000_000 * cached_price) + \
           ((output_tokens + reasoning_tokens) / 1_000_000 * output_price)

    return cost


def get_project_names(conn):
    cursor = conn.cursor()
    cursor.execute("SELECT id, name, worktree FROM project")
    projects = {}
    for row in cursor.fetchall():
        proj_id, name, worktree = row
        # Handle cases where name is None, empty, or literal "None" string
        if name and name != "None":
            display_name = name
        elif worktree and worktree != "/":
            display_name = os.path.basename(worktree)
        else:
            display_name = proj_id[:8]
        projects[proj_id] = {
            "name": display_name,
            "directory": worktree or ""
        }
    return projects


def analyze(db_path=None, silent=False, status_callback=None):
    if db_path is None:
        db_path = get_default_db_path()

    db_path = os.path.normpath(db_path)

    if not os.path.exists(db_path):
        if not silent:
            print(f"Error: Database '{db_path}' not found.")
        return None

    if not silent:
        print(f"Analyzing sessions in {db_path}...")

    stats = analyze_database(db_path, status_callback=status_callback)

    if not silent:
        if stats["total_sessions"] == 0:
            print("No sessions found in the database.")
        else:
            display_stats(stats)

    return stats


def analyze_database(db_path, status_callback=None):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    if status_callback:
        status_callback('Loading project data...')
    project_map = get_project_names(conn)

    stats = {
        "total_sessions": 0,
        "total_messages": 0,
        "total_parts": 0,
        "total_cost": 0.0,
        "total_api_cost": 0.0,
        "total_tokens": {"total": 0, "input": 0, "output": 0, "reasoning": 0, "cache_read": 0, "cache_write": 0},
        "model_usage": defaultdict(lambda: {"input": 0, "output": 0, "reasoning": 0, "cache_read": 0, "cache_write": 0, "cost": 0.0, "api_cost": 0.0, "messages": 0}),
        "project_usage": defaultdict(lambda: {"messages": 0, "cost": 0.0, "api_cost": 0.0, "sessions": 0, "name": "", "directory": ""}),
        "tool_usage": defaultdict(int),
        "session_durations": [],
        "active_days": defaultdict(int)
    }

    if status_callback:
        status_callback('Analyzing sessions...')
    cursor.execute("SELECT COUNT(*) FROM session")
    stats["total_sessions"] = cursor.fetchone()[0]

    if status_callback:
        status_callback('Analyzing messages...')
    cursor.execute("SELECT COUNT(*) FROM message")
    stats["total_messages"] = cursor.fetchone()[0]

    cursor.execute("SELECT COUNT(*) FROM part")
    stats["total_parts"] = cursor.fetchone()[0]

    if status_callback:
        status_callback('Processing project usage...')
    cursor.execute('''
        SELECT s.project_id, COUNT(DISTINCT s.id) as sessions, COUNT(m.id) as messages
        FROM session s
        LEFT JOIN message m ON m.session_id = s.id
        GROUP BY s.project_id
    ''')
    for row in cursor.fetchall():
        proj_id, sessions, messages = row
        proj_info = project_map.get(proj_id, {"name": "Unknown", "directory": ""})
        usage = stats["project_usage"][proj_id]
        usage["sessions"] = sessions
        usage["messages"] = messages
        usage["name"] = proj_info["name"]
        usage["directory"] = proj_info["directory"]

    if status_callback:
        status_callback('Processing model usage...')
    cursor.execute('''
        SELECT m.session_id, m.data
        FROM message m
        WHERE json_extract(m.data, "$.role") = "assistant"
    ''')
    for row in cursor.fetchall():
        session_id = row[0]
        try:
            msg_data = json.loads(row[1])
        except (json.JSONDecodeError, TypeError):
            continue

        provider = msg_data.get("providerID", "unknown")
        model = msg_data.get("modelID", "unknown")
        model_key = f"{provider}/{model}"

        cost = msg_data.get("cost", 0.0)
        tokens = msg_data.get("tokens", {})

        input_tokens = tokens.get("input", 0)
        output_tokens = tokens.get("output", 0)
        reasoning_tokens = tokens.get("reasoning", 0)
        cache = tokens.get("cache", {})
        cache_read = cache.get("read", 0)
        cache_write = cache.get("write", 0)

        stats["model_usage"][model_key]["input"] += input_tokens
        stats["model_usage"][model_key]["output"] += output_tokens
        stats["model_usage"][model_key]["reasoning"] += reasoning_tokens
        stats["model_usage"][model_key]["cache_read"] += cache_read
        stats["model_usage"][model_key]["cache_write"] += cache_write
        stats["model_usage"][model_key]["cost"] += cost
        stats["model_usage"][model_key]["messages"] += 1

        api_cost = calculate_api_cost(model_key, input_tokens, output_tokens, reasoning_tokens, cache_read)
        stats["model_usage"][model_key]["api_cost"] += api_cost

        stats["total_cost"] += cost
        stats["total_api_cost"] += api_cost
        stats["total_tokens"]["total"] += tokens.get("total", 0)
        stats["total_tokens"]["input"] += input_tokens
        stats["total_tokens"]["output"] += output_tokens
        stats["total_tokens"]["reasoning"] += reasoning_tokens
        stats["total_tokens"]["cache_read"] += cache_read
        stats["total_tokens"]["cache_write"] += cache_write

        if session_id:
            cursor2 = conn.cursor()
            cursor2.execute('SELECT project_id FROM session WHERE id = ?', (session_id,))
            proj_row = cursor2.fetchone()
            if proj_row:
                stats["project_usage"][proj_row[0]]["cost"] += cost
                stats["project_usage"][proj_row[0]]["api_cost"] += api_cost

    if status_callback:
        status_callback('Processing tool usage...')
    cursor.execute('''
        SELECT json_extract(data, "$.tool"), COUNT(*)
        FROM part
        WHERE json_extract(data, "$.type") = "tool"
        GROUP BY json_extract(data, "$.tool")
    ''')
    for row in cursor.fetchall():
        tool_name = row[0] or "unknown"
        stats["tool_usage"][tool_name] = row[1]

    if status_callback:
        status_callback('Processing session durations...')
    cursor.execute('''
        SELECT title, time_created, time_updated,
               (time_updated - time_created) as duration_ms
        FROM session
        WHERE time_created IS NOT NULL AND time_updated IS NOT NULL
    ''')
    for row in cursor.fetchall():
        title, time_created, time_updated, duration_ms = row
        if duration_ms and duration_ms > 0:
            duration_sec = duration_ms / 1000.0
            date_str = datetime.fromtimestamp(time_created / 1000).strftime("%Y-%m-%d")
            stats["session_durations"].append({
                "title": title or "Untitled",
                "duration": duration_sec,
                "date": date_str
            })
            stats["active_days"][date_str] += 1

    conn.close()
    return stats


def display_stats(stats):
    console = Console(width=120)
    console.print("\n[bold cyan]OpenCode Stats Plus[/bold cyan]\n")

    summary_table = Table(title="Overall Summary", box=box.ROUNDED)
    summary_table.add_column("Metric", style="cyan")
    summary_table.add_column("Value", style="magenta")

    summary_table.add_row("Total Sessions", str(stats["total_sessions"]))
    summary_table.add_row("Total Messages", str(stats["total_messages"]))
    summary_table.add_row("Total Parts", str(stats["total_parts"]))
    summary_table.add_row("Actual Cost (OpenCode)", f"${stats['total_cost']:.4f}")
    summary_table.add_row("Hypothetical API Cost", f"${stats['total_api_cost']:.4f}")
    summary_table.add_row("Total Tokens", format_number(stats["total_tokens"]["total"]))

    console.print(summary_table)

    if stats["project_usage"]:
        proj_table = Table(title="Usage by Project", box=box.ROUNDED)
        proj_table.add_column("Project", style="cyan")
        proj_table.add_column("Sessions", justify="right")
        proj_table.add_column("Messages", justify="right")
        proj_table.add_column("OpenCode Cost", justify="right", style="bold yellow")
        proj_table.add_column("API Cost", justify="right", style="bold red")

        sorted_projs = sorted(stats["project_usage"].values(), key=lambda x: x["cost"] + x["api_cost"], reverse=True)
        for usage in sorted_projs:
            proj_table.add_row(
                usage["name"],
                str(usage["sessions"]),
                str(usage["messages"]),
                f"${usage['cost']:.4f}",
                f"${usage['api_cost']:.4f}"
            )
        console.print(proj_table)

    if stats["model_usage"]:
        model_table = Table(title="Usage by Model", box=box.ROUNDED)
        model_table.add_column("Model", style="green")
        model_table.add_column("Msgs", justify="right")
        model_table.add_column("Input", justify="right")
        model_table.add_column("Output", justify="right")
        model_table.add_column("Reason", justify="right", style="dim")
        model_table.add_column("Cache", justify="right", style="dim")
        model_table.add_column("Actual", justify="right", style="bold yellow")
        model_table.add_column("API Cost", justify="right", style="bold red")

        sorted_models = sorted(stats["model_usage"].items(), key=lambda x: x[1]["api_cost"], reverse=True)

        for model, usage in sorted_models:
            # Truncate very long model names
            display_model = model if len(model) <= 50 else model[:47] + "..."
            model_table.add_row(
                display_model,
                str(usage["messages"]),
                format_number(usage["input"]),
                format_number(usage["output"]),
                format_number(usage["reasoning"]),
                format_number(usage["cache_read"]),
                f"${usage['cost']:.4f}",
                f"${usage['api_cost']:.4f}"
            )
        console.print(model_table)

    if stats["tool_usage"]:
        tool_table = Table(title="Top Tools Used", box=box.ROUNDED)
        tool_table.add_column("Tool", style="blue")
        tool_table.add_column("Calls", justify="right")

        sorted_tools = sorted(stats["tool_usage"].items(), key=lambda x: x[1], reverse=True)[:10]
        for tool, count in sorted_tools:
            tool_table.add_row(tool, str(count))

        console.print(tool_table)

    if stats["active_days"] or stats["session_durations"]:
        stats_table = Table(title="Activity Highlights", box=box.ROUNDED)
        stats_table.add_column("Category", style="cyan")
        stats_table.add_column("Details", style="white")

        if stats["active_days"]:
            top_day = max(stats["active_days"].items(), key=lambda x: x[1])
            stats_table.add_row("Most Active Day", f"{top_day[0]} ({top_day[1]} sessions)")

        if stats["session_durations"]:
            longest_session = max(stats["session_durations"], key=lambda x: x["duration"])
            stats_table.add_row("Longest Session", f"{longest_session['date']} - {format_duration(longest_session['duration'])}")
            stats_table.add_row("Session Title", f"{longest_session['title'][:60]}")

        console.print(stats_table)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Analyze OpenCode usage statistics.")
    parser.add_argument("--path", "-p", type=str, help="Path to OpenCode database. Defaults to ~/.local/share/opencode/opencode.db")
    parser.add_argument("--silent", "-s", action="store_true", help="Run in silent mode (only for programmatic check)")

    args = parser.parse_args()
    path = args.path if args.path else get_default_db_path()

    analyze(db_path=path, silent=args.silent)