utils/get_github_job_time.py - main - githubmirror/transformers

utils/get_github_job_time.py
69 строк2 KB
Tarek Ziade
CI: use a single function for GH calls (#47474)
28 июл 2026, 17:03
28 июл 2026, 17:03d0e91c7
import argparse
import math
import time
import traceback

import dateutil.parser as date_parser
from github_utils import get_github_json


def extract_time_from_single_job(job):
    """Extract time info from a single job in a GitHub Actions workflow run"""

    job_info = {}

    start = job["started_at"]
    end = job["completed_at"]

    start_datetime = date_parser.parse(start)
    end_datetime = date_parser.parse(end)

    duration_in_min = round((end_datetime - start_datetime).total_seconds() / 60.0)

    job_info["started_at"] = start
    job_info["completed_at"] = end
    job_info["duration"] = duration_in_min

    return job_info


def get_job_time(workflow_run_id, token=None):
    """Extract time info for all jobs in a GitHub Actions workflow run"""

    url = f"https://api.github.com/repos/huggingface/transformers/actions/runs/{workflow_run_id}/jobs?per_page=50"
    result = get_github_json(url, token=token)
    job_time = {}

    try:
        job_time.update({job["name"]: extract_time_from_single_job(job) for job in result["jobs"]})
        pages_to_iterate_over = math.ceil((result["total_count"] - 50) / 50)

        for i in range(pages_to_iterate_over):
            time.sleep(1)
            result = get_github_json(url + f"&page={i + 2}", token=token)
            job_time.update({job["name"]: extract_time_from_single_job(job) for job in result["jobs"]})

        return job_time
    except Exception:
        print(f"Unknown error, could not fetch links:\n{traceback.format_exc()}")

    return {}


if __name__ == "__main__":
    r"""
    Example:

        python get_github_job_time.py --workflow_run_id 2945609517
    """

    parser = argparse.ArgumentParser()
    # Required parameters
    parser.add_argument("--workflow_run_id", type=str, required=True, help="A GitHub Actions workflow run id.")
    args = parser.parse_args()

    job_time = get_job_time(args.workflow_run_id)
    job_time = dict(sorted(job_time.items(), key=lambda item: item[1]["duration"], reverse=True))

    for k, v in job_time.items():
        print(f"{k}: {v['duration']}")