sync_labels.py

#!/usr/bin/env python3
"""Sync GitHub labels from a JSON definition file to one or more repositories.

Usage:
    python sync_labels.py --token ghp_xxx
    python sync_labels.py --token ghp_xxx --repos 3dcitydb/3dcitydb
    python sync_labels.py --token ghp_xxx --dry-run
    python sync_labels.py --token ghp_xxx --remove-all --dry-run
    python sync_labels.py --token ghp_xxx --remove-only
    python sync_labels.py --token ghp_xxx --file my-labels.json
"""

import argparse
import json
import os
import sys
from urllib.parse import quote

import requests

API_BASE = "https://api.github.com"
DEFAULT_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "labels.json")


def load_labels(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def normalize_color(color):
    return color.lstrip("#").lower()


def get_existing_labels(session, repo):
    """Fetch all existing labels for a repo, handling pagination."""
    labels = []
    url = f"{API_BASE}/repos/{repo}/labels"
    params = {"per_page": 100}
    while url:
        resp = session.get(url, params=params)
        resp.raise_for_status()
        labels.extend(resp.json())
        # Follow pagination links
        url = resp.links.get("next", {}).get("url")
        params = None  # params already encoded in the next URL
    return labels


def check_rate_limit(session):
    resp = session.get(f"{API_BASE}/rate_limit")
    if resp.ok:
        data = resp.json()
        remaining = data["resources"]["core"]["remaining"]
        limit = data["resources"]["core"]["limit"]
        if remaining < 50:
            print(f"  WARNING: GitHub API rate limit low: {remaining}/{limit} remaining")
        return remaining
    return None


def remove_all_labels(session, repo, dry_run=False):
    """Remove all existing labels from a repository."""
    existing = get_existing_labels(session, repo)
    if not existing:
        print("  No existing labels to remove.")
        return 0

    removed = 0
    for label in existing:
        name = label["name"]
        if dry_run:
            print(f"  DELETE  {name}")
        else:
            url = f"{API_BASE}/repos/{repo}/labels/{quote(name, safe='')}"
            resp = session.delete(url)
            resp.raise_for_status()
            print(f"  DELETED {name}")
        removed += 1

    print(f"\n  Removed: {removed} label(s)")
    return removed


def sync_repo(session, repo, target_labels, dry_run=False):
    """Sync labels for a single repository. Returns counts of actions taken."""
    print(f"\n{'=' * 60}")
    print(f"Repository: {repo}")
    print(f"{'=' * 60}")

    existing = get_existing_labels(session, repo)
    existing_map = {label["name"].lower(): label for label in existing}

    created = 0
    updated = 0
    unchanged = 0

    for target in target_labels:
        name = target["name"]
        color = normalize_color(target["color"])
        description = target.get("description", "")
        name_lower = name.lower()

        if name_lower in existing_map:
            existing_label = existing_map[name_lower]
            existing_color = normalize_color(existing_label["color"])
            existing_desc = existing_label.get("description") or ""

            if existing_color != color or existing_desc != description:
                if dry_run:
                    changes = []
                    if existing_color != color:
                        changes.append(f"color: {existing_color} -> {color}")
                    if existing_desc != description:
                        changes.append("description changed")
                    print(f"  UPDATE  {name} ({', '.join(changes)})")
                else:
                    url = f"{API_BASE}/repos/{repo}/labels/{quote(existing_label['name'], safe='')}"
                    resp = session.patch(url, json={
                        "new_name": name,
                        "color": color,
                        "description": description,
                    })
                    resp.raise_for_status()
                    print(f"  UPDATED {name}")
                updated += 1
            else:
                unchanged += 1
        else:
            if dry_run:
                print(f"  CREATE  {name} (#{color})")
            else:
                url = f"{API_BASE}/repos/{repo}/labels"
                resp = session.post(url, json={
                    "name": name,
                    "color": color,
                    "description": description,
                })
                resp.raise_for_status()
                print(f"  CREATED {name}")
            created += 1

    print(f"\n  Summary: {created} created, {updated} updated, {unchanged} unchanged")
    return created, updated, unchanged


def build_target_labels(data, repo):
    """Build the full list of target labels for a repo (shared + repo-specific)."""
    labels = list(data["shared_labels"])
    repo_specific = data.get("repo_labels", {}).get(repo, [])
    labels.extend(repo_specific)
    return labels


def main():
    parser = argparse.ArgumentParser(
        description="Sync GitHub labels from a JSON definition file."
    )
    parser.add_argument(
        "--token",
        default=os.environ.get("GITHUB_TOKEN"),
        help="GitHub personal access token (or set GITHUB_TOKEN env var)",
    )
    parser.add_argument(
        "--file",
        default=DEFAULT_FILE,
        help=f"Path to labels JSON file (default: {DEFAULT_FILE})",
    )
    parser.add_argument(
        "--repos",
        nargs="+",
        help="Specific repos to sync (default: all repos in JSON file)",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would change without applying",
    )
    parser.add_argument(
        "--remove-all",
        action="store_true",
        help="Remove all existing labels from target repos before syncing",
    )
    parser.add_argument(
        "--remove-only",
        action="store_true",
        help="Remove all existing labels from target repos without syncing",
    )
    args = parser.parse_args()

    if args.remove_all and args.remove_only:
        print("Error: --remove-all and --remove-only are mutually exclusive.",
              file=sys.stderr)
        sys.exit(1)

    if not args.token:
        print("Error: GitHub token required. Use --token or set GITHUB_TOKEN env var.",
              file=sys.stderr)
        sys.exit(1)

    data = load_labels(args.file)
    repos = args.repos or data["repositories"]

    # Validate requested repos are in the JSON file
    known_repos = set(data["repositories"])
    for repo in repos:
        if repo not in known_repos:
            print(f"Warning: '{repo}' is not listed in {args.file}, "
                  f"only shared labels will be applied.", file=sys.stderr)

    session = requests.Session()
    session.headers.update({
        "Authorization": f"token {args.token}",
        "Accept": "application/vnd.github.v3+json",
        "X-GitHub-Api-Version": "2022-11-28",
    })

    if args.dry_run:
        print("DRY RUN - no changes will be made\n")

    check_rate_limit(session)

    total_created = 0
    total_updated = 0
    total_unchanged = 0

    for repo in repos:
        try:
            if args.remove_all or args.remove_only:
                print(f"\n{'=' * 60}")
                print(f"Repository: {repo}")
                print(f"{'=' * 60}")
                remove_all_labels(session, repo, dry_run=args.dry_run)
                if args.remove_only:
                    continue
            target_labels = build_target_labels(data, repo)
            c, u, uc = sync_repo(session, repo, target_labels, dry_run=args.dry_run)
            total_created += c
            total_updated += u
            total_unchanged += uc
        except requests.HTTPError as e:
            print(f"\n  ERROR syncing {repo}: {e}", file=sys.stderr)
            if e.response is not None:
                print(f"  Response: {e.response.text}", file=sys.stderr)

    check_rate_limit(session)

    print(f"\n{'=' * 60}")
    print(f"Total: {total_created} created, {total_updated} updated, "
          f"{total_unchanged} unchanged across {len(repos)} repo(s)")
    if args.dry_run:
        print("(DRY RUN - no changes were made)")


if __name__ == "__main__":
    main()