GitLab: A Python Script Displaying Latest Pipelines in a Group's Projects

display-latest-pipelines.py

#
# Display, in console, latest pipelines from each project in a given group
#
# python display-latest-pipelines.py --group-id=8784450 [--watch] [--token=$GITLAB_TOKEN] [--host=gitlab.com] [--exclude='TPs Benoit C,whatever'] [--stages-width=30]
#
import argparse
from datetime import datetime
from enum import Enum
import os
import pytz
import requests
import sys
import unicodedata

class Color(Enum):
    GREEN = "\033[92m"
    GREY = "\033[90m"
    CYAN = "\033[96m"
    RED = "\033[91m"
    YELLOW = "\033[93m"
    BLUE = "\033[94m"
    RESET = "\033[0m"
    NO_CHANGE = ""

parser = argparse.ArgumentParser(description='Retrieve GitLab pipeline data for projects in a group')
parser.add_argument('--host', type=str, default="gitlab.com", help='Hostname of the GitLab instance')
parser.add_argument('--token', type=str, default=None, help='GitLab API access token (default: $GITLAB_TOKEN (exported) environment variable)')
parser.add_argument('--group-id', type=int, help='ID of the group to retrieve projects from')
parser.add_argument('--exclude', type=str, default="", help='Comma-separated list of project names to exclude (default: none)')
parser.add_argument('--watch', action='store_true', help='Run indefinitely while refreshing output')
parser.add_argument('--stages-width', type=int, default=42, help='Width for stages display (default: 42)')
args = parser.parse_args()

if args.token is None:
    args.token = os.getenv('GITLAB_TOKEN', 'NONE')

headers = {"Private-Token": args.token}
projects_url = f"https://{args.host}/api/v4/groups/{args.group_id}/projects?include_subgroups=true&simple=true"

def print_or_gather(output, text):
    if args.watch:
        output.append(text)
    else:
        print(text)

def count_emoji(text):
    """Count the number of emojis in the input text."""
    custom_lengths = {
        "\U0001F3D7": 0,  # Construction sign 🏗️
        # Add more special characters as needed.
    }
    count = 0
    for char in text:
        if unicodedata.category(char).startswith('So'):
            if char in custom_lengths:
                count += custom_lengths[char]
            else:
                count += 1
    return count

def fetch_pipelines():
    response = requests.get(projects_url, headers=headers)
    if response.status_code != 200:
        print(f"\n{Color.RED.value}Failed to call GitLab instance: {response.json()}{Color.RESET.value}")
        return
    projects = response.json()

    pipeline_data = {}
    no_pipelines_projects = []
    excluded_projects = set(args.exclude.split(','))

    output = []

    for project in projects:
        if project["name"] in excluded_projects:
            continue
        pipeline_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines?per_page=1&sort=desc&order_by=id"
        response = requests.get(pipeline_url, headers=headers)
        if not response.json():
            no_pipelines_projects.append(project['name'])
            continue
        pipeline = response.json()[0]

        updated_time = datetime.strptime(pipeline["updated_at"], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=pytz.utc).astimezone(pytz.timezone('Europe/Paris'))

        updated_at_human_readable = updated_time.strftime("%d %b %Y at %H:%M:%S")
        time_diff = datetime.now(pytz.utc) - updated_time
        delta = time_diff.total_seconds()
        if delta < 120:
            updated_ago = f'{int(delta)} seconds'
        elif delta < 7200: # 2 hours in seconds
            updated_ago = f'{int(delta / 60)} minutes'
        elif delta < 172800: # 2 days in seconds
            updated_ago = f'{int(delta / 3600)} hours'
        else:
            updated_ago = f'{int(delta / 86400)} days'
        match pipeline["status"]:
            case "success":
                color = Color.GREEN
            case "created" | "waiting_for_resource" | "preparing" | "pending" | "canceled" | "skipped" | "manual":
                color = Color.GREY
            case "running":
                color = Color.BLUE
            case "failed":
                color = Color.RED
        print_or_gather(output,f"\n↓ {color.value}{project['name']} for {pipeline['ref']} : {pipeline['status']} (since {updated_at_human_readable}, {updated_ago} ago){Color.RESET.value}")
        job_data = {}
        jobs_url = f"https://gitlab.com/api/v4/projects/{project['id']}/pipelines/{pipeline['id']}/jobs"
        response = requests.get(jobs_url, headers=headers)
        jobs = response.json()

        for job in list(reversed(jobs)):
            job_name = job["name"]
            stage = job["stage"]
            job_status = job["status"]

            match (job_status, pipeline["status"]):
                case ("success", _):
                    emoji = "🟢"
                    job_color = Color.GREEN
                case ("running", _):
                    emoji = "🔵"
                    job_color = Color.BLUE
                case ("pending" | "created", _):
                    emoji = "🔘"
                    job_color = Color.NO_CHANGE
                case ("skipped" | "canceled", _):
                    emoji = "🔘"
                    job_color = Color.GREY
                case ("warning", _):
                    emoji = "🟠"
                    job_color = Color.YELLOW
                case ("manual", _):
                    emoji = "▶️"
                    job_color = Color.NO_CHANGE
                case ("failed", "success"):
                    emoji = "🟠"
                    job_color = Color.YELLOW
                case ("failed", _):
                    emoji = "🔴"
                    job_color = Color.RED
                case (_, _):
                    print(job_status)
            if stage not in job_data:
                job_data[stage] = []
            job_data[stage].append((job_name, job_status, job_color, emoji))

        # Sort jobs within each stage alphabetically by job name
        for stage in job_data:
            job_data[stage].sort(key=lambda x: x[0])

        # Find the maximum number of jobs in any stage for this pipeline
        max_jobs = max(len(jobs) for jobs in job_data.values())

        lines = [" "] * (max_jobs + 1)
        lines[0] = "" # stages start with a border character instead of a space
        # Print out the job data for each stage, padding to make all stages content the same length
        for stage, jobs in job_data.items():
            stage = "[ "+stage+" ]"
            lines[0] = f"{lines[0]}╔{stage.center(args.stages_width - 3 - count_emoji(stage), '═').upper()}╗ "
            for i, (job_name, job_status, job_color, emoji) in enumerate(jobs, start=1):
                # emojis in job names make this exercise a bit more difficult: ljust make them expand the size, so we compensate
                lines[i] = f"{lines[i]}{emoji} {job_color.value}{job_name[:args.stages_width - 6 - count_emoji(job_name)].ljust(args.stages_width - 3 - count_emoji(job_name))}{Color.RESET.value}"
            for j in range(len(jobs) + 1, max_jobs + 1):
                lines[j] = lines[j] + " ".ljust(args.stages_width)

        for line in lines:
            print_or_gather(output,line)

    if no_pipelines_projects:
        print_or_gather(output,f"\n\033[90mProjects without pipeline: {', '.join(no_pipelines_projects)}\033[0m")

    return "\n".join(output)

try:
    if args.watch:
            while True:
                output = fetch_pipelines()
                sys.stdout.write("\x1b[2J\x1b[H")  # Clear the screen
                sys.stdout.flush()
                print(output)
    else:
        fetch_pipelines()
except KeyboardInterrupt:
    pass