gitlab-project-activity.py

#!/usr/bin/env python3
"""Export GitLab group projects by latest activity, including repository size.

Usage:
  GITLAB_TOKEN=... uv run python gitlab-project-activity.py https://gitlab.example.com parent/group
  GITLAB_TOKEN=... uv run python gitlab-project-activity.py https://gitlab.example.com parent/group --output activity.csv

Sample output (final screen):
  Scanning your-gitlab-group; writing to projects-20261002-174233.csv
  Status:         stopped
  Projects Found: 22
  Last 7 Days:    0
  Last 30 Days:   12
  Last 90 Days:   10
  Last 6 months:  0
  Last 12 months: 0
  Older:          0
  Done: 22 projects in projects-20261002-174233.csv

Creates a new CSV and appends each page as it arrives. Existing files are never
overwritten; an interrupted run leaves a partial CSV. Includes nested subgroups
but not projects shared into the group. GitLab's lastActivityAt includes creation
and other activity, not just commits and merge requests. Repository size may be
unavailable without Reporter access.
"""

import argparse
import csv
import os
import sys
from calendar import monthrange
from datetime import datetime, timedelta, timezone

import gitlab

FIELDS = ("project_id", "path", "last_activity_at", "repository_size_bytes", "web_url")
QUERY = """
query($group: ID!, $after: String, $first: Int!) {
  group(fullPath: $group) {
    projects(first: $first, after: $after, includeSubgroups: true, sort: ACTIVITY_DESC) {
      nodes {
        id
        fullPath
        webUrl
        lastActivityAt
        statistics { repositorySize }
      }
      pageInfo { hasNextPage endCursor }
    }
  }
}
"""


def months_ago(now, months):
    year, month = divmod(now.year * 12 + now.month - months - 1, 12)
    return now.replace(
        year=year, month=month + 1, day=min(now.day, monthrange(year, month + 1)[1])
    )


def activity_bucket(timestamp, cutoffs):
    if not timestamp:
        return len(cutoffs)  # No recorded activity is least recent.
    activity = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
    return next(
        (index for index, cutoff in enumerate(cutoffs) if activity >= cutoff),
        len(cutoffs),
    )


def show_progress(total, counts, status="running", redraw=False):
    tty = sys.stdout.isatty()
    if redraw and tty:
        sys.stdout.write("\x1b[8A")
    lines = (
        ("Status:", status),
        ("Projects Found:", total),
        ("Last 7 Days:", counts[0]),
        ("Last 30 Days:", counts[1]),
        ("Last 90 Days:", counts[2]),
        ("Last 6 months:", counts[3]),
        ("Last 12 months:", counts[4]),
        ("Older:", counts[5]),
    )
    for label, value in lines:
        print(("\x1b[2K" if tty else "") + f"{label:<15} {value}")
    sys.stdout.flush()


def project_pages(gl, group_path):
    after = None
    page_size = 100
    while True:
        response = gl.http_post(
            f"{gl.url}/api/graphql",
            post_data={
                "query": QUERY,
                "variables": {
                    "group": group_path,
                    "after": after,
                    "first": page_size,
                },
            },
        )
        if response.get("errors"):
            if page_size > 1 and any(
                "complexity" in error.get("message", "").lower()
                for error in response["errors"]
            ):
                page_size //= 2
                continue
            raise RuntimeError(f"GitLab GraphQL errors: {response['errors']}")
        if response["data"]["group"] is None:
            raise ValueError(f"group not found or not visible: {group_path}")
        projects = response["data"]["group"]["projects"]
        page = projects["pageInfo"]
        if page["hasNextPage"] and (
            not page["endCursor"] or page["endCursor"] == after
        ):
            raise RuntimeError("GitLab returned a stalled project cursor")
        # ponytail: streaming trusts GitLab's ACTIVITY_DESC order; buffer and sort if
        # exact ordering under concurrent project activity becomes necessary.
        yield [
            {
                "project_id": int(project["id"].rsplit("/", 1)[-1]),
                "path": project["fullPath"],
                "last_activity_at": project["lastActivityAt"],
                "repository_size_bytes": (project["statistics"] or {}).get(
                    "repositorySize"
                ),
                "web_url": project["webUrl"],
            }
            for project in projects["nodes"]
            if project["fullPath"].startswith(group_path + "/")
        ]
        if not page["hasNextPage"]:
            break
        after = page["endCursor"]


def main():
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("url", help="GitLab instance root URL (not /api/v4)")
    parser.add_argument("group", help="Top-level group ID or full path")
    parser.add_argument(
        "--output", help="New CSV filename (default: timestamped projects-*.csv)"
    )
    args = parser.parse_args()
    token = os.environ.get("GITLAB_TOKEN")
    if not token:
        parser.error("set GITLAB_TOKEN to an API-scoped personal access token")

    gl = gitlab.Gitlab(args.url, private_token=token, timeout=30, keep_base_url=True)
    group_path = gl.groups.get(args.group).full_path
    now = datetime.now(timezone.utc)
    cutoffs = (
        now - timedelta(days=7),
        now - timedelta(days=30),
        now - timedelta(days=90),
        months_ago(now, 6),
        months_ago(now, 12),
    )
    counts = [0] * 6
    output = args.output or f"projects-{now:%Y%m%d-%H%M%S}.csv"
    with open(output, "x", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(file, fieldnames=FIELDS)
        writer.writeheader()
        file.flush()
        print(f"Scanning {group_path}; writing to {output}", flush=True)
        total = 0
        tty = sys.stdout.isatty()
        if tty:
            show_progress(total, counts)
        try:
            for rows in project_pages(gl, group_path):
                writer.writerows(rows)
                file.flush()
                total += len(rows)
                for row in rows:
                    counts[activity_bucket(row["last_activity_at"], cutoffs)] += 1
                if tty:
                    show_progress(total, counts, redraw=True)
        finally:
            show_progress(total, counts, status="stopped", redraw=tty)
    print(f"Done: {total} projects in {output}")


if __name__ == "__main__":
    main()