copilot_usage.py

#!/usr/bin/env python3
"""Fetch GitHub Copilot per-user usage metrics and post a summary to Slack."""

import json
import os
import sys
from collections import defaultdict
from datetime import datetime, timedelta, timezone

import requests

GITHUB_API_BASE = "https://api.github.com"
API_VERSION = "2022-11-28"


def get_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        print(f"Error: {name} environment variable is required", file=sys.stderr)
        sys.exit(1)
    return value


def github_headers(token: str) -> dict:
    return {
        "Accept": "application/vnd.github+json",
        "Authorization": f"Bearer {token}",
        "X-GitHub-Api-Version": API_VERSION,
    }


def fetch_user_report(token: str, org: str, day: str) -> list[dict]:
    """Fetch per-user usage metrics for a specific day.

    Returns the parsed NDJSON records from the downloaded report files.
    """
    url = f"{GITHUB_API_BASE}/orgs/{org}/copilot/metrics/reports/users-1-day"
    resp = requests.get(
        url, headers=github_headers(token), params={"day": day}, timeout=30
    )

    if resp.status_code == 204:
        print(f"No data available for {day}")
        return []

    resp.raise_for_status()
    data = resp.json()

    download_links = data.get("download_links", [])
    if not download_links:
        print(f"No download links returned for {day}")
        return []

    records = []
    for link in download_links:
        dl_resp = requests.get(link, timeout=60)
        dl_resp.raise_for_status()
        # Reports are NDJSON (one JSON object per line)
        for line in dl_resp.text.strip().splitlines():
            line = line.strip()
            if line:
                records.append(json.loads(line))

    return records


def build_user_summary(records: list[dict]) -> list[dict]:
    """Build a per-user summary from the raw NDJSON records."""
    users = []

    for record in records:
        user_login = record.get("user_login", "unknown")
        interactions = record.get("user_initiated_interaction_count", 0)
        code_gen = record.get("code_generation_activity_count", 0)
        code_accept = record.get("code_acceptance_activity_count", 0)
        loc_added = record.get("loc_added_sum", 0)
        loc_suggested = record.get("loc_suggested_to_add_sum", 0)

        # Model breakdown from totals_by_model_feature
        models = defaultdict(int)
        for entry in record.get("totals_by_model_feature", []):
            model_name = entry.get("model", "unknown")
            count = entry.get("user_initiated_interaction_count", 0)
            models[model_name] += count

        # Also check totals_by_language_model
        for entry in record.get("totals_by_language_model", []):
            model_name = entry.get("model", "unknown")
            count = entry.get("user_initiated_interaction_count", 0)
            if model_name not in models:
                models[model_name] += count

        # CLI token usage
        cli_tokens = 0
        cli_data = record.get("totals_by_cli")
        if cli_data:
            token_usage = cli_data.get("token_usage", {})
            cli_tokens = (
                token_usage.get("output_tokens_sum", 0)
                + token_usage.get("prompt_tokens_sum", 0)
            )

        # Feature breakdown
        features = {}
        for entry in record.get("totals_by_feature", []):
            feature_name = entry.get("feature", "unknown")
            feat_interactions = entry.get("user_initiated_interaction_count", 0)
            feat_code_gen = entry.get("code_generation_activity_count", 0)
            features[feature_name] = {
                "interactions": feat_interactions,
                "code_generations": feat_code_gen,
            }

        used_agent = record.get("used_agent", False)
        used_chat = record.get("used_chat", False)
        used_cli = record.get("used_cli", False)

        users.append(
            {
                "user": user_login,
                "interactions": interactions,
                "code_generations": code_gen,
                "code_acceptances": code_accept,
                "loc_added": loc_added,
                "loc_suggested": loc_suggested,
                "models": dict(models),
                "features": features,
                "cli_tokens": cli_tokens,
                "used_agent": used_agent,
                "used_chat": used_chat,
                "used_cli": used_cli,
            }
        )

    # Sort by interactions descending
    users.sort(key=lambda u: u["interactions"], reverse=True)
    return users


def format_slack_message(users: list[dict], org: str, day: str) -> dict:
    """Format the user summary as a Slack Block Kit message."""
    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"Copilot Usage Report — {day}",
            },
        },
        {
            "type": "context",
            "elements": [
                {
                    "type": "mrkdwn",
                    "text": f"Organization: *{org}* | Active users: *{len(users)}*",
                }
            ],
        },
        {"type": "divider"},
    ]

    if not users:
        blocks.append(
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": "_No usage data for this day._"},
            }
        )
        return {"blocks": blocks}

    # Summary table
    lines = []
    for u in users:
        activity_flags = []
        if u["used_chat"]:
            activity_flags.append("chat")
        if u["used_agent"]:
            activity_flags.append("agent")
        if u["used_cli"]:
            activity_flags.append("cli")
        activity_str = ", ".join(activity_flags) if activity_flags else "completions"

        line = (
            f"*{u['user']}* — "
            f"{u['interactions']} interactions, "
            f"{u['code_generations']} generations, "
            f"{u['loc_added']} LoC added "
            f"({activity_str})"
        )
        lines.append(line)

    # Slack sections have a 3000 char limit; split if needed
    chunk = []
    chunk_len = 0
    for line in lines:
        if chunk_len + len(line) + 1 > 2900:
            blocks.append(
                {
                    "type": "section",
                    "text": {"type": "mrkdwn", "text": "\n".join(chunk)},
                }
            )
            chunk = []
            chunk_len = 0
        chunk.append(line)
        chunk_len += len(line) + 1

    if chunk:
        blocks.append(
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": "\n".join(chunk)},
            }
        )

    # Model breakdown (aggregate across all users)
    model_totals = defaultdict(int)
    for u in users:
        for model, count in u["models"].items():
            model_totals[model] += count

    if model_totals:
        blocks.append({"type": "divider"})
        model_lines = ["*Model usage (interactions):*"]
        for model, count in sorted(
            model_totals.items(), key=lambda x: x[1], reverse=True
        ):
            model_lines.append(f"• {model}: {count}")
        blocks.append(
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": "\n".join(model_lines)},
            }
        )

    # CLI token totals
    total_cli_tokens = sum(u["cli_tokens"] for u in users)
    if total_cli_tokens > 0:
        cli_users = [u for u in users if u["cli_tokens"] > 0]
        cli_lines = [f"*CLI token usage (total: {total_cli_tokens:,}):*"]
        for u in sorted(cli_users, key=lambda x: x["cli_tokens"], reverse=True):
            cli_lines.append(f"• {u['user']}: {u['cli_tokens']:,} tokens")
        blocks.append(
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": "\n".join(cli_lines)},
            }
        )

    return {"blocks": blocks}


def post_to_slack(webhook_url: str, message: dict) -> None:
    resp = requests.post(webhook_url, json=message, timeout=15)
    resp.raise_for_status()
    if resp.text != "ok":
        print(f"Slack response: {resp.text}", file=sys.stderr)


def main():
    github_token = get_env("GH_TOKEN")
    org = os.environ.get("GITHUB_ORG", "shadow-robot")
    slack_webhook = get_env("SLACK_WEBHOOK_URL")

    # Default to yesterday
    day = os.environ.get("REPORT_DAY")
    if not day:
        yesterday = datetime.now(timezone.utc) - timedelta(days=1)
        day = yesterday.strftime("%Y-%m-%d")

    print(f"Fetching Copilot usage for {org} on {day}...")

    records = fetch_user_report(github_token, org, day)
    print(f"Retrieved {len(records)} user records")

    users = build_user_summary(records)
    message = format_slack_message(users, org, day)

    print("Posting to Slack...")
    post_to_slack(slack_webhook, message)
    print("Done!")


if __name__ == "__main__":
    main()