get_commit_report.py

#!/usr/bin/env python3
"""
Fetch commit report from Bitbucket (per user, for a date range) using the REST API.

Usage:
  cp .env.example .env   # then edit .env with your credentials
  python get_commit_report.py --workspace MYWORKSPACE --from 2025-01-01 --to 2025-01-31
  python get_commit_report.py --workspace MYWORKSPACE --repo my-repo --from 2025-01-01 --to 2025-01-31

If --repo is omitted, all repositories in the workspace are included.
Optional: --branch main  (uses POST /commits with include to limit to that branch)
"""

import argparse
import os
import sys

from dotenv import load_dotenv

# Load .env from script directory so it works from any cwd
load_dotenv(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env"))
from collections import defaultdict
from datetime import datetime, timezone

try:
    from tqdm import tqdm
except ImportError:
    tqdm = None

# Prefer requests if available
try:
    import requests
except ImportError:
    requests = None


def get_auth():
    user = (os.environ.get("BITBUCKET_USER") or "").strip()
    token = (
        os.environ.get("BITBUCKET_APP_PASSWORD")
        or os.environ.get("BITBUCKET_API_TOKEN")
        or ""
    ).strip()
    if not user or not token:
        print(
            "Set BITBUCKET_USER and BITBUCKET_APP_PASSWORD (or BITBUCKET_API_TOKEN) in .env.",
            file=sys.stderr,
        )
        sys.exit(1)
    if "example.com" in user or "your-" in token:
        print(
            "Replace placeholder values in .env with your real Bitbucket email and API token.",
            file=sys.stderr,
        )
        sys.exit(1)
    return (user, token)


def _check_401(response, url: str):
    """On 401, print helpful message and exit."""
    if response.status_code != 401:
        return
    print("Bitbucket returned 401 Unauthorized. Check:", file=sys.stderr)
    print(
        "  • BITBUCKET_USER must be your Bitbucket account email (not username).",
        file=sys.stderr,
    )
    print(
        "  • Use an API token: Bitbucket → Personal settings → API tokens → Create token (Repository read).",
        file=sys.stderr,
    )
    print(
        "  • You must have access to the workspace (e.g. be a member of MYWORKSPACE).",
        file=sys.stderr,
    )
    sys.exit(1)


REQUEST_TIMEOUT = 30


def fetch_commits_get(
    workspace: str, repo: str, auth, pagelen: int = 100, page: int = 1
):
    url = (
        f"https://api.bitbucket.org/2.0/repositories/{workspace}/{repo}/commits"
        f"?pagelen={pagelen}&page={page}"
    )
    if requests:
        r = requests.get(url, auth=auth, timeout=REQUEST_TIMEOUT)
        _check_401(r, url)
        r.raise_for_status()
        return r.json()
    # Fallback: urllib
    import base64
    import urllib.request

    req = urllib.request.Request(url)
    req.add_header(
        "Authorization",
        "Basic " + base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode(),
    )
    with urllib.request.urlopen(req) as resp:
        import json

        return json.load(resp)


def fetch_commits_post(
    workspace: str, repo: str, auth, branch: str, pagelen: int = 100, page: int = 1
):
    url = f"https://api.bitbucket.org/2.0/repositories/{workspace}/{repo}/commits?pagelen={pagelen}&page={page}"
    body = {"include": f"refs/heads/{branch}"}
    if requests:
        r = requests.post(url, auth=auth, json=body, timeout=REQUEST_TIMEOUT)
        _check_401(r, url)
        r.raise_for_status()
        return r.json()
    import base64
    import json
    import urllib.request

    req = urllib.request.Request(url, data=json.dumps(body).encode(), method="POST")
    req.add_header(
        "Authorization",
        "Basic " + base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode(),
    )
    req.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(req) as resp:
        return json.load(resp)


def fetch_repositories(workspace: str, auth, pagelen: int = 100, page: int = 1):
    """List repositories in a workspace (paginated)."""
    url = (
        f"https://api.bitbucket.org/2.0/repositories/{workspace}"
        f"?pagelen={pagelen}&page={page}"
    )
    if requests:
        r = requests.get(url, auth=auth, timeout=REQUEST_TIMEOUT)
        _check_401(r, url)
        r.raise_for_status()
        return r.json()
    import base64
    import urllib.request

    req = urllib.request.Request(url)
    req.add_header(
        "Authorization",
        "Basic " + base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode(),
    )
    with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
        import json

        return json.load(resp)


def commit_author_key(c):
    """Unique key for grouping by author."""
    author = c.get("author") or {}
    user = author.get("user") or {}
    if user.get("uuid"):
        return (
            "uuid",
            user["uuid"],
            user.get("display_name") or user.get("nickname") or "?",
        )
    if user.get("nickname"):
        return (
            "nickname",
            user["nickname"],
            user.get("display_name") or user["nickname"],
        )
    raw = author.get("raw") or "unknown"
    return ("raw", raw, raw)


def collect_commits_for_repo(
    workspace: str,
    repo: str,
    auth,
    start,
    end,
    branch,
    pagelen: int,
    repo_display_name: str = None,
    repo_index: int = None,
    repo_total: int = None,
):
    """Fetch commits for one repo in the date range; return (by_user, total_in_range)."""
    by_user = defaultdict(int)
    total_in_range = 0
    page = 1
    past_end = False
    display = repo_display_name or repo

    if branch:

        def fetch(p):
            return fetch_commits_post(workspace, repo, auth, branch, pagelen, p)

    else:

        def fetch(p):
            return fetch_commits_get(workspace, repo, auth, pagelen, p)

    pbar_commits = None
    try:
        while True:
            data = fetch(page)
            values = data.get("values") or []

            # Create progress bar after first response (we may get total from API)
            if pbar_commits is None and tqdm is not None:
                total_size = data.get(
                    "size"
                )  # Bitbucket paginated response sometimes has total
                pbar_commits = tqdm(
                    total=total_size,
                    desc=(
                        f"Repo {repo_index}/{repo_total}: {display}"
                        if repo_total
                        else display
                    ),
                    unit=" commits",
                    position=1,
                    leave=False,
                    file=sys.stderr,
                )

            if pbar_commits is not None:
                pbar_commits.update(len(values))

            if not values:
                break
            for c in values:
                date_str = c.get("date")
                if not date_str:
                    continue
                commit_dt = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
                if commit_dt < start:
                    past_end = True
                    break
                if commit_dt > end:
                    continue
                total_in_range += 1
                key = commit_author_key(c)
                by_user[key] += 1
            if past_end:
                break
            if not data.get("next"):
                break
            page += 1
    finally:
        if pbar_commits is not None:
            pbar_commits.close()

    return by_user, total_in_range


def get_workspace_repos(workspace: str, auth, pagelen: int = 100):
    """Yield (slug, name) for each repo in the workspace."""
    page = 1
    while True:
        data = fetch_repositories(workspace, auth, pagelen, page)
        for r in data.get("values") or []:
            yield (r["slug"], r.get("name") or r["slug"])
        if not data.get("next"):
            break
        page += 1


def main():
    ap = argparse.ArgumentParser(
        description="Bitbucket commit report per user for a date range"
    )
    ap.add_argument("--workspace", required=True, help="Bitbucket workspace slug")
    ap.add_argument(
        "--repo",
        default=None,
        help="Repository slug (omit to include all repos in workspace)",
    )
    ap.add_argument(
        "--from", dest="date_from", required=True, help="Start date YYYY-MM-DD"
    )
    ap.add_argument("--to", dest="date_to", required=True, help="End date YYYY-MM-DD")
    ap.add_argument(
        "--branch", default=None, help="Optional: limit to branch (e.g. main)"
    )
    ap.add_argument("--pagelen", type=int, default=100, help="Page size (default 100)")
    args = ap.parse_args()

    try:
        start = datetime.strptime(args.date_from, "%Y-%m-%d").replace(
            tzinfo=timezone.utc
        )
        end = datetime.strptime(args.date_to, "%Y-%m-%d").replace(
            hour=23, minute=59, second=59, tzinfo=timezone.utc
        )
    except ValueError:
        print("Invalid date; use YYYY-MM-DD.", file=sys.stderr)
        sys.exit(1)

    auth = get_auth()
    by_user = defaultdict(int)
    total_in_range = 0

    if args.repo:
        repos = [(args.repo, args.repo)]
    else:
        repos = list(get_workspace_repos(args.workspace, auth, args.pagelen))
        if not repos:
            print("No repositories found in workspace.", file=sys.stderr)
            sys.exit(1)

    num_repos = len(repos)
    repo_iterator = (
        tqdm(
            repos,
            desc="Repos",
            total=num_repos,
            unit=" repo",
            position=0,
            leave=True,
            file=sys.stderr,
        )
        if tqdm
        else repos
    )

    for idx, (slug, name) in enumerate(repo_iterator, start=1):
        repo_by_user, repo_total = collect_commits_for_repo(
            args.workspace,
            slug,
            auth,
            start,
            end,
            args.branch,
            args.pagelen,
            repo_display_name=name,
            repo_index=idx,
            repo_total=num_repos,
        )
        for key, count in repo_by_user.items():
            by_user[key] += count
        total_in_range += repo_total

    # Merge counts by display name so the same person doesn't appear twice
    by_display = defaultdict(int)
    for (_, _, display), count in by_user.items():
        by_display[display.strip() or "unknown"] += count

    scope = f"repo {args.repo}" if args.repo else f"workspace {args.workspace}"
    print(
        f"\nCommits in {args.date_from} .. {args.date_to} ({scope}, branch: {args.branch or 'default'}): {total_in_range}\n"
    )
    print("User (display)                    Count")
    print("-" * 45)
    for display, count in sorted(by_display.items(), key=lambda x: -x[1]):
        print(f"  {(display[:32]):32}  {count}")


if __name__ == "__main__":
    main()