active_committers.py

#!/usr/bin/env python3
"""
Count unique active committers in a Bitbucket Cloud workspace over a rolling window.

Authentication: Atlassian API token + your Atlassian account email (Basic Auth).
  Generate a token at: https://id.atlassian.com/manage-profile/security/api-tokens
  No scope selection needed — classic tokens inherit your account permissions.
  Use with: --email you@example.com --token <token>
"""

import argparse
import datetime
import re
import sys
import time

import requests

# ── defaults (override via CLI flags) ──────────────────────────────────────────
DEFAULT_DAYS = 90
MAX_API_CALLS_BEFORE_SLEEP = 950   # Bitbucket Cloud cap is 1 000 calls/hour
RATE_LIMIT_SLEEP_SECONDS = 3900    # 65 minutes — a little over the 60-min window

BB_API_BASE = "https://api.bitbucket.org/2.0"

# ── globals ────────────────────────────────────────────────────────────────────
_api_call_count = 0
_session = requests.Session()


# ── CLI ────────────────────────────────────────────────────────────────────────

def parse_args():
    p = argparse.ArgumentParser(
        description="Count unique active committers in a Bitbucket Cloud workspace."
    )
    p.add_argument("--email",
                   help="Atlassian account email address (required with --token)")
    p.add_argument("--token",
                   help="Atlassian API token from id.atlassian.com. "
                        "Not required for public workspaces.")
    p.add_argument("--workspace",  required=True,
                   help="Bitbucket Cloud workspace slug")
    p.add_argument("--repo",
                   help="Limit to a single repository slug (default: all repos)")
    p.add_argument("--days", type=int, default=DEFAULT_DAYS,
                   help=f"Lookback window in days (default: {DEFAULT_DAYS})")
    p.add_argument("--list-authors", action="store_true",
                   help="Print each unique author after the summary")
    return p.parse_args()


# ── HTTP helpers ───────────────────────────────────────────────────────────────

def _get_json(url, params=None):
    """Single authenticated GET, returns parsed JSON. Raises on HTTP errors."""
    global _api_call_count

    if _api_call_count >= MAX_API_CALLS_BEFORE_SLEEP:
        print(
            f"\nApproaching Bitbucket rate limit ({_api_call_count} calls). "
            f"Sleeping {RATE_LIMIT_SLEEP_SECONDS // 60} minutes …",
            flush=True,
        )
        time.sleep(RATE_LIMIT_SLEEP_SECONDS)
        print("Resuming.", flush=True)
        _api_call_count = 0

    resp = _session.get(url, params=params, timeout=30)
    _api_call_count += 1

    if resp.status_code == 429:
        retry_after = int(resp.headers.get("Retry-After", RATE_LIMIT_SLEEP_SECONDS))
        print(f"\nRate-limited by Bitbucket. Sleeping {retry_after}s …", flush=True)
        time.sleep(retry_after)
        return _get_json(url, params)   # retry once

    resp.raise_for_status()
    return resp.json()


def _paginate(url, params=None):
    """Yield every item from a paginated Bitbucket endpoint."""
    while url:
        data = _get_json(url, params)
        yield from data.get("values", [])
        url = data.get("next")
        params = None   # 'next' URL already carries query params


# ── repo discovery ─────────────────────────────────────────────────────────────

def get_repos(workspace, single_repo=None):
    if single_repo:
        url = f"{BB_API_BASE}/repositories/{workspace}/{single_repo}"
        repo = _get_json(url)
        return [{
            "full_name": repo["full_name"],
            "commits_url": repo["links"]["commits"]["href"],
        }]

    repos = []
    for repo in _paginate(f"{BB_API_BASE}/repositories/{workspace}"):
        repos.append({
            "full_name": repo["full_name"],
            "commits_url": repo["links"]["commits"]["href"],
        })
    return repos


# ── commit harvesting ──────────────────────────────────────────────────────────

_EMAIL_RE = re.compile(r"<([^>]+)>")


def _extract_email(raw_author):
    """Return lowercased email from 'Name <email>' or the raw string itself."""
    m = _EMAIL_RE.search(raw_author)
    return m.group(1).lower() if m else raw_author.strip().lower()


def get_active_authors_for_repo(commits_url, cutoff_date):
    """
    Return a set of (raw_author, email) tuples for commits after cutoff_date.
    Stops paginating as soon as commits fall outside the window.
    """
    authors = {}   # email → raw_author (keep first seen display name)

    for commit in _paginate(commits_url):
        raw_date = commit["date"]                       # "2024-03-15T12:00:00+00:00"
        commit_date = datetime.datetime.fromisoformat(raw_date).replace(tzinfo=None)

        if commit_date < cutoff_date:
            break   # commits are newest-first; everything after this is older

        raw_author = commit["author"]["raw"]
        email = _extract_email(raw_author)
        if email not in authors:
            authors[email] = raw_author

    return authors


# ── main ───────────────────────────────────────────────────────────────────────

def main():
    args = parse_args()

    if args.token and args.email:
        _session.auth = (args.email, args.token)
    elif args.token and not args.email:
        print("Error: --email is required when using --token", file=sys.stderr)
        sys.exit(1)

    cutoff_date = datetime.datetime.utcnow() - datetime.timedelta(days=args.days)

    print(f"Workspace : {args.workspace}")
    print(f"Lookback  : {args.days} days (since {cutoff_date.date()})")
    if args.repo:
        print(f"Repo      : {args.repo}")
    print()

    print("Fetching repository list …", end=" ", flush=True)
    try:
        repos = get_repos(args.workspace, args.repo)
    except requests.HTTPError as exc:
        print(f"\nError fetching repos: {exc}", file=sys.stderr)
        sys.exit(1)
    print(f"{len(repos)} repo(s) found.")

    all_authors = {}   # email → raw_author

    for i, repo in enumerate(repos, 1):
        name = repo["full_name"]
        print(f"  [{i}/{len(repos)}] {name}", end=" … ", flush=True)

        try:
            authors = get_active_authors_for_repo(repo["commits_url"], cutoff_date)
        except requests.HTTPError as exc:
            print(f"skipped ({exc})")
            continue

        new = sum(1 for e in authors if e not in all_authors)
        all_authors.update(authors)
        print(f"{len(authors)} active committer(s) (+{new} new unique)")

    print()
    print("=" * 50)
    print(f"Total unique active committers : {len(all_authors)}")
    print(f"Lookback window                : {args.days} days")
    print(f"Repositories scanned           : {len(repos)}")
    print(f"API calls made                 : {_api_call_count}")
    print("=" * 50)

    if args.list_authors:
        print("\nAuthors:")
        for raw in sorted(all_authors.values()):
            print(f"  {raw}")


if __name__ == "__main__":
    main()